Commit 8b511219 authored by Itamar Haber's avatar Itamar Haber
Browse files

Merge remote-tracking branch 'upstream/unstable' into help_subcommands

parents 51eb6cb3 62a4b817
...@@ -54,6 +54,7 @@ int keyspaceEventsStringToFlags(char *classes) { ...@@ -54,6 +54,7 @@ int keyspaceEventsStringToFlags(char *classes) {
case 'e': flags |= NOTIFY_EVICTED; break; case 'e': flags |= NOTIFY_EVICTED; break;
case 'K': flags |= NOTIFY_KEYSPACE; break; case 'K': flags |= NOTIFY_KEYSPACE; break;
case 'E': flags |= NOTIFY_KEYEVENT; break; case 'E': flags |= NOTIFY_KEYEVENT; break;
case 't': flags |= NOTIFY_STREAM; break;
default: return -1; default: return -1;
} }
} }
...@@ -79,6 +80,7 @@ sds keyspaceEventsFlagsToString(int flags) { ...@@ -79,6 +80,7 @@ sds keyspaceEventsFlagsToString(int flags) {
if (flags & NOTIFY_ZSET) res = sdscatlen(res,"z",1); if (flags & NOTIFY_ZSET) res = sdscatlen(res,"z",1);
if (flags & NOTIFY_EXPIRED) res = sdscatlen(res,"x",1); if (flags & NOTIFY_EXPIRED) res = sdscatlen(res,"x",1);
if (flags & NOTIFY_EVICTED) res = sdscatlen(res,"e",1); if (flags & NOTIFY_EVICTED) res = sdscatlen(res,"e",1);
if (flags & NOTIFY_STREAM) res = sdscatlen(res,"t",1);
} }
if (flags & NOTIFY_KEYSPACE) res = sdscatlen(res,"K",1); if (flags & NOTIFY_KEYSPACE) res = sdscatlen(res,"K",1);
if (flags & NOTIFY_KEYEVENT) res = sdscatlen(res,"E",1); if (flags & NOTIFY_KEYEVENT) res = sdscatlen(res,"E",1);
......
...@@ -232,6 +232,13 @@ robj *createZsetZiplistObject(void) { ...@@ -232,6 +232,13 @@ robj *createZsetZiplistObject(void) {
return o; return o;
} }
robj *createStreamObject(void) {
stream *s = streamNew();
robj *o = createObject(OBJ_STREAM,s);
o->encoding = OBJ_ENCODING_STREAM;
return o;
}
robj *createModuleObject(moduleType *mt, void *value) { robj *createModuleObject(moduleType *mt, void *value) {
moduleValue *mv = zmalloc(sizeof(*mv)); moduleValue *mv = zmalloc(sizeof(*mv));
mv->type = mt; mv->type = mt;
...@@ -303,6 +310,10 @@ void freeModuleObject(robj *o) { ...@@ -303,6 +310,10 @@ void freeModuleObject(robj *o) {
zfree(mv); zfree(mv);
} }
void freeStreamObject(robj *o) {
freeStream(o->ptr);
}
void incrRefCount(robj *o) { void incrRefCount(robj *o) {
if (o->refcount != OBJ_SHARED_REFCOUNT) o->refcount++; if (o->refcount != OBJ_SHARED_REFCOUNT) o->refcount++;
} }
...@@ -316,6 +327,7 @@ void decrRefCount(robj *o) { ...@@ -316,6 +327,7 @@ void decrRefCount(robj *o) {
case OBJ_ZSET: freeZsetObject(o); break; case OBJ_ZSET: freeZsetObject(o); break;
case OBJ_HASH: freeHashObject(o); break; case OBJ_HASH: freeHashObject(o); break;
case OBJ_MODULE: freeModuleObject(o); break; case OBJ_MODULE: freeModuleObject(o); break;
case OBJ_STREAM: freeStreamObject(o); break;
default: serverPanic("Unknown object type"); break; default: serverPanic("Unknown object type"); break;
} }
zfree(o); zfree(o);
...@@ -788,6 +800,49 @@ size_t objectComputeSize(robj *o, size_t sample_size) { ...@@ -788,6 +800,49 @@ size_t objectComputeSize(robj *o, size_t sample_size) {
} else { } else {
serverPanic("Unknown hash encoding"); serverPanic("Unknown hash encoding");
} }
} else if (o->type == OBJ_STREAM) {
stream *s = o->ptr;
/* Note: to guess the size of the radix tree is not trivial, so we
* approximate it considering 64 bytes of data overhead for each
* key (the ID), and then adding the number of bare nodes, plus some
* overhead due by the data and child pointers. This secret recipe
* was obtained by checking the average radix tree created by real
* workloads, and then adjusting the constants to get numbers that
* more or less match the real memory usage.
*
* Actually the number of nodes and keys may be different depending
* on the insertion speed and thus the ability of the radix tree
* to compress prefixes. */
asize = sizeof(*o);
asize += s->rax->numele * 64;
asize += s->rax->numnodes * sizeof(raxNode);
asize += s->rax->numnodes * 32*7; /* Add a few child pointers... */
/* Now we have to add the listpacks. The last listpack is often non
* complete, so we estimate the size of the first N listpacks, and
* use the average to compute the size of the first N-1 listpacks, and
* finally add the real size of the last node. */
raxIterator ri;
raxStart(&ri,s->rax);
raxSeek(&ri,"^",NULL,0);
size_t lpsize = 0, samples = 0;
while(samples < sample_size && raxNext(&ri)) {
unsigned char *lp = ri.data;
lpsize += lpBytes(lp);
samples++;
}
if (s->rax->numele <= samples) {
asize += lpsize;
} else {
if (samples) lpsize /= samples; /* Compute the average. */
asize += lpsize * (s->rax->numele-1);
/* No need to check if seek succeeded, we enter this branch only
* if there are a few elements in the radix tree. */
raxSeek(&ri,"$",NULL,0);
raxNext(&ri);
asize += lpBytes(ri.data);
}
raxStop(&ri);
} else if (o->type == OBJ_MODULE) { } else if (o->type == OBJ_MODULE) {
moduleValue *mv = o->ptr; moduleValue *mv = o->ptr;
moduleType *mt = mv->type; moduleType *mt = mv->type;
...@@ -1045,10 +1100,14 @@ void objectCommand(client *c) { ...@@ -1045,10 +1100,14 @@ void objectCommand(client *c) {
if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.nullbulk)) if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.nullbulk))
== NULL) return; == NULL) return;
if (!(server.maxmemory_policy & MAXMEMORY_FLAG_LFU)) { if (!(server.maxmemory_policy & MAXMEMORY_FLAG_LFU)) {
addReplyError(c,"A non-LFU maxmemory policy is selected, access frequency not tracked. Please note that when switching between policies at runtime LRU and LFU data will take some time to adjust."); addReplyError(c,"An LFU maxmemory policy is not selected, access frequency not tracked. Please note that when switching between policies at runtime LRU and LFU data will take some time to adjust.");
return; return;
} }
addReplyLongLong(c,o->lru&255); /* LFUDecrAndReturn should be called
* in case of the key has not been accessed for a long time,
* because we update the access time only
* when the key is read or overwritten. */
addReplyLongLong(c,LFUDecrAndReturn(o));
} else { } else {
addReplyErrorFormat(c, "Unknown subcommand or wrong number of arguments for '%s'. Try OBJECT help", addReplyErrorFormat(c, "Unknown subcommand or wrong number of arguments for '%s'. Try OBJECT help",
(char *)c->argv[1]->ptr); (char *)c->argv[1]->ptr);
......
...@@ -149,7 +149,7 @@ REDIS_STATIC quicklistNode *quicklistCreateNode(void) { ...@@ -149,7 +149,7 @@ REDIS_STATIC quicklistNode *quicklistCreateNode(void) {
} }
/* Return cached quicklist count */ /* Return cached quicklist count */
unsigned int quicklistCount(const quicklist *ql) { return ql->count; } unsigned long quicklistCount(const quicklist *ql) { return ql->count; }
/* Free entire quicklist. */ /* Free entire quicklist. */
void quicklistRelease(quicklist *quicklist) { void quicklistRelease(quicklist *quicklist) {
......
...@@ -64,7 +64,7 @@ typedef struct quicklistLZF { ...@@ -64,7 +64,7 @@ typedef struct quicklistLZF {
char compressed[]; char compressed[];
} quicklistLZF; } quicklistLZF;
/* quicklist is a 32 byte struct (on 64-bit systems) describing a quicklist. /* quicklist is a 40 byte struct (on 64-bit systems) describing a quicklist.
* 'count' is the number of total entries. * 'count' is the number of total entries.
* 'len' is the number of quicklist nodes. * 'len' is the number of quicklist nodes.
* 'compress' is: -1 if compression disabled, otherwise it's the number * 'compress' is: -1 if compression disabled, otherwise it's the number
...@@ -74,7 +74,7 @@ typedef struct quicklist { ...@@ -74,7 +74,7 @@ typedef struct quicklist {
quicklistNode *head; quicklistNode *head;
quicklistNode *tail; quicklistNode *tail;
unsigned long count; /* total count of all entries in all ziplists */ unsigned long count; /* total count of all entries in all ziplists */
unsigned int len; /* number of quicklistNodes */ unsigned long len; /* number of quicklistNodes */
int fill : 16; /* fill factor for individual nodes */ int fill : 16; /* fill factor for individual nodes */
unsigned int compress : 16; /* depth of end nodes not to compress;0=off */ unsigned int compress : 16; /* depth of end nodes not to compress;0=off */
} quicklist; } quicklist;
...@@ -154,7 +154,7 @@ int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data, ...@@ -154,7 +154,7 @@ int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data,
void *(*saver)(unsigned char *data, unsigned int sz)); void *(*saver)(unsigned char *data, unsigned int sz));
int quicklistPop(quicklist *quicklist, int where, unsigned char **data, int quicklistPop(quicklist *quicklist, int where, unsigned char **data,
unsigned int *sz, long long *slong); unsigned int *sz, long long *slong);
unsigned int quicklistCount(const quicklist *ql); unsigned long quicklistCount(const quicklist *ql);
int quicklistCompare(unsigned char *p1, unsigned char *p2, int p2_len); int quicklistCompare(unsigned char *p1, unsigned char *p2, int p2_len);
size_t quicklistGetLzf(const quicklistNode *node, void **data); size_t quicklistGetLzf(const quicklistNode *node, void **data);
......
...@@ -131,7 +131,7 @@ static inline void raxStackFree(raxStack *ts) { ...@@ -131,7 +131,7 @@ static inline void raxStackFree(raxStack *ts) {
} }
/* ---------------------------------------------------------------------------- /* ----------------------------------------------------------------------------
* Radis tree implementation * Radix tree implementation
* --------------------------------------------------------------------------*/ * --------------------------------------------------------------------------*/
/* Allocate a new non compressed node with the specified number of children. /* Allocate a new non compressed node with the specified number of children.
...@@ -873,7 +873,8 @@ raxNode *raxRemoveChild(raxNode *parent, raxNode *child) { ...@@ -873,7 +873,8 @@ raxNode *raxRemoveChild(raxNode *parent, raxNode *child) {
memmove(((char*)cp)-1,cp,(parent->size-taillen-1)*sizeof(raxNode**)); memmove(((char*)cp)-1,cp,(parent->size-taillen-1)*sizeof(raxNode**));
/* Move the remaining "tail" pointer at the right position as well. */ /* Move the remaining "tail" pointer at the right position as well. */
memmove(((char*)c)-1,c+1,taillen*sizeof(raxNode**)+parent->iskey*sizeof(void*)); size_t valuelen = (parent->iskey && !parent->isnull) ? sizeof(void*) : 0;
memmove(((char*)c)-1,c+1,taillen*sizeof(raxNode**)+valuelen);
/* 4. Update size. */ /* 4. Update size. */
parent->size--; parent->size--;
...@@ -1092,28 +1093,36 @@ int raxRemove(rax *rax, unsigned char *s, size_t len, void **old) { ...@@ -1092,28 +1093,36 @@ int raxRemove(rax *rax, unsigned char *s, size_t len, void **old) {
/* This is the core of raxFree(): performs a depth-first scan of the /* This is the core of raxFree(): performs a depth-first scan of the
* tree and releases all the nodes found. */ * tree and releases all the nodes found. */
void raxRecursiveFree(rax *rax, raxNode *n) { void raxRecursiveFree(rax *rax, raxNode *n, void (*free_callback)(void*)) {
debugnode("free traversing",n); debugnode("free traversing",n);
int numchildren = n->iscompr ? 1 : n->size; int numchildren = n->iscompr ? 1 : n->size;
raxNode **cp = raxNodeLastChildPtr(n); raxNode **cp = raxNodeLastChildPtr(n);
while(numchildren--) { while(numchildren--) {
raxNode *child; raxNode *child;
memcpy(&child,cp,sizeof(child)); memcpy(&child,cp,sizeof(child));
raxRecursiveFree(rax,child); raxRecursiveFree(rax,child,free_callback);
cp--; cp--;
} }
debugnode("free depth-first",n); debugnode("free depth-first",n);
if (free_callback && n->iskey && !n->isnull)
free_callback(raxGetData(n));
rax_free(n); rax_free(n);
rax->numnodes--; rax->numnodes--;
} }
/* Free a whole radix tree. */ /* Free a whole radix tree, calling the specified callback in order to
void raxFree(rax *rax) { * free the auxiliary data. */
raxRecursiveFree(rax,rax->head); void raxFreeWithCallback(rax *rax, void (*free_callback)(void*)) {
raxRecursiveFree(rax,rax->head,free_callback);
assert(rax->numnodes == 0); assert(rax->numnodes == 0);
rax_free(rax); rax_free(rax);
} }
/* Free a whole radix tree. */
void raxFree(rax *rax) {
raxFreeWithCallback(rax,NULL);
}
/* ------------------------------- Iterator --------------------------------- */ /* ------------------------------- Iterator --------------------------------- */
/* Initialize a Rax iterator. This call should be performed a single time /* Initialize a Rax iterator. This call should be performed a single time
...@@ -1175,7 +1184,7 @@ void raxIteratorDelChars(raxIterator *it, size_t count) { ...@@ -1175,7 +1184,7 @@ void raxIteratorDelChars(raxIterator *it, size_t count) {
* The function returns 1 on success or 0 on out of memory. */ * The function returns 1 on success or 0 on out of memory. */
int raxIteratorNextStep(raxIterator *it, int noup) { int raxIteratorNextStep(raxIterator *it, int noup) {
if (it->flags & RAX_ITER_EOF) { if (it->flags & RAX_ITER_EOF) {
return 0; return 1;
} else if (it->flags & RAX_ITER_JUST_SEEKED) { } else if (it->flags & RAX_ITER_JUST_SEEKED) {
it->flags &= ~RAX_ITER_JUST_SEEKED; it->flags &= ~RAX_ITER_JUST_SEEKED;
return 1; return 1;
...@@ -1187,10 +1196,6 @@ int raxIteratorNextStep(raxIterator *it, int noup) { ...@@ -1187,10 +1196,6 @@ int raxIteratorNextStep(raxIterator *it, int noup) {
size_t orig_stack_items = it->stack.items; size_t orig_stack_items = it->stack.items;
raxNode *orig_node = it->node; raxNode *orig_node = it->node;
/* Clear the EOF flag: it will be set again if the EOF condition
* is still valid. */
it->flags &= ~RAX_ITER_EOF;
while(1) { while(1) {
int children = it->node->iscompr ? 1 : it->node->size; int children = it->node->iscompr ? 1 : it->node->size;
if (!noup && children) { if (!noup && children) {
...@@ -1291,7 +1296,7 @@ int raxSeekGreatest(raxIterator *it) { ...@@ -1291,7 +1296,7 @@ int raxSeekGreatest(raxIterator *it) {
* effect to the one of raxIteratorPrevSte(). */ * effect to the one of raxIteratorPrevSte(). */
int raxIteratorPrevStep(raxIterator *it, int noup) { int raxIteratorPrevStep(raxIterator *it, int noup) {
if (it->flags & RAX_ITER_EOF) { if (it->flags & RAX_ITER_EOF) {
return 0; return 1;
} else if (it->flags & RAX_ITER_JUST_SEEKED) { } else if (it->flags & RAX_ITER_JUST_SEEKED) {
it->flags &= ~RAX_ITER_JUST_SEEKED; it->flags &= ~RAX_ITER_JUST_SEEKED;
return 1; return 1;
...@@ -1412,6 +1417,7 @@ int raxSeek(raxIterator *it, const char *op, unsigned char *ele, size_t len) { ...@@ -1412,6 +1417,7 @@ int raxSeek(raxIterator *it, const char *op, unsigned char *ele, size_t len) {
it->node = it->rt->head; it->node = it->rt->head;
if (!raxSeekGreatest(it)) return 0; if (!raxSeekGreatest(it)) return 0;
assert(it->node->iskey); assert(it->node->iskey);
it->data = raxGetData(it->node);
return 1; return 1;
} }
...@@ -1430,6 +1436,7 @@ int raxSeek(raxIterator *it, const char *op, unsigned char *ele, size_t len) { ...@@ -1430,6 +1436,7 @@ int raxSeek(raxIterator *it, const char *op, unsigned char *ele, size_t len) {
/* We found our node, since the key matches and we have an /* We found our node, since the key matches and we have an
* "equal" condition. */ * "equal" condition. */
if (!raxIteratorAddChars(it,ele,len)) return 0; /* OOM. */ if (!raxIteratorAddChars(it,ele,len)) return 0; /* OOM. */
it->data = raxGetData(it->node);
} else if (lt || gt) { } else if (lt || gt) {
/* Exact key not found or eq flag not set. We have to set as current /* Exact key not found or eq flag not set. We have to set as current
* key the one represented by the node we stopped at, and perform * key the one represented by the node we stopped at, and perform
...@@ -1502,6 +1509,7 @@ int raxSeek(raxIterator *it, const char *op, unsigned char *ele, size_t len) { ...@@ -1502,6 +1509,7 @@ int raxSeek(raxIterator *it, const char *op, unsigned char *ele, size_t len) {
* the previous sub-tree. */ * the previous sub-tree. */
if (nodechar < keychar) { if (nodechar < keychar) {
if (!raxSeekGreatest(it)) return 0; if (!raxSeekGreatest(it)) return 0;
it->data = raxGetData(it->node);
} else { } else {
if (!raxIteratorAddChars(it,it->node->data,it->node->size)) if (!raxIteratorAddChars(it,it->node->data,it->node->size))
return 0; return 0;
...@@ -1647,6 +1655,19 @@ void raxStop(raxIterator *it) { ...@@ -1647,6 +1655,19 @@ void raxStop(raxIterator *it) {
raxStackFree(&it->stack); raxStackFree(&it->stack);
} }
/* Return if the iterator is in an EOF state. This happens when raxSeek()
* failed to seek an appropriate element, so that raxNext() or raxPrev()
* will return zero, or when an EOF condition was reached while iterating
* with raxNext() and raxPrev(). */
int raxEOF(raxIterator *it) {
return it->flags & RAX_ITER_EOF;
}
/* Return the number of elements inside the radix tree. */
uint64_t raxSize(rax *rax) {
return rax->numele;
}
/* ----------------------------- Introspection ------------------------------ */ /* ----------------------------- Introspection ------------------------------ */
/* This function is mostly used for debugging and learning purposes. /* This function is mostly used for debugging and learning purposes.
......
...@@ -148,6 +148,7 @@ int raxInsert(rax *rax, unsigned char *s, size_t len, void *data, void **old); ...@@ -148,6 +148,7 @@ int raxInsert(rax *rax, unsigned char *s, size_t len, void *data, void **old);
int raxRemove(rax *rax, unsigned char *s, size_t len, void **old); int raxRemove(rax *rax, unsigned char *s, size_t len, void **old);
void *raxFind(rax *rax, unsigned char *s, size_t len); void *raxFind(rax *rax, unsigned char *s, size_t len);
void raxFree(rax *rax); void raxFree(rax *rax);
void raxFreeWithCallback(rax *rax, void (*free_callback)(void*));
void raxStart(raxIterator *it, rax *rt); void raxStart(raxIterator *it, rax *rt);
int raxSeek(raxIterator *it, const char *op, unsigned char *ele, size_t len); int raxSeek(raxIterator *it, const char *op, unsigned char *ele, size_t len);
int raxNext(raxIterator *it); int raxNext(raxIterator *it);
...@@ -155,6 +156,8 @@ int raxPrev(raxIterator *it); ...@@ -155,6 +156,8 @@ int raxPrev(raxIterator *it);
int raxRandomWalk(raxIterator *it, size_t steps); int raxRandomWalk(raxIterator *it, size_t steps);
int raxCompare(raxIterator *iter, const char *op, unsigned char *key, size_t key_len); int raxCompare(raxIterator *iter, const char *op, unsigned char *key, size_t key_len);
void raxStop(raxIterator *it); void raxStop(raxIterator *it);
int raxEOF(raxIterator *it);
void raxShow(rax *rax); void raxShow(rax *rax);
uint64_t raxSize(rax *rax);
#endif #endif
...@@ -31,6 +31,7 @@ ...@@ -31,6 +31,7 @@
#include "lzf.h" /* LZF compression library */ #include "lzf.h" /* LZF compression library */
#include "zipmap.h" #include "zipmap.h"
#include "endianconv.h" #include "endianconv.h"
#include "stream.h"
#include <math.h> #include <math.h>
#include <sys/types.h> #include <sys/types.h>
...@@ -622,6 +623,8 @@ int rdbSaveObjectType(rio *rdb, robj *o) { ...@@ -622,6 +623,8 @@ int rdbSaveObjectType(rio *rdb, robj *o) {
return rdbSaveType(rdb,RDB_TYPE_HASH); return rdbSaveType(rdb,RDB_TYPE_HASH);
else else
serverPanic("Unknown hash encoding"); serverPanic("Unknown hash encoding");
case OBJ_STREAM:
return rdbSaveType(rdb,RDB_TYPE_STREAM_LISTPACKS);
case OBJ_MODULE: case OBJ_MODULE:
return rdbSaveType(rdb,RDB_TYPE_MODULE_2); return rdbSaveType(rdb,RDB_TYPE_MODULE_2);
default: default:
...@@ -762,7 +765,39 @@ ssize_t rdbSaveObject(rio *rdb, robj *o) { ...@@ -762,7 +765,39 @@ ssize_t rdbSaveObject(rio *rdb, robj *o) {
} else { } else {
serverPanic("Unknown hash encoding"); serverPanic("Unknown hash encoding");
} }
} else if (o->type == OBJ_STREAM) {
/* Store how many listpacks we have inside the radix tree. */
stream *s = o->ptr;
rax *rax = s->rax;
if ((n = rdbSaveLen(rdb,raxSize(rax))) == -1) return -1;
nwritten += n;
/* Serialize all the listpacks inside the radix tree as they are,
* when loading back, we'll use the first entry of each listpack
* to insert it back into the radix tree. */
raxIterator ri;
raxStart(&ri,rax);
raxSeek(&ri,"^",NULL,0);
while (raxNext(&ri)) {
unsigned char *lp = ri.data;
size_t lp_bytes = lpBytes(lp);
if ((n = rdbSaveRawString(rdb,ri.key,ri.key_len)) == -1) return -1;
nwritten += n;
if ((n = rdbSaveRawString(rdb,lp,lp_bytes)) == -1) return -1;
nwritten += n;
}
raxStop(&ri);
/* Save the number of elements inside the stream. We cannot obtain
* this easily later, since our macro nodes should be checked for
* number of items: not a great CPU / space tradeoff. */
if ((n = rdbSaveLen(rdb,s->length)) == -1) return -1;
nwritten += n;
/* Save the last entry ID. */
if ((n = rdbSaveLen(rdb,s->last_id.ms)) == -1) return -1;
nwritten += n;
if ((n = rdbSaveLen(rdb,s->last_id.seq)) == -1) return -1;
nwritten += n;
} else if (o->type == OBJ_MODULE) { } else if (o->type == OBJ_MODULE) {
/* Save a module-specific value. */ /* Save a module-specific value. */
RedisModuleIO io; RedisModuleIO io;
...@@ -943,6 +978,20 @@ int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi) { ...@@ -943,6 +978,20 @@ int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi) {
} }
di = NULL; /* So that we don't release it again on error. */ di = NULL; /* So that we don't release it again on error. */
/* If we are storing the replication information on disk, persist
* the script cache as well: on successful PSYNC after a restart, we need
* to be able to process any EVALSHA inside the replication backlog the
* master will send us. */
if (rsi && dictSize(server.lua_scripts)) {
di = dictGetIterator(server.lua_scripts);
while((de = dictNext(di)) != NULL) {
robj *body = dictGetVal(de);
if (rdbSaveAuxField(rdb,"lua",3,body->ptr,sdslen(body->ptr)) == -1)
goto werr;
}
dictReleaseIterator(di);
}
/* EOF opcode */ /* EOF opcode */
if (rdbSaveType(rdb,RDB_OPCODE_EOF) == -1) goto werr; if (rdbSaveType(rdb,RDB_OPCODE_EOF) == -1) goto werr;
...@@ -1395,6 +1444,45 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) { ...@@ -1395,6 +1444,45 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
rdbExitReportCorruptRDB("Unknown RDB encoding type %d",rdbtype); rdbExitReportCorruptRDB("Unknown RDB encoding type %d",rdbtype);
break; break;
} }
} else if (rdbtype == RDB_TYPE_STREAM_LISTPACKS) {
o = createStreamObject();
stream *s = o->ptr;
uint64_t listpacks = rdbLoadLen(rdb,NULL);
while(listpacks--) {
/* Get the master ID, the one we'll use as key of the radix tree
* node: the entries inside the listpack itself are delta-encoded
* relatively to this ID. */
sds nodekey = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS,NULL);
if (sdslen(nodekey) != sizeof(streamID)) {
rdbExitReportCorruptRDB("Stream node key entry is not the "
"size of a stream ID");
}
/* Load the listpack. */
unsigned char *lp =
rdbGenericLoadStringObject(rdb,RDB_LOAD_PLAIN,NULL);
if (lp == NULL) return NULL;
unsigned char *first = lpFirst(lp);
if (first == NULL) {
/* Serialized listpacks should never be empty, since on
* deletion we should remove the radix tree key if the
* resulting listpack is emtpy. */
rdbExitReportCorruptRDB("Empty listpack inside stream");
}
/* Insert the key in the radix tree. */
int retval = raxInsert(s->rax,
(unsigned char*)nodekey,sizeof(streamID),lp,NULL);
sdsfree(nodekey);
if (!retval)
rdbExitReportCorruptRDB("Listpack re-added with existing key");
}
/* Load total number of items inside the stream. */
s->length = rdbLoadLen(rdb,NULL);
/* Load the last entry ID. */
s->last_id.ms = rdbLoadLen(rdb,NULL);
s->last_id.seq = rdbLoadLen(rdb,NULL);
} else if (rdbtype == RDB_TYPE_MODULE || rdbtype == RDB_TYPE_MODULE_2) { } else if (rdbtype == RDB_TYPE_MODULE || rdbtype == RDB_TYPE_MODULE_2) {
uint64_t moduleid = rdbLoadLen(rdb,NULL); uint64_t moduleid = rdbLoadLen(rdb,NULL);
moduleType *mt = moduleTypeLookupModuleByID(moduleid); moduleType *mt = moduleTypeLookupModuleByID(moduleid);
...@@ -1589,6 +1677,13 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi) { ...@@ -1589,6 +1677,13 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi) {
} }
} else if (!strcasecmp(auxkey->ptr,"repl-offset")) { } else if (!strcasecmp(auxkey->ptr,"repl-offset")) {
if (rsi) rsi->repl_offset = strtoll(auxval->ptr,NULL,10); if (rsi) rsi->repl_offset = strtoll(auxval->ptr,NULL,10);
} else if (!strcasecmp(auxkey->ptr,"lua")) {
/* Load the script back in memory. */
if (luaCreateFunction(NULL,server.lua,auxval) == NULL) {
rdbExitReportCorruptRDB(
"Can't load Lua script from RDB file! "
"BODY: %s", auxval->ptr);
}
} else { } else {
/* We ignore fields we don't understand, as by AUX field /* We ignore fields we don't understand, as by AUX field
* contract. */ * contract. */
......
...@@ -69,8 +69,9 @@ ...@@ -69,8 +69,9 @@
#define RDB_ENC_INT32 2 /* 32 bit signed integer */ #define RDB_ENC_INT32 2 /* 32 bit signed integer */
#define RDB_ENC_LZF 3 /* string compressed with FASTLZ */ #define RDB_ENC_LZF 3 /* string compressed with FASTLZ */
/* Dup object types to RDB object types. Only reason is readability (are we /* Map object types to RDB object types. Macros starting with OBJ_ are for
* dealing with RDB types or with in-memory object types?). */ * memory storage and may change. Instead RDB types must be fixed because
* we store them on disk. */
#define RDB_TYPE_STRING 0 #define RDB_TYPE_STRING 0
#define RDB_TYPE_LIST 1 #define RDB_TYPE_LIST 1
#define RDB_TYPE_SET 2 #define RDB_TYPE_SET 2
...@@ -89,10 +90,11 @@ ...@@ -89,10 +90,11 @@
#define RDB_TYPE_ZSET_ZIPLIST 12 #define RDB_TYPE_ZSET_ZIPLIST 12
#define RDB_TYPE_HASH_ZIPLIST 13 #define RDB_TYPE_HASH_ZIPLIST 13
#define RDB_TYPE_LIST_QUICKLIST 14 #define RDB_TYPE_LIST_QUICKLIST 14
#define RDB_TYPE_STREAM_LISTPACKS 15
/* NOTE: WHEN ADDING NEW RDB TYPE, UPDATE rdbIsObjectType() BELOW */ /* NOTE: WHEN ADDING NEW RDB TYPE, UPDATE rdbIsObjectType() BELOW */
/* Test if a type is an object type. */ /* Test if a type is an object type. */
#define rdbIsObjectType(t) ((t >= 0 && t <= 7) || (t >= 9 && t <= 14)) #define rdbIsObjectType(t) ((t >= 0 && t <= 7) || (t >= 9 && t <= 15))
/* Special RDB opcodes (saved/loaded with rdbSaveType/rdbLoadType). */ /* Special RDB opcodes (saved/loaded with rdbSaveType/rdbLoadType). */
#define RDB_OPCODE_AUX 250 #define RDB_OPCODE_AUX 250
......
...@@ -193,12 +193,12 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) { ...@@ -193,12 +193,12 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) {
buf[9] = '\0'; buf[9] = '\0';
if (memcmp(buf,"REDIS",5) != 0) { if (memcmp(buf,"REDIS",5) != 0) {
rdbCheckError("Wrong signature trying to load DB from file"); rdbCheckError("Wrong signature trying to load DB from file");
return 1; goto err;
} }
rdbver = atoi(buf+5); rdbver = atoi(buf+5);
if (rdbver < 1 || rdbver > RDB_VERSION) { if (rdbver < 1 || rdbver > RDB_VERSION) {
rdbCheckError("Can't handle RDB format version %d",rdbver); rdbCheckError("Can't handle RDB format version %d",rdbver);
return 1; goto err;
} }
startLoading(fp); startLoading(fp);
...@@ -270,7 +270,7 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) { ...@@ -270,7 +270,7 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) {
} else { } else {
if (!rdbIsObjectType(type)) { if (!rdbIsObjectType(type)) {
rdbCheckError("Invalid object type: %d", type); rdbCheckError("Invalid object type: %d", type);
return 1; goto err;
} }
rdbstate.key_type = type; rdbstate.key_type = type;
} }
...@@ -307,6 +307,7 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) { ...@@ -307,6 +307,7 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) {
rdbCheckInfo("RDB file was saved with checksum disabled: no check performed."); rdbCheckInfo("RDB file was saved with checksum disabled: no check performed.");
} else if (cksum != expected) { } else if (cksum != expected) {
rdbCheckError("RDB CRC error"); rdbCheckError("RDB CRC error");
goto err;
} else { } else {
rdbCheckInfo("Checksum OK"); rdbCheckInfo("Checksum OK");
} }
...@@ -321,6 +322,8 @@ eoferr: /* unexpected end of file is handled here with a fatal exit */ ...@@ -321,6 +322,8 @@ eoferr: /* unexpected end of file is handled here with a fatal exit */
} else { } else {
rdbCheckError("Unexpected EOF reading RDB file"); rdbCheckError("Unexpected EOF reading RDB file");
} }
err:
if (closefile) fclose(fp);
return 1; return 1;
} }
......
...@@ -107,6 +107,7 @@ static struct config { ...@@ -107,6 +107,7 @@ static struct config {
char *pattern; char *pattern;
char *rdb_filename; char *rdb_filename;
int bigkeys; int bigkeys;
int hotkeys;
int stdinarg; /* get last arg from stdin. (-x option) */ int stdinarg; /* get last arg from stdin. (-x option) */
char *auth; char *auth;
int output; /* output mode, see OUTPUT_* defines */ int output; /* output mode, see OUTPUT_* defines */
...@@ -710,7 +711,7 @@ int isColorTerm(void) { ...@@ -710,7 +711,7 @@ int isColorTerm(void) {
return t != NULL && strstr(t,"xterm") != NULL; return t != NULL && strstr(t,"xterm") != NULL;
} }
/* Helpe function for sdsCatColorizedLdbReply() appending colorize strings /* Helper function for sdsCatColorizedLdbReply() appending colorize strings
* to an SDS string. */ * to an SDS string. */
sds sdscatcolor(sds o, char *s, size_t len, char *color) { sds sdscatcolor(sds o, char *s, size_t len, char *color) {
if (!isColorTerm()) return sdscatlen(o,s,len); if (!isColorTerm()) return sdscatlen(o,s,len);
...@@ -1129,6 +1130,8 @@ static int parseOptions(int argc, char **argv) { ...@@ -1129,6 +1130,8 @@ static int parseOptions(int argc, char **argv) {
config.pipe_timeout = atoi(argv[++i]); config.pipe_timeout = atoi(argv[++i]);
} else if (!strcmp(argv[i],"--bigkeys")) { } else if (!strcmp(argv[i],"--bigkeys")) {
config.bigkeys = 1; config.bigkeys = 1;
} else if (!strcmp(argv[i],"--hotkeys")) {
config.hotkeys = 1;
} else if (!strcmp(argv[i],"--eval") && !lastarg) { } else if (!strcmp(argv[i],"--eval") && !lastarg) {
config.eval = argv[++i]; config.eval = argv[++i];
} else if (!strcmp(argv[i],"--ldb")) { } else if (!strcmp(argv[i],"--ldb")) {
...@@ -1229,6 +1232,8 @@ static void usage(void) { ...@@ -1229,6 +1232,8 @@ static void usage(void) {
" no reply is received within <n> seconds.\n" " no reply is received within <n> seconds.\n"
" Default timeout: %d. Use 0 to wait forever.\n" " Default timeout: %d. Use 0 to wait forever.\n"
" --bigkeys Sample Redis keys looking for big keys.\n" " --bigkeys Sample Redis keys looking for big keys.\n"
" --hotkeys Sample Redis keys looking for hot keys.\n"
" only works when maxmemory-policy is *lfu.\n"
" --scan List all keys using the SCAN command.\n" " --scan List all keys using the SCAN command.\n"
" --pattern <pat> Useful with --scan to specify a SCAN pattern.\n" " --pattern <pat> Useful with --scan to specify a SCAN pattern.\n"
" --intrinsic-latency <sec> Run a test to measure intrinsic system latency.\n" " --intrinsic-latency <sec> Run a test to measure intrinsic system latency.\n"
...@@ -2069,7 +2074,8 @@ static void pipeMode(void) { ...@@ -2069,7 +2074,8 @@ static void pipeMode(void) {
#define TYPE_SET 2 #define TYPE_SET 2
#define TYPE_HASH 3 #define TYPE_HASH 3
#define TYPE_ZSET 4 #define TYPE_ZSET 4
#define TYPE_NONE 5 #define TYPE_STREAM 5
#define TYPE_NONE 6
static redisReply *sendScan(unsigned long long *it) { static redisReply *sendScan(unsigned long long *it) {
redisReply *reply = redisCommand(context, "SCAN %llu", *it); redisReply *reply = redisCommand(context, "SCAN %llu", *it);
...@@ -2128,6 +2134,8 @@ static int toIntType(char *key, char *type) { ...@@ -2128,6 +2134,8 @@ static int toIntType(char *key, char *type) {
return TYPE_HASH; return TYPE_HASH;
} else if(!strcmp(type, "zset")) { } else if(!strcmp(type, "zset")) {
return TYPE_ZSET; return TYPE_ZSET;
} else if(!strcmp(type, "stream")) {
return TYPE_STREAM;
} else if(!strcmp(type, "none")) { } else if(!strcmp(type, "none")) {
return TYPE_NONE; return TYPE_NONE;
} else { } else {
...@@ -2216,7 +2224,7 @@ static void findBigKeys(void) { ...@@ -2216,7 +2224,7 @@ static void findBigKeys(void) {
unsigned long long biggest[5] = {0}, counts[5] = {0}, totalsize[5] = {0}; unsigned long long biggest[5] = {0}, counts[5] = {0}, totalsize[5] = {0};
unsigned long long sampled = 0, total_keys, totlen=0, *sizes=NULL, it=0; unsigned long long sampled = 0, total_keys, totlen=0, *sizes=NULL, it=0;
sds maxkeys[5] = {0}; sds maxkeys[5] = {0};
char *typename[] = {"string","list","set","hash","zset"}; char *typename[] = {"string","list","set","hash","zset","stream"};
char *typeunit[] = {"bytes","items","members","fields","members"}; char *typeunit[] = {"bytes","items","members","fields","members"};
redisReply *reply, *keys; redisReply *reply, *keys;
unsigned int arrsize=0, i; unsigned int arrsize=0, i;
...@@ -2343,6 +2351,129 @@ static void findBigKeys(void) { ...@@ -2343,6 +2351,129 @@ static void findBigKeys(void) {
exit(0); exit(0);
} }
static void getKeyFreqs(redisReply *keys, unsigned long long *freqs) {
redisReply *reply;
unsigned int i;
/* Pipeline OBJECT freq commands */
for(i=0;i<keys->elements;i++) {
redisAppendCommand(context, "OBJECT freq %s", keys->element[i]->str);
}
/* Retrieve freqs */
for(i=0;i<keys->elements;i++) {
if(redisGetReply(context, (void**)&reply)!=REDIS_OK) {
fprintf(stderr, "Error getting freq for key '%s' (%d: %s)\n",
keys->element[i]->str, context->err, context->errstr);
exit(1);
} else if(reply->type != REDIS_REPLY_INTEGER) {
if(reply->type == REDIS_REPLY_ERROR) {
fprintf(stderr, "Error: %s\n", reply->str);
exit(1);
} else {
fprintf(stderr, "Warning: OBJECT freq on '%s' failed (may have been deleted)\n", keys->element[i]->str);
freqs[i] = 0;
}
} else {
freqs[i] = reply->integer;
}
freeReplyObject(reply);
}
}
#define HOTKEYS_SAMPLE 16
static void findHotKeys(void) {
redisReply *keys, *reply;
unsigned long long counters[HOTKEYS_SAMPLE] = {0};
sds hotkeys[HOTKEYS_SAMPLE] = {NULL};
unsigned long long sampled = 0, total_keys, *freqs = NULL, it = 0;
unsigned int arrsize = 0, i, k;
double pct;
/* Total keys pre scanning */
total_keys = getDbSize();
/* Status message */
printf("\n# Scanning the entire keyspace to find hot keys as well as\n");
printf("# average sizes per key type. You can use -i 0.1 to sleep 0.1 sec\n");
printf("# per 100 SCAN commands (not usually needed).\n\n");
/* SCAN loop */
do {
/* Calculate approximate percentage completion */
pct = 100 * (double)sampled/total_keys;
/* Grab some keys and point to the keys array */
reply = sendScan(&it);
keys = reply->element[1];
/* Reallocate our freqs array if we need to */
if(keys->elements > arrsize) {
freqs = zrealloc(freqs, sizeof(unsigned long long)*keys->elements);
if(!freqs) {
fprintf(stderr, "Failed to allocate storage for keys!\n");
exit(1);
}
arrsize = keys->elements;
}
getKeyFreqs(keys, freqs);
/* Now update our stats */
for(i=0;i<keys->elements;i++) {
sampled++;
/* Update overall progress */
if(sampled % 1000000 == 0) {
printf("[%05.2f%%] Sampled %llu keys so far\n", pct, sampled);
}
/* Use eviction pool here */
k = 0;
while (k < HOTKEYS_SAMPLE && freqs[i] > counters[k]) k++;
if (k == 0) continue;
k--;
if (k == 0 || counters[k] == 0) {
sdsfree(hotkeys[k]);
} else {
sdsfree(hotkeys[0]);
memmove(counters,counters+1,sizeof(counters[0])*k);
memmove(hotkeys,hotkeys+1,sizeof(hotkeys[0])*k);
}
counters[k] = freqs[i];
hotkeys[k] = sdsnew(keys->element[i]->str);
printf(
"[%05.2f%%] Hot key '%s' found so far with counter %llu\n",
pct, keys->element[i]->str, freqs[i]);
}
/* Sleep if we've been directed to do so */
if(sampled && (sampled %100) == 0 && config.interval) {
usleep(config.interval);
}
freeReplyObject(reply);
} while(it != 0);
if (freqs) zfree(freqs);
/* We're done */
printf("\n-------- summary -------\n\n");
printf("Sampled %llu keys in the keyspace!\n", sampled);
for (i=1; i<= HOTKEYS_SAMPLE; i++) {
k = HOTKEYS_SAMPLE - i;
if(counters[k]>0) {
printf("hot key found with counter: %llu\tkeyname: %s\n", counters[k], hotkeys[k]);
sdsfree(hotkeys[k]);
}
}
exit(0);
}
/*------------------------------------------------------------------------------ /*------------------------------------------------------------------------------
* Stats mode * Stats mode
*--------------------------------------------------------------------------- */ *--------------------------------------------------------------------------- */
...@@ -2453,7 +2584,7 @@ static void statMode(void) { ...@@ -2453,7 +2584,7 @@ static void statMode(void) {
sprintf(buf,"%ld",aux); sprintf(buf,"%ld",aux);
printf("%-8s",buf); printf("%-8s",buf);
/* Requets */ /* Requests */
aux = getLongInfoField(reply->str,"total_commands_processed"); aux = getLongInfoField(reply->str,"total_commands_processed");
sprintf(buf,"%ld (+%ld)",aux,requests == 0 ? 0 : aux-requests); sprintf(buf,"%ld (+%ld)",aux,requests == 0 ? 0 : aux-requests);
printf("%-19s",buf); printf("%-19s",buf);
...@@ -2720,6 +2851,7 @@ int main(int argc, char **argv) { ...@@ -2720,6 +2851,7 @@ int main(int argc, char **argv) {
config.pipe_mode = 0; config.pipe_mode = 0;
config.pipe_timeout = REDIS_CLI_DEFAULT_PIPE_TIMEOUT; config.pipe_timeout = REDIS_CLI_DEFAULT_PIPE_TIMEOUT;
config.bigkeys = 0; config.bigkeys = 0;
config.hotkeys = 0;
config.stdinarg = 0; config.stdinarg = 0;
config.auth = NULL; config.auth = NULL;
config.eval = NULL; config.eval = NULL;
...@@ -2780,6 +2912,12 @@ int main(int argc, char **argv) { ...@@ -2780,6 +2912,12 @@ int main(int argc, char **argv) {
findBigKeys(); findBigKeys();
} }
/* Find hot keys */
if (config.hotkeys) {
if (cliConnect(0) == REDIS_ERR) exit(1);
findHotKeys();
}
/* Stat mode */ /* Stat mode */
if (config.stat_mode) { if (config.stat_mode) {
if (cliConnect(0) == REDIS_ERR) exit(1); if (cliConnect(0) == REDIS_ERR) exit(1);
......
...@@ -2205,7 +2205,7 @@ void replicationResurrectCachedMaster(int newfd) { ...@@ -2205,7 +2205,7 @@ void replicationResurrectCachedMaster(int newfd) {
server.repl_state = REPL_STATE_CONNECTED; server.repl_state = REPL_STATE_CONNECTED;
/* Re-add to the list of clients. */ /* Re-add to the list of clients. */
listAddNodeTail(server.clients,server.master); linkClient(server.master);
if (aeCreateFileEvent(server.el, newfd, AE_READABLE, if (aeCreateFileEvent(server.el, newfd, AE_READABLE,
readQueryFromClient, server.master)) { readQueryFromClient, server.master)) {
serverLog(LL_WARNING,"Error resurrecting the cached master, impossible to add the readable handler: %s", strerror(errno)); serverLog(LL_WARNING,"Error resurrecting the cached master, impossible to add the readable handler: %s", strerror(errno));
......
...@@ -1141,18 +1141,38 @@ int redis_math_randomseed (lua_State *L) { ...@@ -1141,18 +1141,38 @@ int redis_math_randomseed (lua_State *L) {
* EVAL and SCRIPT commands implementation * EVAL and SCRIPT commands implementation
* ------------------------------------------------------------------------- */ * ------------------------------------------------------------------------- */
/* Define a lua function with the specified function name and body. /* Define a Lua function with the specified body.
* The function name musts be a 42 characters long string, since all the * The function name will be generated in the following form:
* functions we defined in the Lua context are in the form:
* *
* f_<hex sha1 sum> * f_<hex sha1 sum>
* *
* On success C_OK is returned, and nothing is left on the Lua stack. * The function increments the reference count of the 'body' object as a
* On error C_ERR is returned and an appropriate error is set in the * side effect of a successful call.
* client context. */ *
int luaCreateFunction(client *c, lua_State *lua, char *funcname, robj *body) { * On success a pointer to an SDS string representing the function SHA1 of the
sds funcdef = sdsempty(); * just added function is returned (and will be valid until the next call
* to scriptingReset() function), otherwise NULL is returned.
*
* The function handles the fact of being called with a script that already
* exists, and in such a case, it behaves like in the success case.
*
* If 'c' is not NULL, on error the client is informed with an appropriate
* error describing the nature of the problem and the Lua interpreter error. */
sds luaCreateFunction(client *c, lua_State *lua, robj *body) {
char funcname[43];
dictEntry *de;
funcname[0] = 'f';
funcname[1] = '_';
sha1hex(funcname+2,body->ptr,sdslen(body->ptr));
sds sha = sdsnewlen(funcname+2,40);
if ((de = dictFind(server.lua_scripts,sha)) != NULL) {
sdsfree(sha);
return dictGetKey(de);
}
sds funcdef = sdsempty();
funcdef = sdscat(funcdef,"function "); funcdef = sdscat(funcdef,"function ");
funcdef = sdscatlen(funcdef,funcname,42); funcdef = sdscatlen(funcdef,funcname,42);
funcdef = sdscatlen(funcdef,"() ",3); funcdef = sdscatlen(funcdef,"() ",3);
...@@ -1160,30 +1180,35 @@ int luaCreateFunction(client *c, lua_State *lua, char *funcname, robj *body) { ...@@ -1160,30 +1180,35 @@ int luaCreateFunction(client *c, lua_State *lua, char *funcname, robj *body) {
funcdef = sdscatlen(funcdef,"\nend",4); funcdef = sdscatlen(funcdef,"\nend",4);
if (luaL_loadbuffer(lua,funcdef,sdslen(funcdef),"@user_script")) { if (luaL_loadbuffer(lua,funcdef,sdslen(funcdef),"@user_script")) {
addReplyErrorFormat(c,"Error compiling script (new function): %s\n", if (c != NULL) {
addReplyErrorFormat(c,
"Error compiling script (new function): %s\n",
lua_tostring(lua,-1)); lua_tostring(lua,-1));
}
lua_pop(lua,1); lua_pop(lua,1);
sdsfree(sha);
sdsfree(funcdef); sdsfree(funcdef);
return C_ERR; return NULL;
} }
sdsfree(funcdef); sdsfree(funcdef);
if (lua_pcall(lua,0,0,0)) { if (lua_pcall(lua,0,0,0)) {
if (c != NULL) {
addReplyErrorFormat(c,"Error running script (new function): %s\n", addReplyErrorFormat(c,"Error running script (new function): %s\n",
lua_tostring(lua,-1)); lua_tostring(lua,-1));
}
lua_pop(lua,1); lua_pop(lua,1);
return C_ERR; sdsfree(sha);
return NULL;
} }
/* We also save a SHA1 -> Original script map in a dictionary /* We also save a SHA1 -> Original script map in a dictionary
* so that we can replicate / write in the AOF all the * so that we can replicate / write in the AOF all the
* EVALSHA commands as EVAL using the original script. */ * EVALSHA commands as EVAL using the original script. */
{ int retval = dictAdd(server.lua_scripts,sha,body);
int retval = dictAdd(server.lua_scripts, serverAssertWithInfo(c ? c : server.lua_client,NULL,retval == DICT_OK);
sdsnewlen(funcname+2,40),body);
serverAssertWithInfo(c,NULL,retval == DICT_OK);
incrRefCount(body); incrRefCount(body);
} return sha;
return C_OK;
} }
/* This is the Lua script "count" hook that we use to detect scripts timeout. */ /* This is the Lua script "count" hook that we use to detect scripts timeout. */
...@@ -1282,10 +1307,10 @@ void evalGenericCommand(client *c, int evalsha) { ...@@ -1282,10 +1307,10 @@ void evalGenericCommand(client *c, int evalsha) {
addReply(c, shared.noscripterr); addReply(c, shared.noscripterr);
return; return;
} }
if (luaCreateFunction(c,lua,funcname,c->argv[1]) == C_ERR) { if (luaCreateFunction(c,lua,c->argv[1]) == NULL) {
lua_pop(lua,1); /* remove the error handler from the stack. */ lua_pop(lua,1); /* remove the error handler from the stack. */
/* The error is sent to the client by luaCreateFunction() /* The error is sent to the client by luaCreateFunction()
* itself when it returns C_ERR. */ * itself when it returns NULL. */
return; return;
} }
/* Now the following is guaranteed to return non nil */ /* Now the following is guaranteed to return non nil */
...@@ -1456,22 +1481,9 @@ void scriptCommand(client *c) { ...@@ -1456,22 +1481,9 @@ void scriptCommand(client *c) {
addReply(c,shared.czero); addReply(c,shared.czero);
} }
} else if (c->argc == 3 && !strcasecmp(c->argv[1]->ptr,"load")) { } else if (c->argc == 3 && !strcasecmp(c->argv[1]->ptr,"load")) {
char funcname[43]; sds sha = luaCreateFunction(c,server.lua,c->argv[2]);
sds sha; if (sha == NULL) return; /* The error was sent by luaCreateFunction(). */
addReplyBulkCBuffer(c,sha,40);
funcname[0] = 'f';
funcname[1] = '_';
sha1hex(funcname+2,c->argv[2]->ptr,sdslen(c->argv[2]->ptr));
sha = sdsnewlen(funcname+2,40);
if (dictFind(server.lua_scripts,sha) == NULL) {
if (luaCreateFunction(c,server.lua,funcname,c->argv[2])
== C_ERR) {
sdsfree(sha);
return;
}
}
addReplyBulkCBuffer(c,funcname+2,40);
sdsfree(sha);
forceCommandPropagation(c,PROPAGATE_REPL|PROPAGATE_AOF); forceCommandPropagation(c,PROPAGATE_REPL|PROPAGATE_AOF);
} else if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"kill")) { } else if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"kill")) {
if (server.lua_caller == NULL) { if (server.lua_caller == NULL) {
......
...@@ -258,7 +258,7 @@ struct redisCommand redisCommandTable[] = { ...@@ -258,7 +258,7 @@ struct redisCommand redisCommandTable[] = {
{"persist",persistCommand,2,"wF",0,NULL,1,1,1,0,0}, {"persist",persistCommand,2,"wF",0,NULL,1,1,1,0,0},
{"slaveof",slaveofCommand,3,"ast",0,NULL,0,0,0,0,0}, {"slaveof",slaveofCommand,3,"ast",0,NULL,0,0,0,0,0},
{"role",roleCommand,1,"lst",0,NULL,0,0,0,0,0}, {"role",roleCommand,1,"lst",0,NULL,0,0,0,0,0},
{"debug",debugCommand,-1,"as",0,NULL,0,0,0,0,0}, {"debug",debugCommand,-2,"as",0,NULL,0,0,0,0,0},
{"config",configCommand,-2,"lat",0,NULL,0,0,0,0,0}, {"config",configCommand,-2,"lat",0,NULL,0,0,0,0,0},
{"subscribe",subscribeCommand,-2,"pslt",0,NULL,0,0,0,0,0}, {"subscribe",subscribeCommand,-2,"pslt",0,NULL,0,0,0,0,0},
{"unsubscribe",unsubscribeCommand,-1,"pslt",0,NULL,0,0,0,0,0}, {"unsubscribe",unsubscribeCommand,-1,"pslt",0,NULL,0,0,0,0,0},
...@@ -302,6 +302,11 @@ struct redisCommand redisCommandTable[] = { ...@@ -302,6 +302,11 @@ struct redisCommand redisCommandTable[] = {
{"pfcount",pfcountCommand,-2,"r",0,NULL,1,-1,1,0,0}, {"pfcount",pfcountCommand,-2,"r",0,NULL,1,-1,1,0,0},
{"pfmerge",pfmergeCommand,-2,"wm",0,NULL,1,-1,1,0,0}, {"pfmerge",pfmergeCommand,-2,"wm",0,NULL,1,-1,1,0,0},
{"pfdebug",pfdebugCommand,-3,"w",0,NULL,0,0,0,0,0}, {"pfdebug",pfdebugCommand,-3,"w",0,NULL,0,0,0,0,0},
{"xadd",xaddCommand,-5,"wmF",0,NULL,1,1,1,0,0},
{"xrange",xrangeCommand,-4,"r",0,NULL,1,1,1,0,0},
{"xrevrange",xrevrangeCommand,-4,"r",0,NULL,1,1,1,0,0},
{"xlen",xlenCommand,2,"rF",0,NULL,1,1,1,0,0},
{"xread",xreadCommand,-3,"rs",0,xreadGetKeys,1,1,1,0,0},
{"post",securityWarningCommand,-1,"lt",0,NULL,0,0,0,0,0}, {"post",securityWarningCommand,-1,"lt",0,NULL,0,0,0,0,0},
{"host:",securityWarningCommand,-1,"lt",0,NULL,0,0,0,0,0}, {"host:",securityWarningCommand,-1,"lt",0,NULL,0,0,0,0,0},
{"latency",latencyCommand,-2,"aslt",0,NULL,0,0,0,0,0} {"latency",latencyCommand,-2,"aslt",0,NULL,0,0,0,0,0}
...@@ -551,6 +556,17 @@ dictType objectKeyPointerValueDictType = { ...@@ -551,6 +556,17 @@ dictType objectKeyPointerValueDictType = {
NULL /* val destructor */ NULL /* val destructor */
}; };
/* Like objectKeyPointerValueDictType(), but values can be destroyed, if
* not NULL, calling zfree(). */
dictType objectKeyHeapPointerValueDictType = {
dictEncObjHash, /* hash function */
NULL, /* key dup */
NULL, /* val dup */
dictEncObjKeyCompare, /* key compare */
dictObjectDestructor, /* key destructor */
dictVanillaFree /* val destructor */
};
/* Set dictionary type. Keys are SDS strings, values are ot used. */ /* Set dictionary type. Keys are SDS strings, values are ot used. */
dictType setDictType = { dictType setDictType = {
dictSdsHash, /* hash function */ dictSdsHash, /* hash function */
...@@ -1411,7 +1427,9 @@ void initServerConfig(void) { ...@@ -1411,7 +1427,9 @@ void initServerConfig(void) {
server.active_defrag_running = 0; server.active_defrag_running = 0;
server.notify_keyspace_events = 0; server.notify_keyspace_events = 0;
server.maxclients = CONFIG_DEFAULT_MAX_CLIENTS; server.maxclients = CONFIG_DEFAULT_MAX_CLIENTS;
server.bpop_blocked_clients = 0; server.blocked_clients = 0;
memset(server.blocked_clients_by_type,0,
sizeof(server.blocked_clients_by_type));
server.maxmemory = CONFIG_DEFAULT_MAXMEMORY; server.maxmemory = CONFIG_DEFAULT_MAXMEMORY;
server.maxmemory_policy = CONFIG_DEFAULT_MAXMEMORY_POLICY; server.maxmemory_policy = CONFIG_DEFAULT_MAXMEMORY_POLICY;
server.maxmemory_samples = CONFIG_DEFAULT_MAXMEMORY_SAMPLES; server.maxmemory_samples = CONFIG_DEFAULT_MAXMEMORY_SAMPLES;
...@@ -1549,16 +1567,29 @@ int restartServer(int flags, mstime_t delay) { ...@@ -1549,16 +1567,29 @@ int restartServer(int flags, mstime_t delay) {
/* Check if we still have accesses to the executable that started this /* Check if we still have accesses to the executable that started this
* server instance. */ * server instance. */
if (access(server.executable,X_OK) == -1) return C_ERR; if (access(server.executable,X_OK) == -1) {
serverLog(LL_WARNING,"Can't restart: this process has no "
"permissions to execute %s", server.executable);
return C_ERR;
}
/* Config rewriting. */ /* Config rewriting. */
if (flags & RESTART_SERVER_CONFIG_REWRITE && if (flags & RESTART_SERVER_CONFIG_REWRITE &&
server.configfile && server.configfile &&
rewriteConfig(server.configfile) == -1) return C_ERR; rewriteConfig(server.configfile) == -1)
{
serverLog(LL_WARNING,"Can't restart: configuration rewrite process "
"failed");
return C_ERR;
}
/* Perform a proper shutdown. */ /* Perform a proper shutdown. */
if (flags & RESTART_SERVER_GRACEFULLY && if (flags & RESTART_SERVER_GRACEFULLY &&
prepareForShutdown(SHUTDOWN_NOFLAGS) != C_OK) return C_ERR; prepareForShutdown(SHUTDOWN_NOFLAGS) != C_OK)
{
serverLog(LL_WARNING,"Can't restart: error preparing for shutdown");
return C_ERR;
}
/* Close all file descriptors, with the exception of stdin, stdout, strerr /* Close all file descriptors, with the exception of stdin, stdout, strerr
* which are useful if we restart a Redis server which is not daemonized. */ * which are useful if we restart a Redis server which is not daemonized. */
...@@ -1570,6 +1601,8 @@ int restartServer(int flags, mstime_t delay) { ...@@ -1570,6 +1601,8 @@ int restartServer(int flags, mstime_t delay) {
/* Execute the server with the original command line. */ /* Execute the server with the original command line. */
if (delay) usleep(delay*1000); if (delay) usleep(delay*1000);
zfree(server.exec_argv[0]);
server.exec_argv[0] = zstrdup(server.executable);
execve(server.executable,server.exec_argv,environ); execve(server.executable,server.exec_argv,environ);
/* If an error occurred here, there is nothing we can do, but exit. */ /* If an error occurred here, there is nothing we can do, but exit. */
...@@ -2445,8 +2478,9 @@ int processCommand(client *c) { ...@@ -2445,8 +2478,9 @@ int processCommand(client *c) {
return C_OK; return C_OK;
} }
/* Only allow INFO and SLAVEOF when slave-serve-stale-data is no and /* Only allow commands with flag "t", such as INFO, SLAVEOF and so on,
* we are a slave with a broken link with master. */ * when slave-serve-stale-data is no and we are a slave with a broken
* link with master. */
if (server.masterhost && server.repl_state != REPL_STATE_CONNECTED && if (server.masterhost && server.repl_state != REPL_STATE_CONNECTED &&
server.repl_serve_stale_data == 0 && server.repl_serve_stale_data == 0 &&
!(c->cmd->flags & CMD_STALE)) !(c->cmd->flags & CMD_STALE))
...@@ -2490,7 +2524,7 @@ int processCommand(client *c) { ...@@ -2490,7 +2524,7 @@ int processCommand(client *c) {
call(c,CMD_CALL_FULL); call(c,CMD_CALL_FULL);
c->woff = server.master_repl_offset; c->woff = server.master_repl_offset;
if (listLength(server.ready_keys)) if (listLength(server.ready_keys))
handleClientsBlockedOnLists(); handleClientsBlockedOnKeys();
} }
return C_OK; return C_OK;
} }
...@@ -2909,7 +2943,7 @@ sds genRedisInfoString(char *section) { ...@@ -2909,7 +2943,7 @@ sds genRedisInfoString(char *section) {
"blocked_clients:%d\r\n", "blocked_clients:%d\r\n",
listLength(server.clients)-listLength(server.slaves), listLength(server.clients)-listLength(server.slaves),
lol, bib, lol, bib,
server.bpop_blocked_clients); server.blocked_clients);
} }
/* Memory */ /* Memory */
......
...@@ -59,6 +59,7 @@ typedef long long mstime_t; /* millisecond time type. */ ...@@ -59,6 +59,7 @@ typedef long long mstime_t; /* millisecond time type. */
#include "anet.h" /* Networking the easy way */ #include "anet.h" /* Networking the easy way */
#include "ziplist.h" /* Compact list data structure */ #include "ziplist.h" /* Compact list data structure */
#include "intset.h" /* Compact integer set structure */ #include "intset.h" /* Compact integer set structure */
#include "stream.h" /* Stream data type header file. */
#include "version.h" /* Version macro */ #include "version.h" /* Version macro */
#include "util.h" /* Misc functions useful in many places */ #include "util.h" /* Misc functions useful in many places */
#include "latency.h" /* Latency monitor API */ #include "latency.h" /* Latency monitor API */
...@@ -255,6 +256,8 @@ typedef long long mstime_t; /* millisecond time type. */ ...@@ -255,6 +256,8 @@ typedef long long mstime_t; /* millisecond time type. */
#define BLOCKED_LIST 1 /* BLPOP & co. */ #define BLOCKED_LIST 1 /* BLPOP & co. */
#define BLOCKED_WAIT 2 /* WAIT for synchronous replication. */ #define BLOCKED_WAIT 2 /* WAIT for synchronous replication. */
#define BLOCKED_MODULE 3 /* Blocked by a loadable module. */ #define BLOCKED_MODULE 3 /* Blocked by a loadable module. */
#define BLOCKED_STREAM 4 /* XREAD. */
#define BLOCKED_NUM 5 /* Number of blocked states. */
/* Client request types */ /* Client request types */
#define PROTO_REQ_INLINE 1 #define PROTO_REQ_INLINE 1
...@@ -424,7 +427,8 @@ typedef long long mstime_t; /* millisecond time type. */ ...@@ -424,7 +427,8 @@ typedef long long mstime_t; /* millisecond time type. */
#define NOTIFY_ZSET (1<<7) /* z */ #define NOTIFY_ZSET (1<<7) /* z */
#define NOTIFY_EXPIRED (1<<8) /* x */ #define NOTIFY_EXPIRED (1<<8) /* x */
#define NOTIFY_EVICTED (1<<9) /* e */ #define NOTIFY_EVICTED (1<<9) /* e */
#define NOTIFY_ALL (NOTIFY_GENERIC | NOTIFY_STRING | NOTIFY_LIST | NOTIFY_SET | NOTIFY_HASH | NOTIFY_ZSET | NOTIFY_EXPIRED | NOTIFY_EVICTED) /* A */ #define NOTIFY_STREAM (1<<10) /* t */
#define NOTIFY_ALL (NOTIFY_GENERIC | NOTIFY_STRING | NOTIFY_LIST | NOTIFY_SET | NOTIFY_HASH | NOTIFY_ZSET | NOTIFY_EXPIRED | NOTIFY_EVICTED | NOTIFY_STREAM) /* A flag */
/* Get the first bind addr or NULL */ /* Get the first bind addr or NULL */
#define NET_FIRST_BIND_ADDR (server.bindaddr_count ? server.bindaddr[0] : NULL) #define NET_FIRST_BIND_ADDR (server.bindaddr_count ? server.bindaddr[0] : NULL)
...@@ -446,11 +450,11 @@ typedef long long mstime_t; /* millisecond time type. */ ...@@ -446,11 +450,11 @@ typedef long long mstime_t; /* millisecond time type. */
/* A redis object, that is a type able to hold a string / list / set */ /* A redis object, that is a type able to hold a string / list / set */
/* The actual Redis Object */ /* The actual Redis Object */
#define OBJ_STRING 0 #define OBJ_STRING 0 /* String object. */
#define OBJ_LIST 1 #define OBJ_LIST 1 /* List object. */
#define OBJ_SET 2 #define OBJ_SET 2 /* Set object. */
#define OBJ_ZSET 3 #define OBJ_ZSET 3 /* Sorted set object. */
#define OBJ_HASH 4 #define OBJ_HASH 4 /* Hash object. */
/* The "module" object type is a special one that signals that the object /* The "module" object type is a special one that signals that the object
* is one directly managed by a Redis module. In this case the value points * is one directly managed by a Redis module. In this case the value points
...@@ -463,7 +467,8 @@ typedef long long mstime_t; /* millisecond time type. */ ...@@ -463,7 +467,8 @@ typedef long long mstime_t; /* millisecond time type. */
* by a 64 bit module type ID, which has a 54 bits module-specific signature * by a 64 bit module type ID, which has a 54 bits module-specific signature
* in order to dispatch the loading to the right module, plus a 10 bits * in order to dispatch the loading to the right module, plus a 10 bits
* encoding version. */ * encoding version. */
#define OBJ_MODULE 5 #define OBJ_MODULE 5 /* Module object. */
#define OBJ_STREAM 6 /* Stream object. */
/* Extract encver / signature from a module type ID. */ /* Extract encver / signature from a module type ID. */
#define REDISMODULE_TYPE_ENCVER_BITS 10 #define REDISMODULE_TYPE_ENCVER_BITS 10
...@@ -575,6 +580,7 @@ typedef struct RedisModuleDigest { ...@@ -575,6 +580,7 @@ typedef struct RedisModuleDigest {
#define OBJ_ENCODING_SKIPLIST 7 /* Encoded as skiplist */ #define OBJ_ENCODING_SKIPLIST 7 /* Encoded as skiplist */
#define OBJ_ENCODING_EMBSTR 8 /* Embedded sds string encoding */ #define OBJ_ENCODING_EMBSTR 8 /* Embedded sds string encoding */
#define OBJ_ENCODING_QUICKLIST 9 /* Encoded as linked list of ziplists */ #define OBJ_ENCODING_QUICKLIST 9 /* Encoded as linked list of ziplists */
#define OBJ_ENCODING_STREAM 10 /* Encoded as a radix tree of listpacks */
#define LRU_BITS 24 #define LRU_BITS 24
#define LRU_CLOCK_MAX ((1<<LRU_BITS)-1) /* Max value of obj->lru */ #define LRU_CLOCK_MAX ((1<<LRU_BITS)-1) /* Max value of obj->lru */
...@@ -586,7 +592,7 @@ typedef struct redisObject { ...@@ -586,7 +592,7 @@ typedef struct redisObject {
unsigned encoding:4; unsigned encoding:4;
unsigned lru:LRU_BITS; /* LRU time (relative to global lru_clock) or unsigned lru:LRU_BITS; /* LRU time (relative to global lru_clock) or
* LFU data (least significant 8 bits frequency * LFU data (least significant 8 bits frequency
* and most significant 16 bits decreas time). */ * and most significant 16 bits access time). */
int refcount; int refcount;
void *ptr; void *ptr;
} robj; } robj;
...@@ -638,12 +644,17 @@ typedef struct blockingState { ...@@ -638,12 +644,17 @@ typedef struct blockingState {
mstime_t timeout; /* Blocking operation timeout. If UNIX current time mstime_t timeout; /* Blocking operation timeout. If UNIX current time
* is > timeout then the operation timed out. */ * is > timeout then the operation timed out. */
/* BLOCKED_LIST */ /* BLOCKED_LIST and BLOCKED_STREAM */
dict *keys; /* The keys we are waiting to terminate a blocking dict *keys; /* The keys we are waiting to terminate a blocking
* operation such as BLPOP. Otherwise NULL. */ * operation such as BLPOP or XREAD. Or NULL. */
robj *target; /* The key that should receive the element, robj *target; /* The key that should receive the element,
* for BRPOPLPUSH. */ * for BRPOPLPUSH. */
/* BLOCK_STREAM */
size_t xread_count; /* XREAD COUNT option. */
robj *xread_group; /* XREAD group name. */
mstime_t xread_retry_time, xread_retry_ttl;
/* BLOCKED_WAIT */ /* BLOCKED_WAIT */
int numreplicas; /* Number of replicas we are waiting for ACK. */ int numreplicas; /* Number of replicas we are waiting for ACK. */
long long reploffset; /* Replication offset to reach. */ long long reploffset; /* Replication offset to reach. */
...@@ -722,6 +733,7 @@ typedef struct client { ...@@ -722,6 +733,7 @@ typedef struct client {
dict *pubsub_channels; /* channels a client is interested in (SUBSCRIBE) */ dict *pubsub_channels; /* channels a client is interested in (SUBSCRIBE) */
list *pubsub_patterns; /* patterns a client is interested in (SUBSCRIBE) */ list *pubsub_patterns; /* patterns a client is interested in (SUBSCRIBE) */
sds peerid; /* Cached peer ID. */ sds peerid; /* Cached peer ID. */
listNode *client_list_node; /* list node in client list */
/* Response buffer */ /* Response buffer */
int bufpos; int bufpos;
...@@ -1118,10 +1130,11 @@ struct redisServer { ...@@ -1118,10 +1130,11 @@ struct redisServer {
unsigned long long maxmemory; /* Max number of memory bytes to use */ unsigned long long maxmemory; /* Max number of memory bytes to use */
int maxmemory_policy; /* Policy for key eviction */ int maxmemory_policy; /* Policy for key eviction */
int maxmemory_samples; /* Pricision of random sampling */ int maxmemory_samples; /* Pricision of random sampling */
unsigned int lfu_log_factor; /* LFU logarithmic counter factor. */ int lfu_log_factor; /* LFU logarithmic counter factor. */
unsigned int lfu_decay_time; /* LFU counter decay factor. */ int lfu_decay_time; /* LFU counter decay factor. */
/* Blocked clients */ /* Blocked clients */
unsigned int bpop_blocked_clients; /* Number of clients blocked by lists */ unsigned int blocked_clients; /* # of clients executing a blocking cmd.*/
unsigned int blocked_clients_by_type[BLOCKED_NUM];
list *unblocked_clients; /* list of clients to unblock before next loop */ list *unblocked_clients; /* list of clients to unblock before next loop */
list *ready_keys; /* List of readyList structures for BLPOP & co */ list *ready_keys; /* List of readyList structures for BLPOP & co */
/* Sort parameters - qsort_r() is only available under BSD so we /* Sort parameters - qsort_r() is only available under BSD so we
...@@ -1288,6 +1301,7 @@ typedef struct { ...@@ -1288,6 +1301,7 @@ typedef struct {
extern struct redisServer server; extern struct redisServer server;
extern struct sharedObjectsStruct shared; extern struct sharedObjectsStruct shared;
extern dictType objectKeyPointerValueDictType; extern dictType objectKeyPointerValueDictType;
extern dictType objectKeyHeapPointerValueDictType;
extern dictType setDictType; extern dictType setDictType;
extern dictType zsetDictType; extern dictType zsetDictType;
extern dictType clusterNodesDictType; extern dictType clusterNodesDictType;
...@@ -1386,6 +1400,7 @@ int handleClientsWithPendingWrites(void); ...@@ -1386,6 +1400,7 @@ int handleClientsWithPendingWrites(void);
int clientHasPendingReplies(client *c); int clientHasPendingReplies(client *c);
void unlinkClient(client *c); void unlinkClient(client *c);
int writeToClient(int fd, client *c, int handler_installed); int writeToClient(int fd, client *c, int handler_installed);
void linkClient(client *c);
#ifdef __GNUC__ #ifdef __GNUC__
void addReplyErrorFormat(client *c, const char *fmt, ...) void addReplyErrorFormat(client *c, const char *fmt, ...)
...@@ -1411,9 +1426,7 @@ int listTypeEqual(listTypeEntry *entry, robj *o); ...@@ -1411,9 +1426,7 @@ int listTypeEqual(listTypeEntry *entry, robj *o);
void listTypeDelete(listTypeIterator *iter, listTypeEntry *entry); void listTypeDelete(listTypeIterator *iter, listTypeEntry *entry);
void listTypeConvert(robj *subject, int enc); void listTypeConvert(robj *subject, int enc);
void unblockClientWaitingData(client *c); void unblockClientWaitingData(client *c);
void handleClientsBlockedOnLists(void);
void popGenericCommand(client *c, int where); void popGenericCommand(client *c, int where);
void signalListAsReady(redisDb *db, robj *key);
/* MULTI/EXEC/WATCH... */ /* MULTI/EXEC/WATCH... */
void unwatchAllKeys(client *c); void unwatchAllKeys(client *c);
...@@ -1456,6 +1469,7 @@ robj *createIntsetObject(void); ...@@ -1456,6 +1469,7 @@ robj *createIntsetObject(void);
robj *createHashObject(void); robj *createHashObject(void);
robj *createZsetObject(void); robj *createZsetObject(void);
robj *createZsetZiplistObject(void); robj *createZsetZiplistObject(void);
robj *createStreamObject(void);
robj *createModuleObject(moduleType *mt, void *value); robj *createModuleObject(moduleType *mt, void *value);
int getLongFromObjectOrReply(client *c, robj *o, long *target, const char *msg); int getLongFromObjectOrReply(client *c, robj *o, long *target, const char *msg);
int checkType(client *c, robj *o, int type); int checkType(client *c, robj *o, int type);
...@@ -1755,6 +1769,7 @@ int *evalGetKeys(struct redisCommand *cmd, robj **argv, int argc, int *numkeys); ...@@ -1755,6 +1769,7 @@ int *evalGetKeys(struct redisCommand *cmd, robj **argv, int argc, int *numkeys);
int *sortGetKeys(struct redisCommand *cmd, robj **argv, int argc, int *numkeys); int *sortGetKeys(struct redisCommand *cmd, robj **argv, int argc, int *numkeys);
int *migrateGetKeys(struct redisCommand *cmd, robj **argv, int argc, int *numkeys); int *migrateGetKeys(struct redisCommand *cmd, robj **argv, int argc, int *numkeys);
int *georadiusGetKeys(struct redisCommand *cmd, robj **argv, int argc, int *numkeys); int *georadiusGetKeys(struct redisCommand *cmd, robj **argv, int argc, int *numkeys);
int *xreadGetKeys(struct redisCommand *cmd, robj **argv, int argc, int *numkeys);
/* Cluster */ /* Cluster */
void clusterInit(void); void clusterInit(void);
...@@ -1782,6 +1797,7 @@ void scriptingInit(int setup); ...@@ -1782,6 +1797,7 @@ void scriptingInit(int setup);
int ldbRemoveChild(pid_t pid); int ldbRemoveChild(pid_t pid);
void ldbKillForkedSessions(void); void ldbKillForkedSessions(void);
int ldbPendingChildren(void); int ldbPendingChildren(void);
sds luaCreateFunction(client *c, lua_State *lua, robj *body);
/* Blocked clients */ /* Blocked clients */
void processUnblockedClients(void); void processUnblockedClients(void);
...@@ -1790,6 +1806,9 @@ void unblockClient(client *c); ...@@ -1790,6 +1806,9 @@ void unblockClient(client *c);
void replyToBlockedClientTimedOut(client *c); void replyToBlockedClientTimedOut(client *c);
int getTimeoutFromObjectOrReply(client *c, robj *object, mstime_t *timeout, int unit); int getTimeoutFromObjectOrReply(client *c, robj *object, mstime_t *timeout, int unit);
void disconnectAllBlockedClients(void); void disconnectAllBlockedClients(void);
void handleClientsBlockedOnKeys(void);
void signalKeyAsReady(redisDb *db, robj *key);
void blockForKeys(client *c, int btype, robj **keys, int numkeys, mstime_t timeout, robj *target, streamID *ids);
/* expire.c -- Handling of expired keys */ /* expire.c -- Handling of expired keys */
void activeExpireCycle(int type); void activeExpireCycle(int type);
...@@ -1803,6 +1822,7 @@ void evictionPoolAlloc(void); ...@@ -1803,6 +1822,7 @@ void evictionPoolAlloc(void);
#define LFU_INIT_VAL 5 #define LFU_INIT_VAL 5
unsigned long LFUGetTimeInMinutes(void); unsigned long LFUGetTimeInMinutes(void);
uint8_t LFULogIncr(uint8_t value); uint8_t LFULogIncr(uint8_t value);
unsigned long LFUDecrAndReturn(robj *o);
/* Keys hashing / comparison functions for dict.c hash tables. */ /* Keys hashing / comparison functions for dict.c hash tables. */
uint64_t dictSdsHash(const void *key); uint64_t dictSdsHash(const void *key);
...@@ -1991,6 +2011,11 @@ void pfdebugCommand(client *c); ...@@ -1991,6 +2011,11 @@ void pfdebugCommand(client *c);
void latencyCommand(client *c); void latencyCommand(client *c);
void moduleCommand(client *c); void moduleCommand(client *c);
void securityWarningCommand(client *c); void securityWarningCommand(client *c);
void xaddCommand(client *c);
void xrangeCommand(client *c);
void xrevrangeCommand(client *c);
void xlenCommand(client *c);
void xreadCommand(client *c);
#if defined(__GNUC__) #if defined(__GNUC__)
void *calloc(size_t count, size_t size) __attribute__ ((deprecated)); void *calloc(size_t count, size_t size) __attribute__ ((deprecated));
......
...@@ -39,7 +39,11 @@ ...@@ -39,7 +39,11 @@
#include <errno.h> /* errno program_invocation_name program_invocation_short_name */ #include <errno.h> /* errno program_invocation_name program_invocation_short_name */
#if !defined(HAVE_SETPROCTITLE) #if !defined(HAVE_SETPROCTITLE)
#define HAVE_SETPROCTITLE (defined __NetBSD__ || defined __FreeBSD__ || defined __OpenBSD__) #if (defined __NetBSD__ || defined __FreeBSD__ || defined __OpenBSD__)
#define HAVE_SETPROCTITLE 1
#else
#define HAVE_SETPROCTITLE 0
#endif
#endif #endif
......
#ifndef STREAM_H
#define STREAM_H
#include "rax.h"
#include "listpack.h"
/* Stream item ID: a 128 bit number composed of a milliseconds time and
* a sequence counter. IDs generated in the same millisecond (or in a past
* millisecond if the clock jumped backward) will use the millisecond time
* of the latest generated ID and an incremented sequence. */
typedef struct streamID {
uint64_t ms; /* Unix time in milliseconds. */
uint64_t seq; /* Sequence number. */
} streamID;
typedef struct stream {
rax *rax; /* The radix tree holding the stream. */
uint64_t length; /* Number of elements inside this stream. */
streamID last_id; /* Zero if there are yet no items. */
} stream;
/* We define an iterator to iterate stream items in an abstract way, without
* caring about the radix tree + listpack representation. Technically speaking
* the iterator is only used inside streamReplyWithRange(), so could just
* be implemented inside the function, but practically there is the AOF
* rewriting code that also needs to iterate the stream to emit the XADD
* commands. */
typedef struct streamIterator {
streamID master_id; /* ID of the master entry at listpack head. */
uint64_t master_fields_count; /* Master entries # of fields. */
unsigned char *master_fields_start; /* Master entries start in listpack. */
unsigned char *master_fields_ptr; /* Master field to emit next. */
int entry_flags; /* Flags of entry we are emitting. */
int rev; /* True if iterating end to start (reverse). */
uint64_t start_key[2]; /* Start key as 128 bit big endian. */
uint64_t end_key[2]; /* End key as 128 bit big endian. */
raxIterator ri; /* Rax iterator. */
unsigned char *lp; /* Current listpack. */
unsigned char *lp_ele; /* Current listpack cursor. */
/* Buffers used to hold the string of lpGet() when the element is
* integer encoded, so that there is no string representation of the
* element inside the listpack itself. */
unsigned char field_buf[LP_INTBUF_SIZE];
unsigned char value_buf[LP_INTBUF_SIZE];
} streamIterator;
/* Prototypes of exported APIs. */
struct client;
stream *streamNew(void);
void freeStream(stream *s);
size_t streamReplyWithRange(struct client *c, stream *s, streamID *start, streamID *end, size_t count, int rev);
void streamIteratorStart(streamIterator *si, stream *s, streamID *start, streamID *end, int rev);
int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields);
void streamIteratorGetField(streamIterator *si, unsigned char **fieldptr, unsigned char **valueptr, int64_t *fieldlen, int64_t *valuelen);
void streamIteratorStop(streamIterator *si);
#endif
...@@ -287,8 +287,8 @@ int hashTypeDelete(robj *o, sds field) { ...@@ -287,8 +287,8 @@ int hashTypeDelete(robj *o, sds field) {
if (fptr != NULL) { if (fptr != NULL) {
fptr = ziplistFind(fptr, (unsigned char*)field, sdslen(field), 1); fptr = ziplistFind(fptr, (unsigned char*)field, sdslen(field), 1);
if (fptr != NULL) { if (fptr != NULL) {
zl = ziplistDelete(zl,&fptr); zl = ziplistDelete(zl,&fptr); /* Delete the key. */
zl = ziplistDelete(zl,&fptr); zl = ziplistDelete(zl,&fptr); /* Delete the value. */
o->ptr = zl; o->ptr = zl;
deleted = 1; deleted = 1;
} }
......
...@@ -603,119 +603,6 @@ void rpoplpushCommand(client *c) { ...@@ -603,119 +603,6 @@ void rpoplpushCommand(client *c) {
* Blocking POP operations * Blocking POP operations
*----------------------------------------------------------------------------*/ *----------------------------------------------------------------------------*/
/* This is how the current blocking POP works, we use BLPOP as example:
* - If the user calls BLPOP and the key exists and contains a non empty list
* then LPOP is called instead. So BLPOP is semantically the same as LPOP
* if blocking is not required.
* - If instead BLPOP is called and the key does not exists or the list is
* empty we need to block. In order to do so we remove the notification for
* new data to read in the client socket (so that we'll not serve new
* requests if the blocking request is not served). Also we put the client
* in a dictionary (db->blocking_keys) mapping keys to a list of clients
* blocking for this keys.
* - If a PUSH operation against a key with blocked clients waiting is
* performed, we mark this key as "ready", and after the current command,
* MULTI/EXEC block, or script, is executed, we serve all the clients waiting
* for this list, from the one that blocked first, to the last, accordingly
* to the number of elements we have in the ready list.
*/
/* Set a client in blocking mode for the specified key, with the specified
* timeout */
void blockForKeys(client *c, robj **keys, int numkeys, mstime_t timeout, robj *target) {
dictEntry *de;
list *l;
int j;
c->bpop.timeout = timeout;
c->bpop.target = target;
if (target != NULL) incrRefCount(target);
for (j = 0; j < numkeys; j++) {
/* If the key already exists in the dict ignore it. */
if (dictAdd(c->bpop.keys,keys[j],NULL) != DICT_OK) continue;
incrRefCount(keys[j]);
/* And in the other "side", to map keys -> clients */
de = dictFind(c->db->blocking_keys,keys[j]);
if (de == NULL) {
int retval;
/* For every key we take a list of clients blocked for it */
l = listCreate();
retval = dictAdd(c->db->blocking_keys,keys[j],l);
incrRefCount(keys[j]);
serverAssertWithInfo(c,keys[j],retval == DICT_OK);
} else {
l = dictGetVal(de);
}
listAddNodeTail(l,c);
}
blockClient(c,BLOCKED_LIST);
}
/* Unblock a client that's waiting in a blocking operation such as BLPOP.
* You should never call this function directly, but unblockClient() instead. */
void unblockClientWaitingData(client *c) {
dictEntry *de;
dictIterator *di;
list *l;
serverAssertWithInfo(c,NULL,dictSize(c->bpop.keys) != 0);
di = dictGetIterator(c->bpop.keys);
/* The client may wait for multiple keys, so unblock it for every key. */
while((de = dictNext(di)) != NULL) {
robj *key = dictGetKey(de);
/* Remove this client from the list of clients waiting for this key. */
l = dictFetchValue(c->db->blocking_keys,key);
serverAssertWithInfo(c,key,l != NULL);
listDelNode(l,listSearchKey(l,c));
/* If the list is empty we need to remove it to avoid wasting memory */
if (listLength(l) == 0)
dictDelete(c->db->blocking_keys,key);
}
dictReleaseIterator(di);
/* Cleanup the client structure */
dictEmpty(c->bpop.keys,NULL);
if (c->bpop.target) {
decrRefCount(c->bpop.target);
c->bpop.target = NULL;
}
}
/* If the specified key has clients blocked waiting for list pushes, this
* function will put the key reference into the server.ready_keys list.
* Note that db->ready_keys is a hash table that allows us to avoid putting
* the same key again and again in the list in case of multiple pushes
* made by a script or in the context of MULTI/EXEC.
*
* The list will be finally processed by handleClientsBlockedOnLists() */
void signalListAsReady(redisDb *db, robj *key) {
readyList *rl;
/* No clients blocking for this key? No need to queue it. */
if (dictFind(db->blocking_keys,key) == NULL) return;
/* Key was already signaled? No need to queue it again. */
if (dictFind(db->ready_keys,key) != NULL) return;
/* Ok, we need to queue this key into server.ready_keys. */
rl = zmalloc(sizeof(*rl));
rl->key = key;
rl->db = db;
incrRefCount(key);
listAddNodeTail(server.ready_keys,rl);
/* We also add the key in the db->ready_keys dictionary in order
* to avoid adding it multiple times into a list with a simple O(1)
* check. */
incrRefCount(key);
serverAssert(dictAdd(db->ready_keys,key,NULL) == DICT_OK);
}
/* This is a helper function for handleClientsBlockedOnLists(). It's work /* This is a helper function for handleClientsBlockedOnLists(). It's work
* is to serve a specific client (receiver) that is blocked on 'key' * is to serve a specific client (receiver) that is blocked on 'key'
* in the context of the specified 'db', doing the following: * in the context of the specified 'db', doing the following:
...@@ -785,97 +672,6 @@ int serveClientBlockedOnList(client *receiver, robj *key, robj *dstkey, redisDb ...@@ -785,97 +672,6 @@ int serveClientBlockedOnList(client *receiver, robj *key, robj *dstkey, redisDb
return C_OK; return C_OK;
} }
/* This function should be called by Redis every time a single command,
* a MULTI/EXEC block, or a Lua script, terminated its execution after
* being called by a client.
*
* All the keys with at least one client blocked that received at least
* one new element via some PUSH operation are accumulated into
* the server.ready_keys list. This function will run the list and will
* serve clients accordingly. Note that the function will iterate again and
* again as a result of serving BRPOPLPUSH we can have new blocking clients
* to serve because of the PUSH side of BRPOPLPUSH. */
void handleClientsBlockedOnLists(void) {
while(listLength(server.ready_keys) != 0) {
list *l;
/* Point server.ready_keys to a fresh list and save the current one
* locally. This way as we run the old list we are free to call
* signalListAsReady() that may push new elements in server.ready_keys
* when handling clients blocked into BRPOPLPUSH. */
l = server.ready_keys;
server.ready_keys = listCreate();
while(listLength(l) != 0) {
listNode *ln = listFirst(l);
readyList *rl = ln->value;
/* First of all remove this key from db->ready_keys so that
* we can safely call signalListAsReady() against this key. */
dictDelete(rl->db->ready_keys,rl->key);
/* If the key exists and it's a list, serve blocked clients
* with data. */
robj *o = lookupKeyWrite(rl->db,rl->key);
if (o != NULL && o->type == OBJ_LIST) {
dictEntry *de;
/* We serve clients in the same order they blocked for
* this key, from the first blocked to the last. */
de = dictFind(rl->db->blocking_keys,rl->key);
if (de) {
list *clients = dictGetVal(de);
int numclients = listLength(clients);
while(numclients--) {
listNode *clientnode = listFirst(clients);
client *receiver = clientnode->value;
robj *dstkey = receiver->bpop.target;
int where = (receiver->lastcmd &&
receiver->lastcmd->proc == blpopCommand) ?
LIST_HEAD : LIST_TAIL;
robj *value = listTypePop(o,where);
if (value) {
/* Protect receiver->bpop.target, that will be
* freed by the next unblockClient()
* call. */
if (dstkey) incrRefCount(dstkey);
unblockClient(receiver);
if (serveClientBlockedOnList(receiver,
rl->key,dstkey,rl->db,value,
where) == C_ERR)
{
/* If we failed serving the client we need
* to also undo the POP operation. */
listTypePush(o,value,where);
}
if (dstkey) decrRefCount(dstkey);
decrRefCount(value);
} else {
break;
}
}
}
if (listTypeLength(o) == 0) {
dbDelete(rl->db,rl->key);
}
/* We don't call signalModifiedKey() as it was already called
* when an element was pushed on the list. */
}
/* Free this item. */
decrRefCount(rl->key);
zfree(rl);
listDelNode(l,ln);
}
listRelease(l); /* We have the new list on place at this point. */
}
}
/* Blocking RPOP/LPOP */ /* Blocking RPOP/LPOP */
void blockingPopGenericCommand(client *c, int where) { void blockingPopGenericCommand(client *c, int where) {
robj *o; robj *o;
...@@ -930,7 +726,7 @@ void blockingPopGenericCommand(client *c, int where) { ...@@ -930,7 +726,7 @@ void blockingPopGenericCommand(client *c, int where) {
} }
/* If the list is empty or the key does not exists we must block */ /* If the list is empty or the key does not exists we must block */
blockForKeys(c, c->argv + 1, c->argc - 2, timeout, NULL); blockForKeys(c,BLOCKED_LIST,c->argv + 1,c->argc - 2,timeout,NULL,NULL);
} }
void blpopCommand(client *c) { void blpopCommand(client *c) {
...@@ -956,7 +752,7 @@ void brpoplpushCommand(client *c) { ...@@ -956,7 +752,7 @@ void brpoplpushCommand(client *c) {
addReply(c, shared.nullbulk); addReply(c, shared.nullbulk);
} else { } else {
/* The list is empty and the client blocks. */ /* The list is empty and the client blocks. */
blockForKeys(c, c->argv + 1, 1, timeout, c->argv[2]); blockForKeys(c,BLOCKED_LIST,c->argv + 1,1,timeout,c->argv[2],NULL);
} }
} else { } else {
if (key->type != OBJ_LIST) { if (key->type != OBJ_LIST) {
......
...@@ -407,7 +407,7 @@ void spopWithCountCommand(client *c) { ...@@ -407,7 +407,7 @@ void spopWithCountCommand(client *c) {
/* Get the count argument */ /* Get the count argument */
if (getLongFromObjectOrReply(c,c->argv[2],&l,NULL) != C_OK) return; if (getLongFromObjectOrReply(c,c->argv[2],&l,NULL) != C_OK) return;
if (l >= 0) { if (l >= 0) {
count = (unsigned) l; count = (unsigned long) l;
} else { } else {
addReply(c,shared.outofrangeerr); addReply(c,shared.outofrangeerr);
return; return;
...@@ -626,7 +626,7 @@ void srandmemberWithCountCommand(client *c) { ...@@ -626,7 +626,7 @@ void srandmemberWithCountCommand(client *c) {
if (getLongFromObjectOrReply(c,c->argv[2],&l,NULL) != C_OK) return; if (getLongFromObjectOrReply(c,c->argv[2],&l,NULL) != C_OK) return;
if (l >= 0) { if (l >= 0) {
count = (unsigned) l; count = (unsigned long) l;
} else { } else {
/* A negative count means: return the same elements multiple times /* A negative count means: return the same elements multiple times
* (i.e. don't remove the extracted element after every extraction). */ * (i.e. don't remove the extracted element after every extraction). */
...@@ -774,15 +774,21 @@ void srandmemberCommand(client *c) { ...@@ -774,15 +774,21 @@ void srandmemberCommand(client *c) {
} }
int qsortCompareSetsByCardinality(const void *s1, const void *s2) { int qsortCompareSetsByCardinality(const void *s1, const void *s2) {
return setTypeSize(*(robj**)s1)-setTypeSize(*(robj**)s2); if (setTypeSize(*(robj**)s1) > setTypeSize(*(robj**)s2)) return 1;
if (setTypeSize(*(robj**)s1) < setTypeSize(*(robj**)s2)) return -1;
return 0;
} }
/* This is used by SDIFF and in this case we can receive NULL that should /* This is used by SDIFF and in this case we can receive NULL that should
* be handled as empty sets. */ * be handled as empty sets. */
int qsortCompareSetsByRevCardinality(const void *s1, const void *s2) { int qsortCompareSetsByRevCardinality(const void *s1, const void *s2) {
robj *o1 = *(robj**)s1, *o2 = *(robj**)s2; robj *o1 = *(robj**)s1, *o2 = *(robj**)s2;
unsigned long first = o1 ? setTypeSize(o1) : 0;
unsigned long second = o2 ? setTypeSize(o2) : 0;
return (o2 ? setTypeSize(o2) : 0) - (o1 ? setTypeSize(o1) : 0); if (first < second) return 1;
if (first > second) return -1;
return 0;
} }
void sinterGenericCommand(client *c, robj **setkeys, void sinterGenericCommand(client *c, robj **setkeys,
......
This diff is collapsed.
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