Commit 77a7ec72 authored by antirez's avatar antirez
Browse files

Merge branch 'unstable' into 5.0 branch

parents 48dfd42d 4ff47a0b
...@@ -285,26 +285,26 @@ void computeDatasetDigest(unsigned char *final) { ...@@ -285,26 +285,26 @@ void computeDatasetDigest(unsigned char *final) {
void debugCommand(client *c) { void debugCommand(client *c) {
if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"help")) { if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"help")) {
const char *help[] = { const char *help[] = {
"assert -- Crash by assertion failed.", "ASSERT -- Crash by assertion failed.",
"change-repl-id -- Change the replication IDs of the instance. Dangerous, should be used only for testing the replication subsystem.", "CHANGE-REPL-ID -- Change the replication IDs of the instance. Dangerous, should be used only for testing the replication subsystem.",
"crash-and-recover <milliseconds> -- Hard crash and restart after <milliseconds> delay.", "CRASH-AND-RECOVER <milliseconds> -- Hard crash and restart after <milliseconds> delay.",
"digest -- Outputs an hex signature representing the current DB content.", "DIGEST -- Output a hex signature representing the current DB content.",
"htstats <dbid> -- Return hash table statistics of the specified Redis database.", "ERROR <string> -- Return a Redis protocol error with <string> as message. Useful for clients unit tests to simulate Redis errors.",
"htstats-key <key> -- Like htstats but for the hash table stored as key's value.", "HTSTATS <dbid> -- Return hash table statistics of the specified Redis database.",
"loadaof -- Flush the AOF buffers on disk and reload the AOF in memory.", "HTSTATS-KEY <key> -- Like htstats but for the hash table stored as key's value.",
"lua-always-replicate-commands (0|1) -- Setting it to 1 makes Lua replication defaulting to replicating single commands, without the script having to enable effects replication.", "LOADAOF -- Flush the AOF buffers on disk and reload the AOF in memory.",
"object <key> -- Show low level info about key and associated value.", "LUA-ALWAYS-REPLICATE-COMMANDS <0|1> -- Setting it to 1 makes Lua replication defaulting to replicating single commands, without the script having to enable effects replication.",
"panic -- Crash the server simulating a panic.", "OBJECT <key> -- Show low level info about key and associated value.",
"populate <count> [prefix] [size] -- Create <count> string keys named key:<num>. If a prefix is specified is used instead of the 'key' prefix.", "PANIC -- Crash the server simulating a panic.",
"reload -- Save the RDB on disk and reload it back in memory.", "POPULATE <count> [prefix] [size] -- Create <count> string keys named key:<num>. If a prefix is specified is used instead of the 'key' prefix.",
"restart -- Graceful restart: save config, db, restart.", "RELOAD -- Save the RDB on disk and reload it back in memory.",
"sdslen <key> -- Show low level SDS string info representing key and value.", "RESTART -- Graceful restart: save config, db, restart.",
"segfault -- Crash the server with sigsegv.", "SDSLEN <key> -- Show low level SDS string info representing key and value.",
"set-active-expire (0|1) -- Setting it to 0 disables expiring keys in background when they are not accessed (otherwise the Redis behavior). Setting it to 1 reenables back the default.", "SEGFAULT -- Crash the server with sigsegv.",
"sleep <seconds> -- Stop the server for <seconds>. Decimals allowed.", "SET-ACTIVE-EXPIRE <0|1> -- Setting it to 0 disables expiring keys in background when they are not accessed (otherwise the Redis behavior). Setting it to 1 reenables back the default.",
"structsize -- Return the size of different Redis core C structures.", "SLEEP <seconds> -- Stop the server for <seconds>. Decimals allowed.",
"ziplist <key> -- Show low level info about the ziplist encoding.", "STRUCTSIZE -- Return the size of different Redis core C structures.",
"error <string> -- Return a Redis protocol error with <string> as message. Useful for clients unit tests to simulate Redis errors.", "ZIPLIST <key> -- Show low level info about the ziplist encoding.",
NULL NULL
}; };
addReplyHelp(c, help); addReplyHelp(c, help);
...@@ -348,7 +348,7 @@ NULL ...@@ -348,7 +348,7 @@ NULL
serverLog(LL_WARNING,"DB reloaded by DEBUG RELOAD"); serverLog(LL_WARNING,"DB reloaded by DEBUG RELOAD");
addReply(c,shared.ok); addReply(c,shared.ok);
} else if (!strcasecmp(c->argv[1]->ptr,"loadaof")) { } else if (!strcasecmp(c->argv[1]->ptr,"loadaof")) {
if (server.aof_state == AOF_ON) flushAppendOnlyFile(1); if (server.aof_state != AOF_OFF) flushAppendOnlyFile(1);
emptyDb(-1,EMPTYDB_NO_FLAGS,NULL); emptyDb(-1,EMPTYDB_NO_FLAGS,NULL);
if (loadAppendOnlyFile(server.aof_filename) != C_OK) { if (loadAppendOnlyFile(server.aof_filename) != C_OK) {
addReply(c,shared.err); addReply(c,shared.err);
...@@ -582,8 +582,7 @@ NULL ...@@ -582,8 +582,7 @@ NULL
clearReplicationId2(); clearReplicationId2();
addReply(c,shared.ok); addReply(c,shared.ok);
} else { } else {
addReplyErrorFormat(c, "Unknown subcommand or wrong number of arguments for '%s'. Try DEBUG HELP", addReplySubcommandSyntaxError(c);
(char*)c->argv[1]->ptr);
return; return;
} }
} }
...@@ -1077,7 +1076,7 @@ void sigsegvHandler(int sig, siginfo_t *info, void *secret) { ...@@ -1077,7 +1076,7 @@ void sigsegvHandler(int sig, siginfo_t *info, void *secret) {
infostring = genRedisInfoString("all"); infostring = genRedisInfoString("all");
serverLogRaw(LL_WARNING|LL_RAW, infostring); serverLogRaw(LL_WARNING|LL_RAW, infostring);
serverLogRaw(LL_WARNING|LL_RAW, "\n------ CLIENT LIST OUTPUT ------\n"); serverLogRaw(LL_WARNING|LL_RAW, "\n------ CLIENT LIST OUTPUT ------\n");
clients = getAllClientsInfoString(); clients = getAllClientsInfoString(-1);
serverLogRaw(LL_WARNING|LL_RAW, clients); serverLogRaw(LL_WARNING|LL_RAW, clients);
sdsfree(infostring); sdsfree(infostring);
sdsfree(clients); sdsfree(clients);
......
...@@ -592,6 +592,171 @@ long defragSet(redisDb *db, dictEntry *kde) { ...@@ -592,6 +592,171 @@ long defragSet(redisDb *db, dictEntry *kde) {
return defragged; return defragged;
} }
/* Defrag callback for radix tree iterator, called for each node,
* used in order to defrag the nodes allocations. */
int defragRaxNode(raxNode **noderef) {
raxNode *newnode = activeDefragAlloc(*noderef);
if (newnode) {
*noderef = newnode;
return 1;
}
return 0;
}
/* returns 0 if no more work needs to be been done, and 1 if time is up and more work is needed. */
int scanLaterStraemListpacks(robj *ob, unsigned long *cursor, long long endtime, long long *defragged) {
static unsigned char last[sizeof(streamID)];
raxIterator ri;
long iterations = 0;
if (ob->type != OBJ_STREAM || ob->encoding != OBJ_ENCODING_STREAM) {
*cursor = 0;
return 0;
}
stream *s = ob->ptr;
raxStart(&ri,s->rax);
if (*cursor == 0) {
/* if cursor is 0, we start new iteration */
defragRaxNode(&s->rax->head);
/* assign the iterator node callback before the seek, so that the
* initial nodes that are processed till the first item are covered */
ri.node_cb = defragRaxNode;
raxSeek(&ri,"^",NULL,0);
} else {
/* if cursor is non-zero, we seek to the static 'last' */
if (!raxSeek(&ri,">", last, sizeof(last))) {
*cursor = 0;
return 0;
}
/* assign the iterator node callback after the seek, so that the
* initial nodes that are processed till now aren't covered */
ri.node_cb = defragRaxNode;
}
(*cursor)++;
while (raxNext(&ri)) {
void *newdata = activeDefragAlloc(ri.data);
if (newdata)
raxSetData(ri.node, ri.data=newdata), (*defragged)++;
if (++iterations > 16) {
if (ustime() > endtime) {
serverAssert(ri.key_len==sizeof(last));
memcpy(last,ri.key,ri.key_len);
raxStop(&ri);
return 1;
}
iterations = 0;
}
}
raxStop(&ri);
*cursor = 0;
return 0;
}
/* optional callback used defrag each rax element (not including the element pointer itself) */
typedef void *(raxDefragFunction)(raxIterator *ri, void *privdata, long *defragged);
/* defrag radix tree including:
* 1) rax struct
* 2) rax nodes
* 3) rax entry data (only if defrag_data is specified)
* 4) call a callback per element, and allow the callback to return a new pointer for the element */
long defragRadixTree(rax **raxref, int defrag_data, raxDefragFunction *element_cb, void *element_cb_data) {
long defragged = 0;
raxIterator ri;
rax* rax;
if ((rax = activeDefragAlloc(*raxref)))
defragged++, *raxref = rax;
rax = *raxref;
raxStart(&ri,rax);
ri.node_cb = defragRaxNode;
defragRaxNode(&rax->head);
raxSeek(&ri,"^",NULL,0);
while (raxNext(&ri)) {
void *newdata = NULL;
if (element_cb)
newdata = element_cb(&ri, element_cb_data, &defragged);
if (defrag_data && !newdata)
newdata = activeDefragAlloc(ri.data);
if (newdata)
raxSetData(ri.node, ri.data=newdata), defragged++;
}
raxStop(&ri);
return defragged;
}
typedef struct {
streamCG *cg;
streamConsumer *c;
} PendingEntryContext;
void* defragStreamConsumerPendingEntry(raxIterator *ri, void *privdata, long *defragged) {
UNUSED(defragged);
PendingEntryContext *ctx = privdata;
streamNACK *nack = ri->data, *newnack;
nack->consumer = ctx->c; /* update nack pointer to consumer */
newnack = activeDefragAlloc(nack);
if (newnack) {
/* update consumer group pointer to the nack */
void *prev;
raxInsert(ctx->cg->pel, ri->key, ri->key_len, newnack, &prev);
serverAssert(prev==nack);
/* note: we don't increment 'defragged' that's done by the caller */
}
return newnack;
}
void* defragStreamConsumer(raxIterator *ri, void *privdata, long *defragged) {
streamConsumer *c = ri->data;
streamCG *cg = privdata;
void *newc = activeDefragAlloc(c);
if (newc) {
/* note: we don't increment 'defragged' that's done by the caller */
c = newc;
}
sds newsds = activeDefragSds(c->name);
if (newsds)
(*defragged)++, c->name = newsds;
if (c->pel) {
PendingEntryContext pel_ctx = {cg, c};
*defragged += defragRadixTree(&c->pel, 0, defragStreamConsumerPendingEntry, &pel_ctx);
}
return newc; /* returns NULL if c was not defragged */
}
void* defragStreamConsumerGroup(raxIterator *ri, void *privdata, long *defragged) {
streamCG *cg = ri->data;
UNUSED(privdata);
if (cg->consumers)
*defragged += defragRadixTree(&cg->consumers, 0, defragStreamConsumer, cg);
if (cg->pel)
*defragged += defragRadixTree(&cg->pel, 0, NULL, NULL);
return NULL;
}
long defragStream(redisDb *db, dictEntry *kde) {
long defragged = 0;
robj *ob = dictGetVal(kde);
serverAssert(ob->type == OBJ_STREAM && ob->encoding == OBJ_ENCODING_STREAM);
stream *s = ob->ptr, *news;
/* handle the main struct */
if ((news = activeDefragAlloc(s)))
defragged++, ob->ptr = s = news;
if (raxSize(s->rax) > server.active_defrag_max_scan_fields) {
rax *newrax = activeDefragAlloc(s->rax);
if (newrax)
defragged++, s->rax = newrax;
defragLater(db, kde);
} else
defragged += defragRadixTree(&s->rax, 1, NULL, NULL);
if (s->cgroups)
defragged += defragRadixTree(&s->cgroups, 1, defragStreamConsumerGroup, NULL);
return defragged;
}
/* for each key we scan in the main dict, this function will attempt to defrag /* for each key we scan in the main dict, this function will attempt to defrag
* all the various pointers it has. Returns a stat of how many pointers were * all the various pointers it has. Returns a stat of how many pointers were
* moved. */ * moved. */
...@@ -660,6 +825,8 @@ long defragKey(redisDb *db, dictEntry *de) { ...@@ -660,6 +825,8 @@ long defragKey(redisDb *db, dictEntry *de) {
} else { } else {
serverPanic("Unknown hash encoding"); serverPanic("Unknown hash encoding");
} }
} else if (ob->type == OBJ_STREAM) {
defragged += defragStream(db, de);
} else if (ob->type == OBJ_MODULE) { } else if (ob->type == OBJ_MODULE) {
/* Currently defragmenting modules private data types /* Currently defragmenting modules private data types
* is not supported. */ * is not supported. */
...@@ -680,7 +847,7 @@ void defragScanCallback(void *privdata, const dictEntry *de) { ...@@ -680,7 +847,7 @@ void defragScanCallback(void *privdata, const dictEntry *de) {
server.stat_active_defrag_scanned++; server.stat_active_defrag_scanned++;
} }
/* Defrag scan callback for for each hash table bicket, /* Defrag scan callback for each hash table bicket,
* used in order to defrag the dictEntry allocations. */ * used in order to defrag the dictEntry allocations. */
void defragDictBucketCallback(void *privdata, dictEntry **bucketref) { void defragDictBucketCallback(void *privdata, dictEntry **bucketref) {
UNUSED(privdata); /* NOTE: this function is also used by both activeDefragCycle and scanLaterHash, etc. don't use privdata */ UNUSED(privdata); /* NOTE: this function is also used by both activeDefragCycle and scanLaterHash, etc. don't use privdata */
...@@ -700,9 +867,8 @@ void defragDictBucketCallback(void *privdata, dictEntry **bucketref) { ...@@ -700,9 +867,8 @@ void defragDictBucketCallback(void *privdata, dictEntry **bucketref) {
* or not, a false detection can cause the defragmenter to waste a lot of CPU * or not, a false detection can cause the defragmenter to waste a lot of CPU
* without the possibility of getting any results. */ * without the possibility of getting any results. */
float getAllocatorFragmentation(size_t *out_frag_bytes) { float getAllocatorFragmentation(size_t *out_frag_bytes) {
size_t resident = server.cron_malloc_stats.allocator_resident; size_t resident, active, allocated;
size_t active = server.cron_malloc_stats.allocator_active; zmalloc_get_allocator_info(&allocated, &active, &resident);
size_t allocated = server.cron_malloc_stats.allocator_allocated;
float frag_pct = ((float)active / allocated)*100 - 100; float frag_pct = ((float)active / allocated)*100 - 100;
size_t frag_bytes = active - allocated; size_t frag_bytes = active - allocated;
float rss_pct = ((float)resident / allocated)*100 - 100; float rss_pct = ((float)resident / allocated)*100 - 100;
...@@ -728,27 +894,29 @@ long defragOtherGlobals() { ...@@ -728,27 +894,29 @@ long defragOtherGlobals() {
return defragged; return defragged;
} }
unsigned long defragLaterItem(dictEntry *de, unsigned long cursor) { /* returns 0 more work may or may not be needed (see non-zero cursor),
long defragged = 0; * and 1 if time is up and more work is needed. */
int defragLaterItem(dictEntry *de, unsigned long *cursor, long long endtime) {
if (de) { if (de) {
robj *ob = dictGetVal(de); robj *ob = dictGetVal(de);
if (ob->type == OBJ_LIST) { if (ob->type == OBJ_LIST) {
defragged += scanLaterList(ob); server.stat_active_defrag_hits += scanLaterList(ob);
cursor = 0; /* list has no scan, we must finish it in one go */ *cursor = 0; /* list has no scan, we must finish it in one go */
} else if (ob->type == OBJ_SET) { } else if (ob->type == OBJ_SET) {
defragged += scanLaterSet(ob, &cursor); server.stat_active_defrag_hits += scanLaterSet(ob, cursor);
} else if (ob->type == OBJ_ZSET) { } else if (ob->type == OBJ_ZSET) {
defragged += scanLaterZset(ob, &cursor); server.stat_active_defrag_hits += scanLaterZset(ob, cursor);
} else if (ob->type == OBJ_HASH) { } else if (ob->type == OBJ_HASH) {
defragged += scanLaterHash(ob, &cursor); server.stat_active_defrag_hits += scanLaterHash(ob, cursor);
} else if (ob->type == OBJ_STREAM) {
return scanLaterStraemListpacks(ob, cursor, endtime, &server.stat_active_defrag_hits);
} else { } else {
cursor = 0; /* object type may have changed since we schedule it for later */ *cursor = 0; /* object type may have changed since we schedule it for later */
} }
} else { } else {
cursor = 0; /* object may have been deleted already */ *cursor = 0; /* object may have been deleted already */
} }
server.stat_active_defrag_hits += defragged; return 0;
return cursor;
} }
/* returns 0 if no more work needs to be been done, and 1 if time is up and more work is needed. */ /* returns 0 if no more work needs to be been done, and 1 if time is up and more work is needed. */
...@@ -788,17 +956,22 @@ int defragLaterStep(redisDb *db, long long endtime) { ...@@ -788,17 +956,22 @@ int defragLaterStep(redisDb *db, long long endtime) {
dictEntry *de = dictFind(db->dict, current_key); dictEntry *de = dictFind(db->dict, current_key);
key_defragged = server.stat_active_defrag_hits; key_defragged = server.stat_active_defrag_hits;
do { do {
cursor = defragLaterItem(de, cursor); int quit = 0;
if (defragLaterItem(de, &cursor, endtime))
quit = 1; /* time is up, we didn't finish all the work */
/* Don't start a new BIG key in this loop, this is because the
* next key can be a list, and scanLaterList must be done in once cycle */
if (!cursor)
quit = 1;
/* Once in 16 scan iterations, 512 pointer reallocations, or 64 fields /* Once in 16 scan iterations, 512 pointer reallocations, or 64 fields
* (if we have a lot of pointers in one hash bucket, or rehashing), * (if we have a lot of pointers in one hash bucket, or rehashing),
* check if we reached the time limit. * check if we reached the time limit. */
* But regardless, don't start a new BIG key in this loop, this is because the if (quit || (++iterations > 16 ||
* next key can be a list, and scanLaterList must be done in once cycle */
if (!cursor || (++iterations > 16 ||
server.stat_active_defrag_hits - prev_defragged > 512 || server.stat_active_defrag_hits - prev_defragged > 512 ||
server.stat_active_defrag_scanned - prev_scanned > 64)) { server.stat_active_defrag_scanned - prev_scanned > 64)) {
if (!cursor || ustime() > endtime) { if (quit || ustime() > endtime) {
if(key_defragged != server.stat_active_defrag_hits) if(key_defragged != server.stat_active_defrag_hits)
server.stat_active_defrag_key_hits++; server.stat_active_defrag_key_hits++;
else else
......
...@@ -327,7 +327,7 @@ int dictReplace(dict *d, void *key, void *val) ...@@ -327,7 +327,7 @@ int dictReplace(dict *d, void *key, void *val)
dictEntry *entry, *existing, auxentry; dictEntry *entry, *existing, auxentry;
/* Try to add the element. If the key /* Try to add the element. If the key
* does not exists dictAdd will suceed. */ * does not exists dictAdd will succeed. */
entry = dictAddRaw(d,key,&existing); entry = dictAddRaw(d,key,&existing);
if (entry) { if (entry) {
dictSetVal(d, entry, val); dictSetVal(d, entry, val);
...@@ -705,8 +705,10 @@ unsigned int dictGetSomeKeys(dict *d, dictEntry **des, unsigned int count) { ...@@ -705,8 +705,10 @@ unsigned int dictGetSomeKeys(dict *d, dictEntry **des, unsigned int count) {
* table, there will be no elements in both tables up to * table, there will be no elements in both tables up to
* the current rehashing index, so we jump if possible. * the current rehashing index, so we jump if possible.
* (this happens when going from big to small table). */ * (this happens when going from big to small table). */
if (i >= d->ht[1].size) i = d->rehashidx; if (i >= d->ht[1].size)
continue; i = d->rehashidx;
else
continue;
} }
if (i >= d->ht[j].size) continue; /* Out of range for this table. */ if (i >= d->ht[j].size) continue; /* Out of range for this table. */
dictEntry *he = d->ht[j].table[i]; dictEntry *he = d->ht[j].table[i];
......
...@@ -43,12 +43,12 @@ uint16_t intrev16(uint16_t v); ...@@ -43,12 +43,12 @@ uint16_t intrev16(uint16_t v);
uint32_t intrev32(uint32_t v); uint32_t intrev32(uint32_t v);
uint64_t intrev64(uint64_t v); uint64_t intrev64(uint64_t v);
/* variants of the function doing the actual convertion only if the target /* variants of the function doing the actual conversion only if the target
* host is big endian */ * host is big endian */
#if (BYTE_ORDER == LITTLE_ENDIAN) #if (BYTE_ORDER == LITTLE_ENDIAN)
#define memrev16ifbe(p) #define memrev16ifbe(p) ((void)(0))
#define memrev32ifbe(p) #define memrev32ifbe(p) ((void)(0))
#define memrev64ifbe(p) #define memrev64ifbe(p) ((void)(0))
#define intrev16ifbe(v) (v) #define intrev16ifbe(v) (v)
#define intrev32ifbe(v) (v) #define intrev32ifbe(v) (v)
#define intrev64ifbe(v) (v) #define intrev64ifbe(v) (v)
......
...@@ -112,7 +112,7 @@ void activeExpireCycle(int type) { ...@@ -112,7 +112,7 @@ void activeExpireCycle(int type) {
if (type == ACTIVE_EXPIRE_CYCLE_FAST) { if (type == ACTIVE_EXPIRE_CYCLE_FAST) {
/* Don't start a fast cycle if the previous cycle did not exit /* Don't start a fast cycle if the previous cycle did not exit
* for time limt. Also don't repeat a fast cycle for the same period * for time limit. Also don't repeat a fast cycle for the same period
* as the fast cycle total duration itself. */ * as the fast cycle total duration itself. */
if (!timelimit_exit) return; if (!timelimit_exit) return;
if (start < last_fast_cycle + ACTIVE_EXPIRE_CYCLE_FAST_DURATION*2) return; if (start < last_fast_cycle + ACTIVE_EXPIRE_CYCLE_FAST_DURATION*2) return;
......
...@@ -145,7 +145,7 @@ double extractUnitOrReply(client *c, robj *unit) { ...@@ -145,7 +145,7 @@ double extractUnitOrReply(client *c, robj *unit) {
/* Input Argument Helper. /* Input Argument Helper.
* Extract the dinstance from the specified two arguments starting at 'argv' * Extract the dinstance from the specified two arguments starting at 'argv'
* that shouldbe in the form: <number> <unit> and return the dinstance in the * that shouldbe in the form: <number> <unit> and return the dinstance in the
* specified unit on success. *conversino is populated with the coefficient * specified unit on success. *conversions is populated with the coefficient
* to use in order to convert meters to the unit. * to use in order to convert meters to the unit.
* *
* On error a value less than zero is returned. */ * On error a value less than zero is returned. */
......
...@@ -144,8 +144,8 @@ int geohashEncode(const GeoHashRange *long_range, const GeoHashRange *lat_range, ...@@ -144,8 +144,8 @@ int geohashEncode(const GeoHashRange *long_range, const GeoHashRange *lat_range,
(longitude - long_range->min) / (long_range->max - long_range->min); (longitude - long_range->min) / (long_range->max - long_range->min);
/* convert to fixed point based on the step size */ /* convert to fixed point based on the step size */
lat_offset *= (1 << step); lat_offset *= (1ULL << step);
long_offset *= (1 << step); long_offset *= (1ULL << step);
hash->bits = interleave64(lat_offset, long_offset); hash->bits = interleave64(lat_offset, long_offset);
return 1; return 1;
} }
......
...@@ -975,7 +975,7 @@ struct commandHelp { ...@@ -975,7 +975,7 @@ struct commandHelp {
"5.0.0" }, "5.0.0" },
{ "XPENDING", { "XPENDING",
"key group [start end count] [consumer]", "key group [start end count] [consumer]",
"Return information and entries from a stream conusmer group pending entries list, that are messages fetched but never acknowledged.", "Return information and entries from a stream consumer group pending entries list, that are messages fetched but never acknowledged.",
14, 14,
"5.0.0" }, "5.0.0" },
{ "XRANGE", { "XRANGE",
......
...@@ -673,7 +673,7 @@ int hllSparseSet(robj *o, long index, uint8_t count) { ...@@ -673,7 +673,7 @@ int hllSparseSet(robj *o, long index, uint8_t count) {
end = p + sdslen(o->ptr) - HLL_HDR_SIZE; end = p + sdslen(o->ptr) - HLL_HDR_SIZE;
first = 0; first = 0;
prev = NULL; /* Points to previos opcode at the end of the loop. */ prev = NULL; /* Points to previous opcode at the end of the loop. */
next = NULL; /* Points to the next opcode at the end of the loop. */ next = NULL; /* Points to the next opcode at the end of the loop. */
span = 0; span = 0;
while(p < end) { while(p < end) {
...@@ -764,7 +764,7 @@ int hllSparseSet(robj *o, long index, uint8_t count) { ...@@ -764,7 +764,7 @@ int hllSparseSet(robj *o, long index, uint8_t count) {
* and is either currently represented by a VAL opcode with len > 1, * and is either currently represented by a VAL opcode with len > 1,
* by a ZERO opcode with len > 1, or by an XZERO opcode. * by a ZERO opcode with len > 1, or by an XZERO opcode.
* *
* In those cases the original opcode must be split into muliple * In those cases the original opcode must be split into multiple
* opcodes. The worst case is an XZERO split in the middle resuling into * opcodes. The worst case is an XZERO split in the middle resuling into
* XZERO - VAL - XZERO, so the resulting sequence max length is * XZERO - VAL - XZERO, so the resulting sequence max length is
* 5 bytes. * 5 bytes.
...@@ -887,7 +887,7 @@ promote: /* Promote to dense representation. */ ...@@ -887,7 +887,7 @@ promote: /* Promote to dense representation. */
* *
* Note that this in turn means that PFADD will make sure the command * Note that this in turn means that PFADD will make sure the command
* is propagated to slaves / AOF, so if there is a sparse -> dense * is propagated to slaves / AOF, so if there is a sparse -> dense
* convertion, it will be performed in all the slaves as well. */ * conversion, it will be performed in all the slaves as well. */
int dense_retval = hllDenseSet(hdr->registers,index,count); int dense_retval = hllDenseSet(hdr->registers,index,count);
serverAssert(dense_retval == 1); serverAssert(dense_retval == 1);
return dense_retval; return dense_retval;
......
...@@ -152,7 +152,7 @@ int latencyResetEvent(char *event_to_reset) { ...@@ -152,7 +152,7 @@ int latencyResetEvent(char *event_to_reset) {
/* ------------------------ Latency reporting (doctor) ---------------------- */ /* ------------------------ Latency reporting (doctor) ---------------------- */
/* Analyze the samples avaialble for a given event and return a structure /* Analyze the samples available for a given event and return a structure
* populate with different metrics, average, MAD, min, max, and so forth. * populate with different metrics, average, MAD, min, max, and so forth.
* Check latency.h definition of struct latenctStat for more info. * Check latency.h definition of struct latenctStat for more info.
* If the specified event has no elements the structure is populate with * If the specified event has no elements the structure is populate with
...@@ -294,7 +294,7 @@ sds createLatencyReport(void) { ...@@ -294,7 +294,7 @@ sds createLatencyReport(void) {
/* Potentially commands. */ /* Potentially commands. */
if (!strcasecmp(event,"command")) { if (!strcasecmp(event,"command")) {
if (server.slowlog_log_slower_than == 0) { if (server.slowlog_log_slower_than < 0) {
advise_slowlog_enabled = 1; advise_slowlog_enabled = 1;
advices++; advices++;
} else if (server.slowlog_log_slower_than/1000 > } else if (server.slowlog_log_slower_than/1000 >
......
...@@ -23,10 +23,10 @@ size_t lazyfreeGetPendingObjectsCount(void) { ...@@ -23,10 +23,10 @@ size_t lazyfreeGetPendingObjectsCount(void) {
* the function just returns the number of elements the object is composed of. * the function just returns the number of elements the object is composed of.
* *
* Objects composed of single allocations are always reported as having a * Objects composed of single allocations are always reported as having a
* single item even if they are actaully logical composed of multiple * single item even if they are actually logical composed of multiple
* elements. * elements.
* *
* For lists the funciton returns the number of elements in the quicklist * For lists the function returns the number of elements in the quicklist
* representing the list. */ * representing the list. */
size_t lazyfreeGetFreeEffort(robj *obj) { size_t lazyfreeGetFreeEffort(robj *obj) {
if (obj->type == OBJ_LIST) { if (obj->type == OBJ_LIST) {
......
...@@ -291,7 +291,7 @@ int lpEncodeGetType(unsigned char *ele, uint32_t size, unsigned char *intenc, ui ...@@ -291,7 +291,7 @@ int lpEncodeGetType(unsigned char *ele, uint32_t size, unsigned char *intenc, ui
/* Store a reverse-encoded variable length field, representing the length /* Store a reverse-encoded variable length field, representing the length
* of the previous element of size 'l', in the target buffer 'buf'. * of the previous element of size 'l', in the target buffer 'buf'.
* The function returns the number of bytes used to encode it, from * The function returns the number of bytes used to encode it, from
* 1 to 5. If 'buf' is NULL the funciton just returns the number of bytes * 1 to 5. If 'buf' is NULL the function just returns the number of bytes
* needed in order to encode the backlen. */ * needed in order to encode the backlen. */
unsigned long lpEncodeBacklen(unsigned char *buf, uint64_t l) { unsigned long lpEncodeBacklen(unsigned char *buf, uint64_t l) {
if (l <= 127) { if (l <= 127) {
...@@ -568,7 +568,7 @@ unsigned char *lpGet(unsigned char *p, int64_t *count, unsigned char *intbuf) { ...@@ -568,7 +568,7 @@ unsigned char *lpGet(unsigned char *p, int64_t *count, unsigned char *intbuf) {
} }
} }
/* Insert, delete or replace the specified element 'ele' of lenght 'len' at /* Insert, delete or replace the specified element 'ele' of length 'len' at
* the specified position 'p', with 'p' being a listpack element pointer * the specified position 'p', with 'p' being a listpack element pointer
* obtained with lpFirst(), lpLast(), lpIndex(), lpNext(), lpPrev() or * obtained with lpFirst(), lpLast(), lpIndex(), lpNext(), lpPrev() or
* lpSeek(). * lpSeek().
...@@ -710,7 +710,7 @@ unsigned char *lpInsert(unsigned char *lp, unsigned char *ele, uint32_t size, un ...@@ -710,7 +710,7 @@ unsigned char *lpInsert(unsigned char *lp, unsigned char *ele, uint32_t size, un
return lp; return lp;
} }
/* Append the specified element 'ele' of lenght 'len' at the end of the /* Append the specified element 'ele' of length 'len' at the end of the
* listpack. It is implemented in terms of lpInsert(), so the return value is * listpack. It is implemented in terms of lpInsert(), so the return value is
* the same as lpInsert(). */ * the same as lpInsert(). */
unsigned char *lpAppend(unsigned char *lp, unsigned char *ele, uint32_t size) { unsigned char *lpAppend(unsigned char *lp, unsigned char *ele, uint32_t size) {
......
/*
* Copyright (c) 2018, Salvatore Sanfilippo <antirez at gmail dot com>
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
*
* * Redistributions of source code must retain the above copyright notice,
* this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* * Neither the name of Redis nor the names of its contributors may be used
* to endorse or promote products derived from this software without
* specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGE.
*/
#include <time.h>
/* This is a safe version of localtime() which contains no locks and is
* fork() friendly. Even the _r version of localtime() cannot be used safely
* in Redis. Another thread may be calling localtime() while the main thread
* forks(). Later when the child process calls localtime() again, for instance
* in order to log something to the Redis log, it may deadlock: in the copy
* of the address space of the forked process the lock will never be released.
*
* This function takes the timezone 'tz' as argument, and the 'dst' flag is
* used to check if daylight saving time is currently in effect. The caller
* of this function should obtain such information calling tzset() ASAP in the
* main() function to obtain the timezone offset from the 'timezone' global
* variable. To obtain the daylight information, if it is currently active or not,
* one trick is to call localtime() in main() ASAP as well, and get the
* information from the tm_isdst field of the tm structure. However the daylight
* time may switch in the future for long running processes, so this information
* should be refreshed at safe times.
*
* Note that this function does not work for dates < 1/1/1970, it is solely
* designed to work with what time(NULL) may return, and to support Redis
* logging of the dates, it's not really a complete implementation. */
static int is_leap_year(time_t year) {
if (year % 4) return 0; /* A year not divisible by 4 is not leap. */
else if (year % 100) return 1; /* If div by 4 and not 100 is surely leap. */
else if (year % 400) return 0; /* If div by 100 *and* 400 is not leap. */
else return 1; /* If div by 100 and not by 400 is leap. */
}
void nolocks_localtime(struct tm *tmp, time_t t, time_t tz, int dst) {
const time_t secs_min = 60;
const time_t secs_hour = 3600;
const time_t secs_day = 3600*24;
t -= tz; /* Adjust for timezone. */
t += 3600*dst; /* Adjust for daylight time. */
time_t days = t / secs_day; /* Days passed since epoch. */
time_t seconds = t % secs_day; /* Remaining seconds. */
tmp->tm_isdst = dst;
tmp->tm_hour = seconds / secs_hour;
tmp->tm_min = (seconds % secs_hour) / secs_min;
tmp->tm_sec = (seconds % secs_hour) % secs_min;
/* 1/1/1970 was a Thursday, that is, day 4 from the POV of the tm structure
* where sunday = 0, so to calculate the day of the week we have to add 4
* and take the modulo by 7. */
tmp->tm_wday = (days+4)%7;
/* Calculate the current year. */
tmp->tm_year = 1970;
while(1) {
/* Leap years have one day more. */
time_t days_this_year = 365 + is_leap_year(tmp->tm_year);
if (days_this_year > days) break;
days -= days_this_year;
tmp->tm_year++;
}
tmp->tm_yday = days; /* Number of day of the current year. */
/* We need to calculate in which month and day of the month we are. To do
* so we need to skip days according to how many days there are in each
* month, and adjust for the leap year that has one more day in February. */
int mdays[12] = {31, 28, 31, 30, 31, 30, 31, 31, 30, 31, 30, 31};
mdays[1] += is_leap_year(tmp->tm_year);
tmp->tm_mon = 0;
while(days >= mdays[tmp->tm_mon]) {
days -= mdays[tmp->tm_mon];
tmp->tm_mon++;
}
tmp->tm_mday = days+1; /* Add 1 since our 'days' is zero-based. */
tmp->tm_year -= 1900; /* Surprisingly tm_year is year-1900. */
}
#ifdef LOCALTIME_TEST_MAIN
#include <stdio.h>
int main(void) {
/* Obtain timezone and daylight info. */
tzset(); /* Now 'timezome' global is populated. */
time_t t = time(NULL);
struct tm *aux = localtime(&t);
int daylight_active = aux->tm_isdst;
struct tm tm;
char buf[1024];
nolocks_localtime(&tm,t,timezone,daylight_active);
strftime(buf,sizeof(buf),"%d %b %H:%M:%S",&tm);
printf("[timezone: %d, dl: %d] %s\n", (int)timezone, (int)daylight_active, buf);
}
#endif
...@@ -2239,6 +2239,9 @@ int RM_HashSet(RedisModuleKey *key, int flags, ...) { ...@@ -2239,6 +2239,9 @@ int RM_HashSet(RedisModuleKey *key, int flags, ...) {
* to avoid a useless copy. */ * to avoid a useless copy. */
if (flags & REDISMODULE_HASH_CFIELDS) if (flags & REDISMODULE_HASH_CFIELDS)
low_flags |= HASH_SET_TAKE_FIELD; low_flags |= HASH_SET_TAKE_FIELD;
robj *argv[2] = {field,value};
hashTypeTryConversion(key->value,argv,0,1);
updated += hashTypeSet(key->value, field->ptr, value->ptr, low_flags); updated += hashTypeSet(key->value, field->ptr, value->ptr, low_flags);
/* If CFIELDS is active, SDS string ownership is now of hashTypeSet(), /* If CFIELDS is active, SDS string ownership is now of hashTypeSet(),
...@@ -2709,9 +2712,9 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch ...@@ -2709,9 +2712,9 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
sds proto = sdsnewlen(c->buf,c->bufpos); sds proto = sdsnewlen(c->buf,c->bufpos);
c->bufpos = 0; c->bufpos = 0;
while(listLength(c->reply)) { while(listLength(c->reply)) {
sds o = listNodeValue(listFirst(c->reply)); clientReplyBlock *o = listNodeValue(listFirst(c->reply));
proto = sdscatsds(proto,o); proto = sdscatlen(proto,o->buf,o->used);
listDelNode(c->reply,listFirst(c->reply)); listDelNode(c->reply,listFirst(c->reply));
} }
reply = moduleCreateCallReplyFromProto(ctx,proto); reply = moduleCreateCallReplyFromProto(ctx,proto);
...@@ -3396,7 +3399,7 @@ void RM_LogRaw(RedisModule *module, const char *levelstr, const char *fmt, va_li ...@@ -3396,7 +3399,7 @@ void RM_LogRaw(RedisModule *module, const char *levelstr, const char *fmt, va_li
* *
* If the specified log level is invalid, verbose is used by default. * If the specified log level is invalid, verbose is used by default.
* There is a fixed limit to the length of the log line this function is able * There is a fixed limit to the length of the log line this function is able
* to emit, this limti is not specified but is guaranteed to be more than * to emit, this limit is not specified but is guaranteed to be more than
* a few lines of text. * a few lines of text.
*/ */
void RM_Log(RedisModuleCtx *ctx, const char *levelstr, const char *fmt, ...) { void RM_Log(RedisModuleCtx *ctx, const char *levelstr, const char *fmt, ...) {
...@@ -3827,7 +3830,7 @@ void moduleReleaseGIL(void) { ...@@ -3827,7 +3830,7 @@ void moduleReleaseGIL(void) {
* *
* Notification callback gets executed with a redis context that can not be * Notification callback gets executed with a redis context that can not be
* used to send anything to the client, and has the db number where the event * used to send anything to the client, and has the db number where the event
* occured as its selected db number. * occurred as its selected db number.
* *
* Notice that it is not necessary to enable norifications in redis.conf for * Notice that it is not necessary to enable norifications in redis.conf for
* module notifications to work. * module notifications to work.
...@@ -3884,7 +3887,7 @@ void moduleNotifyKeyspaceEvent(int type, const char *event, robj *key, int dbid) ...@@ -3884,7 +3887,7 @@ void moduleNotifyKeyspaceEvent(int type, const char *event, robj *key, int dbid)
} }
} }
/* Unsubscribe any notification subscirbers this module has upon unloading */ /* Unsubscribe any notification subscribers this module has upon unloading */
void moduleUnsubscribeNotifications(RedisModule *module) { void moduleUnsubscribeNotifications(RedisModule *module) {
listIter li; listIter li;
listNode *ln; listNode *ln;
...@@ -4362,7 +4365,7 @@ void moduleInitModulesSystem(void) { ...@@ -4362,7 +4365,7 @@ void moduleInitModulesSystem(void) {
* because the server must be fully initialized before loading modules. * because the server must be fully initialized before loading modules.
* *
* The function aborts the server on errors, since to start with missing * The function aborts the server on errors, since to start with missing
* modules is not considered sane: clients may rely on the existance of * modules is not considered sane: clients may rely on the existence of
* given commands, loading AOF also may need some modules to exist, and * given commands, loading AOF also may need some modules to exist, and
* if this instance is a slave, it must understand commands from master. */ * if this instance is a slave, it must understand commands from master. */
void moduleLoadFromQueue(void) { void moduleLoadFromQueue(void) {
...@@ -4499,7 +4502,15 @@ int moduleUnload(sds name) { ...@@ -4499,7 +4502,15 @@ int moduleUnload(sds name) {
* MODULE LOAD <path> [args...] */ * MODULE LOAD <path> [args...] */
void moduleCommand(client *c) { void moduleCommand(client *c) {
char *subcmd = c->argv[1]->ptr; char *subcmd = c->argv[1]->ptr;
if (c->argc == 2 && !strcasecmp(subcmd,"help")) {
const char *help[] = {
"LIST -- Return a list of loaded modules.",
"LOAD <path> [arg ...] -- Load a module library from <path>.",
"UNLOAD <name> -- Unload a module.",
NULL
};
addReplyHelp(c, help);
} else
if (!strcasecmp(subcmd,"load") && c->argc >= 3) { if (!strcasecmp(subcmd,"load") && c->argc >= 3) {
robj **argv = NULL; robj **argv = NULL;
int argc = 0; int argc = 0;
...@@ -4548,7 +4559,8 @@ void moduleCommand(client *c) { ...@@ -4548,7 +4559,8 @@ void moduleCommand(client *c) {
} }
dictReleaseIterator(di); dictReleaseIterator(di);
} else { } else {
addReply(c,shared.syntaxerr); addReplySubcommandSyntaxError(c);
return;
} }
} }
......
# gendoc.rb -- Converts the top-comments inside module.c to modules API # gendoc.rb -- Converts the top-comments inside module.c to modules API
# reference documentaiton in markdown format. # reference documentation in markdown format.
# Convert the C comment to markdown # Convert the C comment to markdown
def markdown(s) def markdown(s)
......
...@@ -56,11 +56,14 @@ size_t getStringObjectSdsUsedMemory(robj *o) { ...@@ -56,11 +56,14 @@ size_t getStringObjectSdsUsedMemory(robj *o) {
/* Client.reply list dup and free methods. */ /* Client.reply list dup and free methods. */
void *dupClientReplyValue(void *o) { void *dupClientReplyValue(void *o) {
return sdsdup(o); clientReplyBlock *old = o;
clientReplyBlock *buf = zmalloc(sizeof(clientReplyBlock) + old->size);
memcpy(buf, o, sizeof(clientReplyBlock) + old->size);
return buf;
} }
void freeClientReplyValue(void *o) { void freeClientReplyValue(void *o) {
sdsfree(o); zfree(o);
} }
int listMatchObjects(void *a, void *b) { int listMatchObjects(void *a, void *b) {
...@@ -75,6 +78,8 @@ void linkClient(client *c) { ...@@ -75,6 +78,8 @@ void linkClient(client *c) {
* this way removing the client in unlinkClient() will not require * this way removing the client in unlinkClient() will not require
* a linear scan, but just a constant time operation. */ * a linear scan, but just a constant time operation. */
c->client_list_node = listLast(server.clients); c->client_list_node = listLast(server.clients);
uint64_t id = htonu64(c->id);
raxInsert(server.clients_index,(unsigned char*)&id,sizeof(id),c,NULL);
} }
client *createClient(int fd) { client *createClient(int fd) {
...@@ -138,6 +143,7 @@ client *createClient(int fd) { ...@@ -138,6 +143,7 @@ client *createClient(int fd) {
c->bpop.target = NULL; c->bpop.target = NULL;
c->bpop.xread_group = NULL; c->bpop.xread_group = NULL;
c->bpop.xread_consumer = NULL; c->bpop.xread_consumer = NULL;
c->bpop.xread_group_noack = 0;
c->bpop.numreplicas = 0; c->bpop.numreplicas = 0;
c->bpop.reploffset = 0; c->bpop.reploffset = 0;
c->woff = 0; c->woff = 0;
...@@ -237,25 +243,35 @@ int _addReplyToBuffer(client *c, const char *s, size_t len) { ...@@ -237,25 +243,35 @@ int _addReplyToBuffer(client *c, const char *s, size_t len) {
void _addReplyStringToList(client *c, const char *s, size_t len) { void _addReplyStringToList(client *c, const char *s, size_t len) {
if (c->flags & CLIENT_CLOSE_AFTER_REPLY) return; if (c->flags & CLIENT_CLOSE_AFTER_REPLY) return;
if (listLength(c->reply) == 0) { listNode *ln = listLast(c->reply);
sds node = sdsnewlen(s,len); clientReplyBlock *tail = ln? listNodeValue(ln): NULL;
listAddNodeTail(c->reply,node);
c->reply_bytes += len; /* Note that 'tail' may be NULL even if we have a tail node, becuase when
} else { * addDeferredMultiBulkLength() is used, it sets a dummy node to NULL just
listNode *ln = listLast(c->reply); * fo fill it later, when the size of the bulk length is set. */
sds tail = listNodeValue(ln);
/* Append to tail string when possible. */
/* Append to this object when possible. If tail == NULL it was if (tail) {
* set via addDeferredMultiBulkLength(). */ /* Copy the part we can fit into the tail, and leave the rest for a
if (tail && sdslen(tail)+len <= PROTO_REPLY_CHUNK_BYTES) { * new node */
tail = sdscatlen(tail,s,len); size_t avail = tail->size - tail->used;
listNodeValue(ln) = tail; size_t copy = avail >= len? len: avail;
c->reply_bytes += len; memcpy(tail->buf + tail->used, s, copy);
} else { tail->used += copy;
sds node = sdsnewlen(s,len); s += copy;
listAddNodeTail(c->reply,node); len -= copy;
c->reply_bytes += len; }
} if (len) {
/* Create a new node, make sure it is allocated to at
* least PROTO_REPLY_CHUNK_BYTES */
size_t size = len < PROTO_REPLY_CHUNK_BYTES? PROTO_REPLY_CHUNK_BYTES: len;
tail = zmalloc(size + sizeof(clientReplyBlock));
/* take over the allocation's internal fragmentation */
tail->size = zmalloc_usable(tail) - sizeof(clientReplyBlock);
tail->used = len;
memcpy(tail->buf, s, len);
listAddNodeTail(c->reply, tail);
c->reply_bytes += tail->size;
} }
asyncCloseClientOnOutputBufferLimitReached(c); asyncCloseClientOnOutputBufferLimitReached(c);
} }
...@@ -326,11 +342,30 @@ void addReplyErrorLength(client *c, const char *s, size_t len) { ...@@ -326,11 +342,30 @@ void addReplyErrorLength(client *c, const char *s, size_t len) {
if (!len || s[0] != '-') addReplyString(c,"-ERR ",5); if (!len || s[0] != '-') addReplyString(c,"-ERR ",5);
addReplyString(c,s,len); addReplyString(c,s,len);
addReplyString(c,"\r\n",2); addReplyString(c,"\r\n",2);
if (c->flags & CLIENT_MASTER) {
/* Sometimes it could be normal that a slave replies to a master with
* an error and this function gets called. Actually the error will never
* be sent because addReply*() against master clients has no effect...
* A notable example is:
*
* EVAL 'redis.call("incr",KEYS[1]); redis.call("nonexisting")' 1 x
*
* Where the master must propagate the first change even if the second
* will produce an error. However it is useful to log such events since
* they are rare and may hint at errors in a script or a bug in Redis. */
if (c->flags & (CLIENT_MASTER|CLIENT_SLAVE)) {
char* to = c->flags & CLIENT_MASTER? "master": "slave";
char* from = c->flags & CLIENT_MASTER? "slave": "master";
char *cmdname = c->lastcmd ? c->lastcmd->name : "<unknown>"; char *cmdname = c->lastcmd ? c->lastcmd->name : "<unknown>";
serverLog(LL_WARNING,"== CRITICAL == This slave is sending an error " serverLog(LL_WARNING,"== CRITICAL == This %s is sending an error "
"to its master: '%s' after processing the command " "to its %s: '%s' after processing the command "
"'%s'", s, cmdname); "'%s'", from, to, s, cmdname);
/* Here we want to panic because when a master is sending an
* error to some slave in the context of replication, this can
* only create some kind of offset or data desynchronization. Better
* to catch it ASAP and crash instead of continuing. */
if (c->flags & CLIENT_SLAVE)
serverPanic("Continuing is unsafe: replication protocol violation.");
} }
} }
...@@ -387,26 +422,41 @@ void *addDeferredMultiBulkLength(client *c) { ...@@ -387,26 +422,41 @@ void *addDeferredMultiBulkLength(client *c) {
/* Populate the length object and try gluing it to the next chunk. */ /* Populate the length object and try gluing it to the next chunk. */
void setDeferredMultiBulkLength(client *c, void *node, long length) { void setDeferredMultiBulkLength(client *c, void *node, long length) {
listNode *ln = (listNode*)node; listNode *ln = (listNode*)node;
sds len, next; clientReplyBlock *next;
char lenstr[128];
size_t lenstr_len = sprintf(lenstr, "*%ld\r\n", length);
/* Abort when *node is NULL: when the client should not accept writes /* Abort when *node is NULL: when the client should not accept writes
* we return NULL in addDeferredMultiBulkLength() */ * we return NULL in addDeferredMultiBulkLength() */
if (node == NULL) return; if (node == NULL) return;
serverAssert(!listNodeValue(ln));
len = sdscatprintf(sdsnewlen("*",1),"%ld\r\n",length);
listNodeValue(ln) = len; /* Normally we fill this dummy NULL node, added by addDeferredMultiBulkLength(),
c->reply_bytes += sdslen(len); * with a new buffer structure containing the protocol needed to specify
if (ln->next != NULL) { * the length of the array following. However sometimes when there is
next = listNodeValue(ln->next); * little memory to move, we may instead remove this NULL node, and prefix
* our protocol in the node immediately after to it, in order to save a
/* Only glue when the next node is non-NULL (an sds in this case) */ * write(2) syscall later. Conditions needed to do it:
if (next != NULL) { *
len = sdscatsds(len,next); * - The next node is non-NULL,
listDelNode(c->reply,ln->next); * - It has enough room already allocated
listNodeValue(ln) = len; * - And not too large (avoid large memmove) */
/* No need to update c->reply_bytes: we are just moving the same if (ln->next != NULL && (next = listNodeValue(ln->next)) &&
* amount of bytes from one node to another. */ next->size - next->used >= lenstr_len &&
} next->used < PROTO_REPLY_CHUNK_BYTES * 4) {
memmove(next->buf + lenstr_len, next->buf, next->used);
memcpy(next->buf, lenstr, lenstr_len);
next->used += lenstr_len;
listDelNode(c->reply,ln);
} else {
/* Create a new node */
clientReplyBlock *buf = zmalloc(lenstr_len + sizeof(clientReplyBlock));
/* Take over the allocation's internal fragmentation */
buf->size = zmalloc_usable(buf) - sizeof(clientReplyBlock);
buf->used = lenstr_len;
memcpy(buf->buf, lenstr, lenstr_len);
listNodeValue(ln) = buf;
c->reply_bytes += buf->size;
} }
asyncCloseClientOnOutputBufferLimitReached(c); asyncCloseClientOnOutputBufferLimitReached(c);
} }
...@@ -560,11 +610,24 @@ void addReplyHelp(client *c, const char **help) { ...@@ -560,11 +610,24 @@ void addReplyHelp(client *c, const char **help) {
setDeferredMultiBulkLength(c,blenp,blen); setDeferredMultiBulkLength(c,blenp,blen);
} }
/* Add a suggestive error reply.
* This function is typically invoked by from commands that support
* subcommands in response to an unknown subcommand or argument error. */
void addReplySubcommandSyntaxError(client *c) {
sds cmd = sdsnew((char*) c->argv[0]->ptr);
sdstoupper(cmd);
addReplyErrorFormat(c,
"Unknown subcommand or wrong number of arguments for '%s'. Try %s HELP.",
(char*)c->argv[1]->ptr,cmd);
sdsfree(cmd);
}
/* Copy 'src' client output buffers into 'dst' client output buffers. /* Copy 'src' client output buffers into 'dst' client output buffers.
* The function takes care of freeing the old output buffers of the * The function takes care of freeing the old output buffers of the
* destination client. */ * destination client. */
void copyClientOutputBuffer(client *dst, client *src) { void copyClientOutputBuffer(client *dst, client *src) {
listRelease(dst->reply); listRelease(dst->reply);
dst->sentlen = 0;
dst->reply = listDup(src->reply); dst->reply = listDup(src->reply);
memcpy(dst->buf,src->buf,src->bufpos); memcpy(dst->buf,src->buf,src->bufpos);
dst->bufpos = src->bufpos; dst->bufpos = src->bufpos;
...@@ -720,6 +783,8 @@ void unlinkClient(client *c) { ...@@ -720,6 +783,8 @@ void unlinkClient(client *c) {
if (c->fd != -1) { if (c->fd != -1) {
/* Remove from the list of active clients. */ /* Remove from the list of active clients. */
if (c->client_list_node) { if (c->client_list_node) {
uint64_t id = htonu64(c->id);
raxRemove(server.clients_index,(unsigned char*)&id,sizeof(id),NULL);
listDelNode(server.clients,c->client_list_node); listDelNode(server.clients,c->client_list_node);
c->client_list_node = NULL; c->client_list_node = NULL;
} }
...@@ -864,12 +929,21 @@ void freeClientsInAsyncFreeQueue(void) { ...@@ -864,12 +929,21 @@ void freeClientsInAsyncFreeQueue(void) {
} }
} }
/* Return a client by ID, or NULL if the client ID is not in the set
* of registered clients. Note that "fake clients", created with -1 as FD,
* are not registered clients. */
client *lookupClientByID(uint64_t id) {
id = htonu64(id);
client *c = raxFind(server.clients_index,(unsigned char*)&id,sizeof(id));
return (c == raxNotFound) ? NULL : c;
}
/* Write data in output buffers to client. Return C_OK if the client /* Write data in output buffers to client. Return C_OK if the client
* is still valid after the call, C_ERR if it was freed. */ * is still valid after the call, C_ERR if it was freed. */
int writeToClient(int fd, client *c, int handler_installed) { int writeToClient(int fd, client *c, int handler_installed) {
ssize_t nwritten = 0, totwritten = 0; ssize_t nwritten = 0, totwritten = 0;
size_t objlen; size_t objlen;
sds o; clientReplyBlock *o;
while(clientHasPendingReplies(c)) { while(clientHasPendingReplies(c)) {
if (c->bufpos > 0) { if (c->bufpos > 0) {
...@@ -886,23 +960,24 @@ int writeToClient(int fd, client *c, int handler_installed) { ...@@ -886,23 +960,24 @@ int writeToClient(int fd, client *c, int handler_installed) {
} }
} else { } else {
o = listNodeValue(listFirst(c->reply)); o = listNodeValue(listFirst(c->reply));
objlen = sdslen(o); objlen = o->used;
if (objlen == 0) { if (objlen == 0) {
c->reply_bytes -= o->size;
listDelNode(c->reply,listFirst(c->reply)); listDelNode(c->reply,listFirst(c->reply));
continue; continue;
} }
nwritten = write(fd, o + c->sentlen, objlen - c->sentlen); nwritten = write(fd, o->buf + c->sentlen, objlen - c->sentlen);
if (nwritten <= 0) break; if (nwritten <= 0) break;
c->sentlen += nwritten; c->sentlen += nwritten;
totwritten += nwritten; totwritten += nwritten;
/* If we fully sent the object on head go to the next one */ /* If we fully sent the object on head go to the next one */
if (c->sentlen == objlen) { if (c->sentlen == objlen) {
c->reply_bytes -= o->size;
listDelNode(c->reply,listFirst(c->reply)); listDelNode(c->reply,listFirst(c->reply));
c->sentlen = 0; c->sentlen = 0;
c->reply_bytes -= objlen;
/* If there are no longer objects in the list, we expect /* If there are no longer objects in the list, we expect
* the count of reply bytes to be exactly zero. */ * the count of reply bytes to be exactly zero. */
if (listLength(c->reply) == 0) if (listLength(c->reply) == 0)
...@@ -1039,7 +1114,7 @@ void resetClient(client *c) { ...@@ -1039,7 +1114,7 @@ void resetClient(client *c) {
* with the error and close the connection. */ * with the error and close the connection. */
int processInlineBuffer(client *c) { int processInlineBuffer(client *c) {
char *newline; char *newline;
int argc, j; int argc, j, linefeed_chars = 1;
sds *argv, aux; sds *argv, aux;
size_t querylen; size_t querylen;
...@@ -1057,7 +1132,7 @@ int processInlineBuffer(client *c) { ...@@ -1057,7 +1132,7 @@ int processInlineBuffer(client *c) {
/* Handle the \r\n case. */ /* Handle the \r\n case. */
if (newline && newline != c->querybuf && *(newline-1) == '\r') if (newline && newline != c->querybuf && *(newline-1) == '\r')
newline--; newline--, linefeed_chars++;
/* Split the input buffer up to the \r\n */ /* Split the input buffer up to the \r\n */
querylen = newline-(c->querybuf); querylen = newline-(c->querybuf);
...@@ -1077,7 +1152,7 @@ int processInlineBuffer(client *c) { ...@@ -1077,7 +1152,7 @@ int processInlineBuffer(client *c) {
c->repl_ack_time = server.unixtime; c->repl_ack_time = server.unixtime;
/* Leave data after the first line of the query in the buffer */ /* Leave data after the first line of the query in the buffer */
sdsrange(c->querybuf,querylen+2,-1); sdsrange(c->querybuf,querylen+linefeed_chars,-1);
/* Setup argv array on client structure */ /* Setup argv array on client structure */
if (argc) { if (argc) {
...@@ -1493,6 +1568,7 @@ sds catClientInfoString(sds s, client *client) { ...@@ -1493,6 +1568,7 @@ sds catClientInfoString(sds s, client *client) {
*p++ = 'S'; *p++ = 'S';
} }
if (client->flags & CLIENT_MASTER) *p++ = 'M'; if (client->flags & CLIENT_MASTER) *p++ = 'M';
if (client->flags & CLIENT_PUBSUB) *p++ = 'P';
if (client->flags & CLIENT_MULTI) *p++ = 'x'; if (client->flags & CLIENT_MULTI) *p++ = 'x';
if (client->flags & CLIENT_BLOCKED) *p++ = 'b'; if (client->flags & CLIENT_BLOCKED) *p++ = 'b';
if (client->flags & CLIENT_DIRTY_CAS) *p++ = 'd'; if (client->flags & CLIENT_DIRTY_CAS) *p++ = 'd';
...@@ -1531,7 +1607,7 @@ sds catClientInfoString(sds s, client *client) { ...@@ -1531,7 +1607,7 @@ sds catClientInfoString(sds s, client *client) {
client->lastcmd ? client->lastcmd->name : "NULL"); client->lastcmd ? client->lastcmd->name : "NULL");
} }
sds getAllClientsInfoString(void) { sds getAllClientsInfoString(int type) {
listNode *ln; listNode *ln;
listIter li; listIter li;
client *client; client *client;
...@@ -1540,6 +1616,7 @@ sds getAllClientsInfoString(void) { ...@@ -1540,6 +1616,7 @@ sds getAllClientsInfoString(void) {
listRewind(server.clients,&li); listRewind(server.clients,&li);
while ((ln = listNext(&li)) != NULL) { while ((ln = listNext(&li)) != NULL) {
client = listNodeValue(ln); client = listNodeValue(ln);
if (type != -1 && getClientType(client) != type) continue;
o = catClientInfoString(o,client); o = catClientInfoString(o,client);
o = sdscatlen(o,"\n",1); o = sdscatlen(o,"\n",1);
} }
...@@ -1553,22 +1630,40 @@ void clientCommand(client *c) { ...@@ -1553,22 +1630,40 @@ void clientCommand(client *c) {
if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"help")) { if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"help")) {
const char *help[] = { const char *help[] = {
"getname -- Return the name of the current connection.", "id -- Return the ID of the current connection.",
"kill <ip:port> -- Kill connection made from <ip:port>.", "getname -- Return the name of the current connection.",
"kill <ip:port> -- Kill connection made from <ip:port>.",
"kill <option> <value> [option value ...] -- Kill connections. Options are:", "kill <option> <value> [option value ...] -- Kill connections. Options are:",
" addr <ip:port> -- Kill connection made from <ip:port>.", " addr <ip:port> -- Kill connection made from <ip:port>",
" type (normal|master|slave|pubsub) -- Kill connections by type.", " type (normal|master|slave|pubsub) -- Kill connections by type.",
" skipme (yes|no) -- Skip killing current connection (default: yes).", " skipme (yes|no) -- Skip killing current connection (default: yes).",
"list -- Return information about client connections.", "list [options ...] -- Return information about client connections. Options:",
"pause <timeout> -- Suspend all Redis clients for <timout> milliseconds.", " type (normal|master|slave|pubsub) -- Return clients of specified type.",
"reply (on|off|skip) -- Control the replies sent to the current connection.", "pause <timeout> -- Suspend all Redis clients for <timout> milliseconds.",
"setname <name> -- Assign the name <name> to the current connection.", "reply (on|off|skip) -- Control the replies sent to the current connection.",
"setname <name> -- Assign the name <name> to the current connection.",
"unblock <clientid> [TIMEOUT|ERROR] -- Unblock the specified blocked client.",
NULL NULL
}; };
addReplyHelp(c, help); addReplyHelp(c, help);
} else if (!strcasecmp(c->argv[1]->ptr,"list") && c->argc == 2) { } else if (!strcasecmp(c->argv[1]->ptr,"id") && c->argc == 2) {
/* CLIENT ID */
addReplyLongLong(c,c->id);
} else if (!strcasecmp(c->argv[1]->ptr,"list")) {
/* CLIENT LIST */ /* CLIENT LIST */
sds o = getAllClientsInfoString(); int type = -1;
if (c->argc == 4 && !strcasecmp(c->argv[2]->ptr,"type")) {
type = getClientTypeByName(c->argv[3]->ptr);
if (type == -1) {
addReplyErrorFormat(c,"Unknown client type '%s'",
(char*) c->argv[3]->ptr);
return;
}
} else if (c->argc != 2) {
addReply(c,shared.syntaxerr);
return;
}
sds o = getAllClientsInfoString(type);
addReplyBulkCBuffer(c,o,sdslen(o)); addReplyBulkCBuffer(c,o,sdslen(o));
sdsfree(o); sdsfree(o);
} else if (!strcasecmp(c->argv[1]->ptr,"reply") && c->argc == 3) { } else if (!strcasecmp(c->argv[1]->ptr,"reply") && c->argc == 3) {
...@@ -1671,6 +1766,38 @@ NULL ...@@ -1671,6 +1766,38 @@ NULL
/* If this client has to be closed, flag it as CLOSE_AFTER_REPLY /* If this client has to be closed, flag it as CLOSE_AFTER_REPLY
* only after we queued the reply to its output buffers. */ * only after we queued the reply to its output buffers. */
if (close_this_client) c->flags |= CLIENT_CLOSE_AFTER_REPLY; if (close_this_client) c->flags |= CLIENT_CLOSE_AFTER_REPLY;
} else if (!strcasecmp(c->argv[1]->ptr,"unblock") && (c->argc == 3 ||
c->argc == 4))
{
/* CLIENT UNBLOCK <id> [timeout|error] */
long long id;
int unblock_error = 0;
if (c->argc == 4) {
if (!strcasecmp(c->argv[3]->ptr,"timeout")) {
unblock_error = 0;
} else if (!strcasecmp(c->argv[3]->ptr,"error")) {
unblock_error = 1;
} else {
addReplyError(c,
"CLIENT UNBLOCK reason should be TIMEOUT or ERROR");
return;
}
}
if (getLongLongFromObjectOrReply(c,c->argv[2],&id,NULL)
!= C_OK) return;
struct client *target = lookupClientByID(id);
if (target && target->flags & CLIENT_BLOCKED) {
if (unblock_error)
addReplyError(target,
"-UNBLOCKED client unblocked via CLIENT UNBLOCK");
else
replyToBlockedClientTimedOut(target);
unblockClient(target);
addReply(c,shared.cone);
} else {
addReply(c,shared.czero);
}
} else if (!strcasecmp(c->argv[1]->ptr,"setname") && c->argc == 3) { } else if (!strcasecmp(c->argv[1]->ptr,"setname") && c->argc == 3) {
int j, len = sdslen(c->argv[2]->ptr); int j, len = sdslen(c->argv[2]->ptr);
char *p = c->argv[2]->ptr; char *p = c->argv[2]->ptr;
...@@ -1821,10 +1948,7 @@ void rewriteClientCommandArgument(client *c, int i, robj *newval) { ...@@ -1821,10 +1948,7 @@ void rewriteClientCommandArgument(client *c, int i, robj *newval) {
* the caller wishes. The main usage of this function currently is * the caller wishes. The main usage of this function currently is
* enforcing the client output length limits. */ * enforcing the client output length limits. */
unsigned long getClientOutputBufferMemoryUsage(client *c) { unsigned long getClientOutputBufferMemoryUsage(client *c) {
unsigned long list_item_size = sizeof(listNode)+5; unsigned long list_item_size = sizeof(listNode) + sizeof(clientReplyBlock);
/* The +5 above means we assume an sds16 hdr, may not be true
* but is not going to be a problem. */
return c->reply_bytes + (list_item_size*listLength(c->reply)); return c->reply_bytes + (list_item_size*listLength(c->reply));
} }
......
...@@ -29,8 +29,8 @@ ...@@ -29,8 +29,8 @@
#include "server.h" #include "server.h"
/* This file implements keyspace events notification via Pub/Sub ad /* This file implements keyspace events notification via Pub/Sub and
* described at http://redis.io/topics/keyspace-events. */ * described at https://redis.io/topics/notifications. */
/* Turn a string representing notification classes into an integer /* Turn a string representing notification classes into an integer
* representing notification classes flags xored. * representing notification classes flags xored.
......
...@@ -123,9 +123,25 @@ robj *createStringObject(const char *ptr, size_t len) { ...@@ -123,9 +123,25 @@ robj *createStringObject(const char *ptr, size_t len) {
return createRawStringObject(ptr,len); return createRawStringObject(ptr,len);
} }
robj *createStringObjectFromLongLong(long long value) { /* Create a string object from a long long value. When possible returns a
* shared integer object, or at least an integer encoded one.
*
* If valueobj is non zero, the function avoids returning a a shared
* integer, because the object is going to be used as value in the Redis key
* space (for instance when the INCR command is used), so we want LFU/LRU
* values specific for each key. */
robj *createStringObjectFromLongLongWithOptions(long long value, int valueobj) {
robj *o; robj *o;
if (value >= 0 && value < OBJ_SHARED_INTEGERS) {
if (server.maxmemory == 0 ||
!(server.maxmemory_policy & MAXMEMORY_FLAG_NO_SHARED_INTEGERS))
{
/* If the maxmemory policy permits, we can still return shared integers
* even if valueobj is true. */
valueobj = 0;
}
if (value >= 0 && value < OBJ_SHARED_INTEGERS && valueobj == 0) {
incrRefCount(shared.integers[value]); incrRefCount(shared.integers[value]);
o = shared.integers[value]; o = shared.integers[value];
} else { } else {
...@@ -140,6 +156,20 @@ robj *createStringObjectFromLongLong(long long value) { ...@@ -140,6 +156,20 @@ robj *createStringObjectFromLongLong(long long value) {
return o; return o;
} }
/* Wrapper for createStringObjectFromLongLongWithOptions() always demanding
* to create a shared object if possible. */
robj *createStringObjectFromLongLong(long long value) {
return createStringObjectFromLongLongWithOptions(value,0);
}
/* Wrapper for createStringObjectFromLongLongWithOptions() avoiding a shared
* object when LFU/LRU info are needed, that is, when the object is used
* as a value in the key space, and Redis is configured to evict based on
* LFU/LRU. */
robj *createStringObjectFromLongLongForValue(long long value) {
return createStringObjectFromLongLongWithOptions(value,1);
}
/* Create a string object from a long double. If humanfriendly is non-zero /* Create a string object from a long double. If humanfriendly is non-zero
* it does not use exponential format and trims trailing zeroes at the end, * it does not use exponential format and trims trailing zeroes at the end,
* however this results in loss of precision. Otherwise exp format is used * however this results in loss of precision. Otherwise exp format is used
...@@ -715,7 +745,7 @@ char *strEncoding(int encoding) { ...@@ -715,7 +745,7 @@ char *strEncoding(int encoding) {
* size of a radix tree that is used to store Stream IDs. * size of a radix tree that is used to store Stream IDs.
* *
* Note: to guess the size of the radix tree is not trivial, so we * Note: to guess the size of the radix tree is not trivial, so we
* approximate it considering 128 bytes of data overhead for each * approximate it considering 16 bytes of data overhead for each
* key (the ID), and then adding the number of bare nodes, plus some * key (the ID), and then adding the number of bare nodes, plus some
* overhead due by the data and child pointers. This secret recipe * overhead due by the data and child pointers. This secret recipe
* was obtained by checking the average radix tree created by real * was obtained by checking the average radix tree created by real
...@@ -874,6 +904,7 @@ size_t objectComputeSize(robj *o, size_t sample_size) { ...@@ -874,6 +904,7 @@ size_t objectComputeSize(robj *o, size_t sample_size) {
* structures and the PEL memory usage. */ * structures and the PEL memory usage. */
raxIterator cri; raxIterator cri;
raxStart(&cri,cg->consumers); raxStart(&cri,cg->consumers);
raxSeek(&cri,"^",NULL,0);
while(raxNext(&cri)) { while(raxNext(&cri)) {
streamConsumer *consumer = cri.data; streamConsumer *consumer = cri.data;
asize += sizeof(*consumer); asize += sizeof(*consumer);
...@@ -1136,6 +1167,32 @@ sds getMemoryDoctorReport(void) { ...@@ -1136,6 +1167,32 @@ sds getMemoryDoctorReport(void) {
return s; return s;
} }
/* Set the object LRU/LFU depending on server.maxmemory_policy.
* The lfu_freq arg is only relevant if policy is MAXMEMORY_FLAG_LFU.
* The lru_idle and lru_clock args are only relevant if policy
* is MAXMEMORY_FLAG_LRU.
* Either or both of them may be <0, in that case, nothing is set. */
void objectSetLRUOrLFU(robj *val, long long lfu_freq, long long lru_idle,
long long lru_clock) {
if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
if (lfu_freq >= 0) {
serverAssert(lfu_freq <= 255);
val->lru = (LFUGetTimeInMinutes()<<8) | lfu_freq;
}
} else if (lru_idle >= 0) {
/* Serialized LRU idle time is in seconds. Scale
* according to the LRU clock resolution this Redis
* instance was compiled with (normally 1000 ms, so the
* below statement will expand to lru_idle*1000/1000. */
lru_idle = lru_idle*1000/LRU_CLOCK_RESOLUTION;
val->lru = lru_clock - lru_idle;
/* If the lru field overflows (since LRU it is a wrapping
* clock), the best we can do is to provide the maximum
* representable idle time. */
if (val->lru < 0) val->lru = lru_clock+1;
}
}
/* ======================= The OBJECT and MEMORY commands =================== */ /* ======================= The OBJECT and MEMORY commands =================== */
/* This is a helper function for the OBJECT command. We need to lookup keys /* This is a helper function for the OBJECT command. We need to lookup keys
...@@ -1197,7 +1254,7 @@ NULL ...@@ -1197,7 +1254,7 @@ NULL
* when the key is read or overwritten. */ * when the key is read or overwritten. */
addReplyLongLong(c,LFUDecrAndReturn(o)); addReplyLongLong(c,LFUDecrAndReturn(o));
} else { } else {
addReplyErrorFormat(c, "Unknown subcommand or wrong number of arguments for '%s'. Try OBJECT help", (char *)c->argv[1]->ptr); addReplySubcommandSyntaxError(c);
} }
} }
......
...@@ -327,9 +327,9 @@ void publishCommand(client *c) { ...@@ -327,9 +327,9 @@ void publishCommand(client *c) {
void pubsubCommand(client *c) { void pubsubCommand(client *c) {
if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"help")) { if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"help")) {
const char *help[] = { const char *help[] = {
"channels [<pattern>] -- Return the currently active channels matching a pattern (default: all).", "CHANNELS [<pattern>] -- Return the currently active channels matching a pattern (default: all).",
"numpat -- Return number of subscriptions to patterns.", "NUMPAT -- Return number of subscriptions to patterns.",
"numsub [channel-1 .. channel-N] -- Returns the number of subscribers for the specified channels (excluding patterns, default: none).", "NUMSUB [channel-1 .. channel-N] -- Returns the number of subscribers for the specified channels (excluding patterns, default: none).",
NULL NULL
}; };
addReplyHelp(c, help); addReplyHelp(c, help);
...@@ -372,7 +372,6 @@ NULL ...@@ -372,7 +372,6 @@ NULL
/* PUBSUB NUMPAT */ /* PUBSUB NUMPAT */
addReplyLongLong(c,listLength(server.pubsub_patterns)); addReplyLongLong(c,listLength(server.pubsub_patterns));
} else { } else {
addReplyErrorFormat(c, "Unknown subcommand or wrong number of arguments for '%s'. Try PUBSUB HELP", addReplySubcommandSyntaxError(c);
(char*)c->argv[1]->ptr);
} }
} }
...@@ -1636,7 +1636,7 @@ int quicklistTest(int argc, char *argv[]) { ...@@ -1636,7 +1636,7 @@ int quicklistTest(int argc, char *argv[]) {
TEST("add to tail of empty list") { TEST("add to tail of empty list") {
quicklist *ql = quicklistNew(-2, options[_i]); quicklist *ql = quicklistNew(-2, options[_i]);
quicklistPushTail(ql, "hello", 6); quicklistPushTail(ql, "hello", 6);
/* 1 for head and 1 for tail beacuse 1 node = head = tail */ /* 1 for head and 1 for tail because 1 node = head = tail */
ql_verify(ql, 1, 1, 1, 1); ql_verify(ql, 1, 1, 1, 1);
quicklistRelease(ql); quicklistRelease(ql);
} }
...@@ -1644,7 +1644,7 @@ int quicklistTest(int argc, char *argv[]) { ...@@ -1644,7 +1644,7 @@ int quicklistTest(int argc, char *argv[]) {
TEST("add to head of empty list") { TEST("add to head of empty list") {
quicklist *ql = quicklistNew(-2, options[_i]); quicklist *ql = quicklistNew(-2, options[_i]);
quicklistPushHead(ql, "hello", 6); quicklistPushHead(ql, "hello", 6);
/* 1 for head and 1 for tail beacuse 1 node = head = tail */ /* 1 for head and 1 for tail because 1 node = head = tail */
ql_verify(ql, 1, 1, 1, 1); ql_verify(ql, 1, 1, 1, 1);
quicklistRelease(ql); quicklistRelease(ql);
} }
......
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