Commit 0f999784 authored by antirez's avatar antirez
Browse files

Merge remote-tracking branch 'origin/unstable' into unstable

parents 74f10793 1b247d13
...@@ -50,17 +50,23 @@ void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) { ...@@ -50,17 +50,23 @@ void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
} }
} }
void replicationFeedMonitors(list *monitors, int dictid, robj **argv, int argc) { void replicationFeedMonitors(redisClient *c, list *monitors, int dictid, robj **argv, int argc) {
listNode *ln; listNode *ln;
listIter li; listIter li;
int j; int j, port;
sds cmdrepr = sdsnew("+"); sds cmdrepr = sdsnew("+");
robj *cmdobj; robj *cmdobj;
char ip[32];
struct timeval tv; struct timeval tv;
gettimeofday(&tv,NULL); gettimeofday(&tv,NULL);
cmdrepr = sdscatprintf(cmdrepr,"%ld.%06ld ",(long)tv.tv_sec,(long)tv.tv_usec); cmdrepr = sdscatprintf(cmdrepr,"%ld.%06ld ",(long)tv.tv_sec,(long)tv.tv_usec);
if (dictid != 0) cmdrepr = sdscatprintf(cmdrepr,"(db %d) ", dictid); if (c->flags & REDIS_LUA_CLIENT) {
cmdrepr = sdscatprintf(cmdrepr,"[%d lua] ", dictid);
} else {
anetPeerToString(c->fd,ip,&port);
cmdrepr = sdscatprintf(cmdrepr,"[%d %s:%d] ", dictid,ip,port);
}
for (j = 0; j < argc; j++) { for (j = 0; j < argc; j++) {
if (argv[j]->encoding == REDIS_ENCODING_INT) { if (argv[j]->encoding == REDIS_ENCODING_INT) {
...@@ -596,7 +602,7 @@ void replicationCron(void) { ...@@ -596,7 +602,7 @@ void replicationCron(void) {
if (slave->replstate == REDIS_REPL_SEND_BULK) continue; if (slave->replstate == REDIS_REPL_SEND_BULK) continue;
if (slave->replstate == REDIS_REPL_ONLINE) { if (slave->replstate == REDIS_REPL_ONLINE) {
/* If the slave is online send a normal ping */ /* If the slave is online send a normal ping */
addReplySds(slave,sdsnew("PING\r\n")); addReplySds(slave,sdsnew("*1\r\n$4\r\nPING\r\n"));
} else { } else {
/* Otherwise we are in the pre-synchronization stage. /* Otherwise we are in the pre-synchronization stage.
* Just a newline will do the work of refreshing the * Just a newline will do the work of refreshing the
......
...@@ -206,15 +206,45 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) { ...@@ -206,15 +206,45 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) {
goto cleanup; goto cleanup;
} }
/* There are commands that are not allowed inside scripts. */
if (cmd->flags & REDIS_CMD_NOSCRIPT) { if (cmd->flags & REDIS_CMD_NOSCRIPT) {
luaPushError(lua, "This Redis command is not allowed from scripts"); luaPushError(lua, "This Redis command is not allowed from scripts");
goto cleanup; goto cleanup;
} }
if (cmd->flags & REDIS_CMD_WRITE && server.lua_random_dirty) { /* Write commands are forbidden against read-only slaves, or if a
* command marked as non-deterministic was already called in the context
* of this script. */
if (cmd->flags & REDIS_CMD_WRITE) {
if (server.lua_random_dirty) {
luaPushError(lua, luaPushError(lua,
"Write commands not allowed after non deterministic commands"); "Write commands not allowed after non deterministic commands");
goto cleanup; goto cleanup;
} else if (server.masterhost && server.repl_slave_ro &&
!(server.lua_caller->flags & REDIS_MASTER))
{
luaPushError(lua, shared.roslaveerr->ptr);
goto cleanup;
} else if (server.stop_writes_on_bgsave_err &&
server.saveparamslen > 0 &&
server.lastbgsave_status == REDIS_ERR)
{
luaPushError(lua, shared.bgsaveerr->ptr);
goto cleanup;
}
}
/* If we reached the memory limit configured via maxmemory, commands that
* could enlarge the memory usage are not allowed, but only if this is the
* first write in the context of this script, otherwise we can't stop
* in the middle. */
if (server.maxmemory && server.lua_write_dirty == 0 &&
(cmd->flags & REDIS_CMD_DENYOOM))
{
if (freeMemoryIfNeeded() == REDIS_ERR) {
luaPushError(lua, shared.oomerr->ptr);
goto cleanup;
}
} }
if (cmd->flags & REDIS_CMD_RANDOM) server.lua_random_dirty = 1; if (cmd->flags & REDIS_CMD_RANDOM) server.lua_random_dirty = 1;
......
...@@ -111,6 +111,24 @@ sds sdsMakeRoomFor(sds s, size_t addlen) { ...@@ -111,6 +111,24 @@ sds sdsMakeRoomFor(sds s, size_t addlen) {
return newsh->buf; return newsh->buf;
} }
/* Reallocate the sds string so that it has no free space at the end. The
* contained string remains not altered, but next concatenation operations
* will require a reallocation. */
sds sdsRemoveFreeSpace(sds s) {
struct sdshdr *sh;
sh = (void*) (s-(sizeof(struct sdshdr)));
sh = zrealloc(sh, sizeof(struct sdshdr)+sh->len+1);
sh->free = 0;
return sh->buf;
}
size_t sdsAllocSize(sds s) {
struct sdshdr *sh = (void*) (s-(sizeof(struct sdshdr)));
return sizeof(*sh)+sh->len+sh->free+1;
}
/* Increment the sds length and decrements the left free space at the /* Increment the sds length and decrements the left free space at the
* end of the string accordingly to 'incr'. Also set the null term * end of the string accordingly to 'incr'. Also set the null term
* in the new end of the string. * in the new end of the string.
......
...@@ -94,5 +94,7 @@ sds sdsmapchars(sds s, char *from, char *to, size_t setlen); ...@@ -94,5 +94,7 @@ sds sdsmapchars(sds s, char *from, char *to, size_t setlen);
/* Low level functions exposed to the user API */ /* Low level functions exposed to the user API */
sds sdsMakeRoomFor(sds s, size_t addlen); sds sdsMakeRoomFor(sds s, size_t addlen);
void sdsIncrLen(sds s, int incr); void sdsIncrLen(sds s, int incr);
sds sdsRemoveFreeSpace(sds s);
size_t sdsAllocSize(sds s);
#endif #endif
This diff is collapsed.
...@@ -259,7 +259,7 @@ void listTypeConvert(robj *subject, int enc) { ...@@ -259,7 +259,7 @@ void listTypeConvert(robj *subject, int enc) {
*----------------------------------------------------------------------------*/ *----------------------------------------------------------------------------*/
void pushGenericCommand(redisClient *c, int where) { void pushGenericCommand(redisClient *c, int where) {
int j, addlen = 0, pushed = 0; int j, waiting = 0, pushed = 0;
robj *lobj = lookupKeyWrite(c->db,c->argv[1]); robj *lobj = lookupKeyWrite(c->db,c->argv[1]);
int may_have_waiting_clients = (lobj == NULL); int may_have_waiting_clients = (lobj == NULL);
...@@ -272,7 +272,7 @@ void pushGenericCommand(redisClient *c, int where) { ...@@ -272,7 +272,7 @@ void pushGenericCommand(redisClient *c, int where) {
c->argv[j] = tryObjectEncoding(c->argv[j]); c->argv[j] = tryObjectEncoding(c->argv[j]);
if (may_have_waiting_clients) { if (may_have_waiting_clients) {
if (handleClientsWaitingListPush(c,c->argv[1],c->argv[j])) { if (handleClientsWaitingListPush(c,c->argv[1],c->argv[j])) {
addlen++; waiting++;
continue; continue;
} else { } else {
may_have_waiting_clients = 0; may_have_waiting_clients = 0;
...@@ -285,9 +285,21 @@ void pushGenericCommand(redisClient *c, int where) { ...@@ -285,9 +285,21 @@ void pushGenericCommand(redisClient *c, int where) {
listTypePush(lobj,c->argv[j],where); listTypePush(lobj,c->argv[j],where);
pushed++; pushed++;
} }
addReplyLongLong(c,addlen + (lobj ? listTypeLength(lobj) : 0)); addReplyLongLong(c, waiting + (lobj ? listTypeLength(lobj) : 0));
if (pushed) signalModifiedKey(c->db,c->argv[1]); if (pushed) signalModifiedKey(c->db,c->argv[1]);
server.dirty += pushed; server.dirty += pushed;
/* Alter the replication of the command accordingly to the number of
* list elements delivered to clients waiting into a blocking operation.
* We do that only if there were waiting clients, and only if still some
* element was pushed into the list (othewise dirty is 0 and nothign will
* be propagated). */
if (waiting && pushed) {
/* CMD KEY a b C D E */
for (j = 0; j < waiting; j++) decrRefCount(c->argv[j+2]);
memmove(c->argv+2,c->argv+2+waiting,sizeof(robj*)*pushed);
c->argc -= waiting;
}
} }
void lpushCommand(redisClient *c) { void lpushCommand(redisClient *c) {
...@@ -655,8 +667,6 @@ void lremCommand(redisClient *c) { ...@@ -655,8 +667,6 @@ void lremCommand(redisClient *c) {
*/ */
void rpoplpushHandlePush(redisClient *origclient, redisClient *c, robj *dstkey, robj *dstobj, robj *value) { void rpoplpushHandlePush(redisClient *origclient, redisClient *c, robj *dstkey, robj *dstobj, robj *value) {
robj *aux;
if (!handleClientsWaitingListPush(origclient,dstkey,value)) { if (!handleClientsWaitingListPush(origclient,dstkey,value)) {
/* Create the list if the key does not exist */ /* Create the list if the key does not exist */
if (!dstobj) { if (!dstobj) {
...@@ -666,27 +676,19 @@ void rpoplpushHandlePush(redisClient *origclient, redisClient *c, robj *dstkey, ...@@ -666,27 +676,19 @@ void rpoplpushHandlePush(redisClient *origclient, redisClient *c, robj *dstkey,
signalModifiedKey(c->db,dstkey); signalModifiedKey(c->db,dstkey);
} }
listTypePush(dstobj,value,REDIS_HEAD); listTypePush(dstobj,value,REDIS_HEAD);
/* If we are pushing as a result of LPUSH against a key /* Additionally propagate this PUSH operation together with
* watched by BRPOPLPUSH, we need to rewrite the command vector * the operation performed by the command. */
* as an LPUSH. {
* robj **argv = zmalloc(sizeof(robj*)*3);
* If this is called directly by RPOPLPUSH (either directly argv[0] = createStringObject("LPUSH",5);
* or via a BRPOPLPUSH where the popped list exists) argv[1] = dstkey;
* we should replicate the RPOPLPUSH command itself. */ argv[2] = value;
if (c != origclient) { incrRefCount(argv[1]);
aux = createStringObject("LPUSH",5); incrRefCount(argv[2]);
rewriteClientCommandVector(origclient,3,aux,dstkey,value); alsoPropagate(server.lpushCommand,c->db->id,argv,3,
decrRefCount(aux); REDIS_PROPAGATE_AOF|REDIS_PROPAGATE_REPL);
} else {
/* Make sure to always use RPOPLPUSH in the replication / AOF,
* even if the original command was BRPOPLPUSH. */
aux = createStringObject("RPOPLPUSH",9);
rewriteClientCommandVector(origclient,3,aux,c->argv[1],c->argv[2]);
decrRefCount(aux);
} }
server.dirty++;
} }
/* Always send the pushed value to the client. */ /* Always send the pushed value to the client. */
addReplyBulk(c,value); addReplyBulk(c,value);
} }
...@@ -717,6 +719,13 @@ void rpoplpushCommand(redisClient *c) { ...@@ -717,6 +719,13 @@ void rpoplpushCommand(redisClient *c) {
signalModifiedKey(c->db,touchedkey); signalModifiedKey(c->db,touchedkey);
decrRefCount(touchedkey); decrRefCount(touchedkey);
server.dirty++; server.dirty++;
/* Replicate this as a simple RPOP since the LPUSH side is replicated
* by rpoplpushHandlePush() call if needed (it may not be needed
* if a client is blocking wait a push against the list). */
rewriteClientCommandVector(c,2,
resetRefCount(createStringObject("RPOP",4)),
c->argv[1]);
} }
} }
...@@ -924,36 +933,22 @@ void blockingPopGenericCommand(redisClient *c, int where) { ...@@ -924,36 +933,22 @@ void blockingPopGenericCommand(redisClient *c, int where) {
return; return;
} else { } else {
if (listTypeLength(o) != 0) { if (listTypeLength(o) != 0) {
/* If the list contains elements fall back to the usual /* Non empty list, this is like a non normal [LR]POP. */
* non-blocking POP operation */ robj *value = listTypePop(o,where);
struct redisCommand *orig_cmd; redisAssert(value != NULL);
robj *argv[2], **orig_argv;
int orig_argc;
/* We need to alter the command arguments before to call
* popGenericCommand() as the command takes a single key. */
orig_argv = c->argv;
orig_argc = c->argc;
orig_cmd = c->cmd;
argv[1] = c->argv[j];
c->argv = argv;
c->argc = 2;
/* Also the return value is different, we need to output
* the multi bulk reply header and the key name. The
* "real" command will add the last element (the value)
* for us. If this souds like an hack to you it's just
* because it is... */
addReplyMultiBulkLen(c,2);
addReplyBulk(c,argv[1]);
popGenericCommand(c,where);
/* Fix the client structure with the original stuff */ addReplyMultiBulkLen(c,2);
c->argv = orig_argv; addReplyBulk(c,c->argv[j]);
c->argc = orig_argc; addReplyBulk(c,value);
c->cmd = orig_cmd; decrRefCount(value);
if (listTypeLength(o) == 0) dbDelete(c->db,c->argv[j]);
signalModifiedKey(c->db,c->argv[j]);
server.dirty++;
/* Replicate it as an [LR]POP instead of B[LR]POP. */
rewriteClientCommandVector(c,2,
(where == REDIS_HEAD) ? shared.lpop : shared.rpop,
c->argv[j]);
return; return;
} }
} }
......
...@@ -5,6 +5,8 @@ ...@@ -5,6 +5,8 @@
#include <ctype.h> #include <ctype.h>
#include <limits.h> #include <limits.h>
#include <math.h> #include <math.h>
#include <unistd.h>
#include <sys/time.h>
#include "util.h" #include "util.h"
...@@ -209,8 +211,8 @@ int ll2string(char *s, size_t len, long long value) { ...@@ -209,8 +211,8 @@ int ll2string(char *s, size_t len, long long value) {
/* Convert a string into a long long. Returns 1 if the string could be parsed /* Convert a string into a long long. Returns 1 if the string could be parsed
* into a (non-overflowing) long long, 0 otherwise. The value will be set to * into a (non-overflowing) long long, 0 otherwise. The value will be set to
* the parsed value when appropriate. */ * the parsed value when appropriate. */
int string2ll(char *s, size_t slen, long long *value) { int string2ll(const char *s, size_t slen, long long *value) {
char *p = s; const char *p = s;
size_t plen = 0; size_t plen = 0;
int negative = 0; int negative = 0;
unsigned long long v; unsigned long long v;
...@@ -275,7 +277,7 @@ int string2ll(char *s, size_t slen, long long *value) { ...@@ -275,7 +277,7 @@ int string2ll(char *s, size_t slen, long long *value) {
/* Convert a string into a long. Returns 1 if the string could be parsed into a /* Convert a string into a long. Returns 1 if the string could be parsed into a
* (non-overflowing) long, 0 otherwise. The value will be set to the parsed * (non-overflowing) long, 0 otherwise. The value will be set to the parsed
* value when appropriate. */ * value when appropriate. */
int string2l(char *s, size_t slen, long *lval) { int string2l(const char *s, size_t slen, long *lval) {
long long llval; long long llval;
if (!string2ll(s,slen,&llval)) if (!string2ll(s,slen,&llval))
...@@ -327,6 +329,52 @@ int d2string(char *buf, size_t len, double value) { ...@@ -327,6 +329,52 @@ int d2string(char *buf, size_t len, double value) {
return len; return len;
} }
/* Generate the Redis "Run ID", a SHA1-sized random number that identifies a
* given execution of Redis, so that if you are talking with an instance
* having run_id == A, and you reconnect and it has run_id == B, you can be
* sure that it is either a different instance or it was restarted. */
void getRandomHexChars(char *p, unsigned int len) {
FILE *fp = fopen("/dev/urandom","r");
char *charset = "0123456789abcdef";
unsigned int j;
if (fp == NULL || fread(p,len,1,fp) == 0) {
/* If we can't read from /dev/urandom, do some reasonable effort
* in order to create some entropy, since this function is used to
* generate run_id and cluster instance IDs */
char *x = p;
unsigned int l = len;
struct timeval tv;
pid_t pid = getpid();
/* Use time and PID to fill the initial array. */
gettimeofday(&tv,NULL);
if (l >= sizeof(tv.tv_usec)) {
memcpy(x,&tv.tv_usec,sizeof(tv.tv_usec));
l -= sizeof(tv.tv_usec);
x += sizeof(tv.tv_usec);
}
if (l >= sizeof(tv.tv_sec)) {
memcpy(x,&tv.tv_sec,sizeof(tv.tv_sec));
l -= sizeof(tv.tv_sec);
x += sizeof(tv.tv_sec);
}
if (l >= sizeof(pid)) {
memcpy(x,&pid,sizeof(pid));
l -= sizeof(pid);
x += sizeof(pid);
}
/* Finally xor it with rand() output, that was already seeded with
* time() at startup. */
for (j = 0; j < len; j++)
p[j] ^= rand();
}
/* Turn it into hex digits taking just 4 bits out of 8 for every byte. */
for (j = 0; j < len; j++)
p[j] = charset[p[j] & 0x0F];
fclose(fp);
}
#ifdef UTIL_TEST_MAIN #ifdef UTIL_TEST_MAIN
#include <assert.h> #include <assert.h>
......
...@@ -5,8 +5,8 @@ int stringmatchlen(const char *p, int plen, const char *s, int slen, int nocase) ...@@ -5,8 +5,8 @@ int stringmatchlen(const char *p, int plen, const char *s, int slen, int nocase)
int stringmatch(const char *p, const char *s, int nocase); int stringmatch(const char *p, const char *s, int nocase);
long long memtoll(const char *p, int *err); long long memtoll(const char *p, int *err);
int ll2string(char *s, size_t len, long long value); int ll2string(char *s, size_t len, long long value);
int string2ll(char *s, size_t slen, long long *value); int string2ll(const char *s, size_t slen, long long *value);
int string2l(char *s, size_t slen, long *value); int string2l(const char *s, size_t slen, long *value);
int d2string(char *buf, size_t len, double value); int d2string(char *buf, size_t len, double value);
#endif #endif
#define REDIS_VERSION "2.9.4" #define REDIS_VERSION "2.9.5"
...@@ -75,6 +75,8 @@ ...@@ -75,6 +75,8 @@
#define ZIP_BIGLEN 254 #define ZIP_BIGLEN 254
/* Different encoding/length possibilities */ /* Different encoding/length possibilities */
#define ZIP_STR_MASK (0xc0)
#define ZIP_INT_MASK (0x30)
#define ZIP_STR_06B (0 << 6) #define ZIP_STR_06B (0 << 6)
#define ZIP_STR_14B (1 << 6) #define ZIP_STR_14B (1 << 6)
#define ZIP_STR_32B (2 << 6) #define ZIP_STR_32B (2 << 6)
...@@ -82,9 +84,8 @@ ...@@ -82,9 +84,8 @@
#define ZIP_INT_32B (0xc0 | 1<<4) #define ZIP_INT_32B (0xc0 | 1<<4)
#define ZIP_INT_64B (0xc0 | 2<<4) #define ZIP_INT_64B (0xc0 | 2<<4)
/* Macro's to determine type */ /* Macro to determine type */
#define ZIP_IS_STR(enc) (((enc) & 0xc0) < 0xc0) #define ZIP_IS_STR(enc) (((enc) & ZIP_STR_MASK) < ZIP_STR_MASK)
#define ZIP_IS_INT(enc) (!ZIP_IS_STR(enc) && ((enc) & 0x30) < 0x30)
/* Utility macros */ /* Utility macros */
#define ZIPLIST_BYTES(zl) (*((uint32_t*)(zl))) #define ZIPLIST_BYTES(zl) (*((uint32_t*)(zl)))
...@@ -110,19 +111,13 @@ typedef struct zlentry { ...@@ -110,19 +111,13 @@ typedef struct zlentry {
unsigned char *p; unsigned char *p;
} zlentry; } zlentry;
/* Return the encoding pointer to by 'p'. */ #define ZIP_ENTRY_ENCODING(ptr, encoding) do { \
static unsigned int zipEntryEncoding(unsigned char *p) { (encoding) = (ptr[0]) & (ZIP_STR_MASK | ZIP_INT_MASK); \
/* String encoding: 2 MSBs */ if (((encoding) & ZIP_STR_MASK) < ZIP_STR_MASK) { \
unsigned char b = p[0] & 0xc0; /* String encoding: 2 MSBs */ \
if (b < 0xc0) { (encoding) &= ZIP_STR_MASK; \
return b; } \
} else { } while(0)
/* Integer encoding: 4 MSBs */
return p[0] & 0xf0;
}
assert(NULL);
return 0;
}
/* Return bytes needed to store integer encoded by 'encoding' */ /* Return bytes needed to store integer encoded by 'encoding' */
static unsigned int zipIntSize(unsigned char encoding) { static unsigned int zipIntSize(unsigned char encoding) {
...@@ -135,36 +130,6 @@ static unsigned int zipIntSize(unsigned char encoding) { ...@@ -135,36 +130,6 @@ static unsigned int zipIntSize(unsigned char encoding) {
return 0; return 0;
} }
/* Decode the encoded length pointed by 'p'. If a pointer to 'lensize' is
* provided, it is set to the number of bytes required to encode the length. */
static unsigned int zipDecodeLength(unsigned char *p, unsigned int *lensize) {
unsigned char encoding = zipEntryEncoding(p);
unsigned int len = 0;
if (ZIP_IS_STR(encoding)) {
switch(encoding) {
case ZIP_STR_06B:
len = p[0] & 0x3f;
if (lensize) *lensize = 1;
break;
case ZIP_STR_14B:
len = ((p[0] & 0x3f) << 8) | p[1];
if (lensize) *lensize = 2;
break;
case ZIP_STR_32B:
len = (p[1] << 24) | (p[2] << 16) | (p[3] << 8) | p[4];
if (lensize) *lensize = 5;
break;
default:
assert(NULL);
}
} else {
len = zipIntSize(encoding);
if (lensize) *lensize = 1;
}
return len;
}
/* Encode the length 'l' writing it in 'p'. If p is NULL it just returns /* Encode the length 'l' writing it in 'p'. If p is NULL it just returns
* the amount of bytes required to encode such a length. */ * the amount of bytes required to encode such a length. */
static unsigned int zipEncodeLength(unsigned char *p, unsigned char encoding, unsigned int rawlen) { static unsigned int zipEncodeLength(unsigned char *p, unsigned char encoding, unsigned int rawlen) {
...@@ -201,18 +166,33 @@ static unsigned int zipEncodeLength(unsigned char *p, unsigned char encoding, un ...@@ -201,18 +166,33 @@ static unsigned int zipEncodeLength(unsigned char *p, unsigned char encoding, un
return len; return len;
} }
/* Decode the length of the previous element stored at "p". */ /* Decode the length encoded in 'ptr'. The 'encoding' variable will hold the
static unsigned int zipPrevDecodeLength(unsigned char *p, unsigned int *lensize) { * entries encoding, the 'lensize' variable will hold the number of bytes
unsigned int len = *p; * required to encode the entries length, and the 'len' variable will hold the
if (len < ZIP_BIGLEN) { * entries length. */
if (lensize) *lensize = 1; #define ZIP_DECODE_LENGTH(ptr, encoding, lensize, len) do { \
} else { ZIP_ENTRY_ENCODING((ptr), (encoding)); \
if (lensize) *lensize = 1+sizeof(len); if ((encoding) < ZIP_STR_MASK) { \
memcpy(&len,p+1,sizeof(len)); if ((encoding) == ZIP_STR_06B) { \
memrev32ifbe(&len); (lensize) = 1; \
} (len) = (ptr)[0] & 0x3f; \
return len; } else if ((encoding) == ZIP_STR_14B) { \
} (lensize) = 2; \
(len) = (((ptr)[0] & 0x3f) << 8) | (ptr)[1]; \
} else if (encoding == ZIP_STR_32B) { \
(lensize) = 5; \
(len) = ((ptr)[1] << 24) | \
((ptr)[2] << 16) | \
((ptr)[3] << 8) | \
((ptr)[4]); \
} else { \
assert(NULL); \
} \
} else { \
(lensize) = 1; \
(len) = zipIntSize(encoding); \
} \
} while(0);
/* Encode the length of the previous entry and write it to "p". Return the /* Encode the length of the previous entry and write it to "p". Return the
* number of bytes needed to encode this length if "p" is NULL. */ * number of bytes needed to encode this length if "p" is NULL. */
...@@ -241,12 +221,43 @@ static void zipPrevEncodeLengthForceLarge(unsigned char *p, unsigned int len) { ...@@ -241,12 +221,43 @@ static void zipPrevEncodeLengthForceLarge(unsigned char *p, unsigned int len) {
memrev32ifbe(p+1); memrev32ifbe(p+1);
} }
/* Return the difference in number of bytes needed to store the new length /* Decode the number of bytes required to store the length of the previous
* "len" on the entry pointed to by "p". */ * element, from the perspective of the entry pointed to by 'ptr'. */
#define ZIP_DECODE_PREVLENSIZE(ptr, prevlensize) do { \
if ((ptr)[0] < ZIP_BIGLEN) { \
(prevlensize) = 1; \
} else { \
(prevlensize) = 5; \
} \
} while(0);
/* Decode the length of the previous element, from the perspective of the entry
* pointed to by 'ptr'. */
#define ZIP_DECODE_PREVLEN(ptr, prevlensize, prevlen) do { \
ZIP_DECODE_PREVLENSIZE(ptr, prevlensize); \
if ((prevlensize) == 1) { \
(prevlen) = (ptr)[0]; \
} else if ((prevlensize) == 5) { \
assert(sizeof((prevlensize)) == 4); \
memcpy(&(prevlen), ((char*)(ptr)) + 1, 4); \
memrev32ifbe(&prevlen); \
} \
} while(0);
/* Return the difference in number of bytes needed to store the length of the
* previous element 'len', in the entry pointed to by 'p'. */
static int zipPrevLenByteDiff(unsigned char *p, unsigned int len) { static int zipPrevLenByteDiff(unsigned char *p, unsigned int len) {
unsigned int prevlensize; unsigned int prevlensize;
zipPrevDecodeLength(p,&prevlensize); ZIP_DECODE_PREVLENSIZE(p, prevlensize);
return zipPrevEncodeLength(NULL,len)-prevlensize; return zipPrevEncodeLength(NULL, len) - prevlensize;
}
/* Return the total number of bytes used by the entry pointed to by 'p'. */
static unsigned int zipRawEntryLength(unsigned char *p) {
unsigned int prevlensize, encoding, lensize, len;
ZIP_DECODE_PREVLENSIZE(p, prevlensize);
ZIP_DECODE_LENGTH(p + prevlensize, encoding, lensize, len);
return prevlensize + lensize + len;
} }
/* Check if string pointed to by 'entry' can be encoded as an integer. /* Check if string pointed to by 'entry' can be encoded as an integer.
...@@ -319,20 +330,14 @@ static int64_t zipLoadInteger(unsigned char *p, unsigned char encoding) { ...@@ -319,20 +330,14 @@ static int64_t zipLoadInteger(unsigned char *p, unsigned char encoding) {
/* Return a struct with all information about an entry. */ /* Return a struct with all information about an entry. */
static zlentry zipEntry(unsigned char *p) { static zlentry zipEntry(unsigned char *p) {
zlentry e; zlentry e;
e.prevrawlen = zipPrevDecodeLength(p,&e.prevrawlensize);
e.len = zipDecodeLength(p+e.prevrawlensize,&e.lensize); ZIP_DECODE_PREVLEN(p, e.prevrawlensize, e.prevrawlen);
e.headersize = e.prevrawlensize+e.lensize; ZIP_DECODE_LENGTH(p + e.prevrawlensize, e.encoding, e.lensize, e.len);
e.encoding = zipEntryEncoding(p+e.prevrawlensize); e.headersize = e.prevrawlensize + e.lensize;
e.p = p; e.p = p;
return e; return e;
} }
/* Return the total number of bytes used by the entry at "p". */
static unsigned int zipRawEntryLength(unsigned char *p) {
zlentry e = zipEntry(p);
return e.headersize + e.len;
}
/* Create a new empty ziplist. */ /* Create a new empty ziplist. */
unsigned char *ziplistNew(void) { unsigned char *ziplistNew(void) {
unsigned int bytes = ZIPLIST_HEADER_SIZE+1; unsigned int bytes = ZIPLIST_HEADER_SIZE+1;
...@@ -628,10 +633,14 @@ unsigned char *ziplistNext(unsigned char *zl, unsigned char *p) { ...@@ -628,10 +633,14 @@ unsigned char *ziplistNext(unsigned char *zl, unsigned char *p) {
* when the *next* element is ZIP_END (there is no next entry). */ * when the *next* element is ZIP_END (there is no next entry). */
if (p[0] == ZIP_END) { if (p[0] == ZIP_END) {
return NULL; return NULL;
} else {
p = p+zipRawEntryLength(p);
return (p[0] == ZIP_END) ? NULL : p;
} }
p += zipRawEntryLength(p);
if (p[0] == ZIP_END) {
return NULL;
}
return p;
} }
/* Return pointer to previous entry in ziplist. */ /* Return pointer to previous entry in ziplist. */
...@@ -729,6 +738,62 @@ unsigned int ziplistCompare(unsigned char *p, unsigned char *sstr, unsigned int ...@@ -729,6 +738,62 @@ unsigned int ziplistCompare(unsigned char *p, unsigned char *sstr, unsigned int
return 0; return 0;
} }
/* Find pointer to the entry equal to the specified entry. Skip 'skip' entries
* between every comparison. Returns NULL when the field could not be found. */
unsigned char *ziplistFind(unsigned char *p, unsigned char *vstr, unsigned int vlen, unsigned int skip) {
int skipcnt = 0;
unsigned char vencoding = 0;
long long vll = 0;
while (p[0] != ZIP_END) {
unsigned int prevlensize, encoding, lensize, len;
unsigned char *q;
ZIP_DECODE_PREVLENSIZE(p, prevlensize);
ZIP_DECODE_LENGTH(p + prevlensize, encoding, lensize, len);
q = p + prevlensize + lensize;
if (skipcnt == 0) {
/* Compare current entry with specified entry */
if (ZIP_IS_STR(encoding)) {
if (len == vlen && memcmp(q, vstr, vlen) == 0) {
return p;
}
} else {
/* Find out if the specified entry can be encoded */
if (vencoding == 0) {
/* UINT_MAX when the entry CANNOT be encoded */
if (!zipTryEncoding(vstr, vlen, &vll, &vencoding)) {
vencoding = UCHAR_MAX;
}
/* Must be non-zero by now */
assert(vencoding);
}
/* Compare current entry with specified entry */
if (encoding == vencoding) {
long long ll = zipLoadInteger(q, encoding);
if (ll == vll) {
return p;
}
}
}
/* Reset skip count */
skipcnt = skip;
} else {
/* Skip entry */
skipcnt--;
}
/* Move to next entry */
p = q + len;
}
return NULL;
}
/* Return length of ziplist. */ /* Return length of ziplist. */
unsigned int ziplistLen(unsigned char *zl) { unsigned int ziplistLen(unsigned char *zl) {
unsigned int len = 0; unsigned int len = 0;
......
...@@ -11,5 +11,6 @@ unsigned char *ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char ...@@ -11,5 +11,6 @@ unsigned char *ziplistInsert(unsigned char *zl, unsigned char *p, unsigned char
unsigned char *ziplistDelete(unsigned char *zl, unsigned char **p); unsigned char *ziplistDelete(unsigned char *zl, unsigned char **p);
unsigned char *ziplistDeleteRange(unsigned char *zl, unsigned int index, unsigned int num); unsigned char *ziplistDeleteRange(unsigned char *zl, unsigned int index, unsigned int num);
unsigned int ziplistCompare(unsigned char *p, unsigned char *s, unsigned int slen); unsigned int ziplistCompare(unsigned char *p, unsigned char *s, unsigned int slen);
unsigned char *ziplistFind(unsigned char *p, unsigned char *vstr, unsigned int vlen, unsigned int skip);
unsigned int ziplistLen(unsigned char *zl); unsigned int ziplistLen(unsigned char *zl);
size_t ziplistBlobLen(unsigned char *zl); size_t ziplistBlobLen(unsigned char *zl);
...@@ -297,8 +297,8 @@ no-appendfsync-on-rewrite no ...@@ -297,8 +297,8 @@ no-appendfsync-on-rewrite no
# have at max a given numer of elements, and the biggest element does not # have at max a given numer of elements, and the biggest element does not
# exceed a given threshold. You can configure this limits with the following # exceed a given threshold. You can configure this limits with the following
# configuration directives. # configuration directives.
hash-max-zipmap-entries 64 hash-max-ziplist-entries 64
hash-max-zipmap-value 512 hash-max-ziplist-value 512
# Similarly to hashes, small lists are also encoded in a special way in order # Similarly to hashes, small lists are also encoded in a special way in order
# to save a lot of space. The special representation is only used when # to save a lot of space. The special representation is only used when
......
set server_path [tmpdir "server.convert-zipmap-hash-on-load"]
# Copy RDB with zipmap encoded hash to server path
exec cp tests/assets/hash-zipmap.rdb $server_path
start_server [list overrides [list "dir" $server_path "dbfilename" "hash-zipmap.rdb"]] {
test "RDB load zipmap hash: converts to ziplist" {
r select 0
assert_match "*ziplist*" [r debug object hash]
assert_equal 2 [r hlen hash]
assert_match {v1 v2} [r hmget hash f1 f2]
}
}
start_server [list overrides [list "dir" $server_path "dbfilename" "hash-zipmap.rdb" "hash-max-ziplist-entries" 1]] {
test "RDB load zipmap hash: converts to hash table when hash-max-ziplist-entries is exceeded" {
r select 0
assert_match "*hashtable*" [r debug object hash]
assert_equal 2 [r hlen hash]
assert_match {v1 v2} [r hmget hash f1 f2]
}
}
start_server [list overrides [list "dir" $server_path "dbfilename" "hash-zipmap.rdb" "hash-max-ziplist-value" 1]] {
test "RDB load zipmap hash: converts to hash table when hash-max-ziplist-value is exceeded" {
r select 0
assert_match "*hashtable*" [r debug object hash]
assert_equal 2 [r hlen hash]
assert_match {v1 v2} [r hmget hash f1 f2]
}
}
set server_path [tmpdir "server.rdb-encoding-test"]
# Copy RDB with different encodings in server path
exec cp tests/assets/encodings.rdb $server_path
start_server [list overrides [list "dir" $server_path "dbfilename" "encodings.rdb"]] {
test "RDB encoding loading test" {
r select 0
csvdump r
} {"compressible","string","aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
"hash","hash","a","1","aa","10","aaa","100","b","2","bb","20","bbb","200","c","3","cc","30","ccc","300","ddd","400","eee","5000000000",
"hash_zipped","hash","a","1","b","2","c","3",
"list","list","1","2","3","a","b","c","100000","6000000000","1","2","3","a","b","c","100000","6000000000","1","2","3","a","b","c","100000","6000000000",
"list_zipped","list","1","2","3","a","b","c","100000","6000000000",
"number","string","10"
"set","set","1","100000","2","3","6000000000","a","b","c",
"set_zipped_1","set","1","2","3","4",
"set_zipped_2","set","100000","200000","300000","400000",
"set_zipped_3","set","1000000000","2000000000","3000000000","4000000000","5000000000","6000000000",
"string","string","Hello World"
"zset","zset","a","1","b","2","c","3","aa","10","bb","20","cc","30","aaa","100","bbb","200","ccc","300","aaaa","1000","cccc","123456789","bbbb","5000000000",
"zset_zipped","zset","a","1","b","2","c","3",
}
}
...@@ -160,7 +160,7 @@ proc ::redis::redis_read_reply fd { ...@@ -160,7 +160,7 @@ proc ::redis::redis_read_reply fd {
- {return -code error [redis_read_line $fd]} - {return -code error [redis_read_line $fd]}
$ {redis_bulk_read $fd} $ {redis_bulk_read $fd}
* {redis_multi_bulk_read $fd} * {redis_multi_bulk_read $fd}
default {return -code error "Bad protocol, $type as reply type byte"} default {return -code error "Bad protocol, '$type' as reply type byte"}
} }
} }
......
...@@ -30,6 +30,8 @@ set ::all_tests { ...@@ -30,6 +30,8 @@ set ::all_tests {
integration/replication-2 integration/replication-2
integration/replication-3 integration/replication-3
integration/aof integration/aof
integration/rdb
integration/convert-zipmap-hash-on-load
unit/pubsub unit/pubsub
unit/slowlog unit/slowlog
unit/scripting unit/scripting
......
...@@ -54,10 +54,10 @@ start_server {tags {"aofrw"}} { ...@@ -54,10 +54,10 @@ start_server {tags {"aofrw"}} {
} }
foreach d {string int} { foreach d {string int} {
foreach e {zipmap hashtable} { foreach e {ziplist hashtable} {
test "AOF rewrite of hash with $e encoding, $d data" { test "AOF rewrite of hash with $e encoding, $d data" {
r flushall r flushall
if {$e eq {zipmap}} {set len 10} else {set len 1000} if {$e eq {ziplist}} {set len 10} else {set len 1000}
for {set j 0} {$j < $len} {incr j} { for {set j 0} {$j < $len} {incr j} {
if {$d eq {string}} { if {$d eq {string}} {
set data [randstring 0 16 alpha] set data [randstring 0 16 alpha]
......
start_server {tags {"introspection"}} { start_server {tags {"introspection"}} {
test {CLIENT LIST} { test {CLIENT LIST} {
r client list r client list
} {*addr=*:* fd=* idle=* flags=N db=9 sub=0 psub=0 qbuf=0 obl=0 oll=0 omem=0 events=r cmd=client*} } {*addr=*:* fd=* age=* idle=* flags=N db=9 sub=0 psub=0 qbuf=0 qbuf-free=* obl=0 oll=0 omem=0 events=r cmd=client*}
} }
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment