Unverified Commit 959d6035 authored by Oran Agra's avatar Oran Agra Committed by GitHub
Browse files

Merge 6.2.2 release

Release 6.2.2
parents 92bde124 aa730ef1
...@@ -199,7 +199,7 @@ quicklistNode *quicklistBookmarkFind(quicklist *ql, const char *name); ...@@ -199,7 +199,7 @@ quicklistNode *quicklistBookmarkFind(quicklist *ql, const char *name);
void quicklistBookmarksClear(quicklist *ql); void quicklistBookmarksClear(quicklist *ql);
#ifdef REDIS_TEST #ifdef REDIS_TEST
int quicklistTest(int argc, char *argv[]); int quicklistTest(int argc, char *argv[], int accurate);
#endif #endif
/* Directions for iterators */ /* Directions for iterators */
......
...@@ -1073,8 +1073,7 @@ size_t rdbSavedObjectLen(robj *o, robj *key) { ...@@ -1073,8 +1073,7 @@ size_t rdbSavedObjectLen(robj *o, robj *key) {
/* Save a key-value pair, with expire time, type, key, value. /* Save a key-value pair, with expire time, type, key, value.
* On error -1 is returned. * On error -1 is returned.
* On success if the key was actually saved 1 is returned, otherwise 0 * On success if the key was actually saved 1 is returned. */
* is returned (the key was already expired). */
int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime) { int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime) {
int savelru = server.maxmemory_policy & MAXMEMORY_FLAG_LRU; int savelru = server.maxmemory_policy & MAXMEMORY_FLAG_LRU;
int savelfu = server.maxmemory_policy & MAXMEMORY_FLAG_LFU; int savelfu = server.maxmemory_policy & MAXMEMORY_FLAG_LFU;
...@@ -2176,16 +2175,17 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key) { ...@@ -2176,16 +2175,17 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key) {
return NULL; return NULL;
} }
moduleType *mt = moduleTypeLookupModuleByID(moduleid); moduleType *mt = moduleTypeLookupModuleByID(moduleid);
char name[10];
if (rdbCheckMode && rdbtype == RDB_TYPE_MODULE_2) { if (rdbCheckMode && rdbtype == RDB_TYPE_MODULE_2) {
char name[10];
moduleTypeNameByID(name,moduleid); moduleTypeNameByID(name,moduleid);
return rdbLoadCheckModuleValue(rdb,name); return rdbLoadCheckModuleValue(rdb,name);
} }
if (mt == NULL) { if (mt == NULL) {
char name[10];
moduleTypeNameByID(name,moduleid); moduleTypeNameByID(name,moduleid);
rdbReportCorruptRDB("The RDB file contains module data I can't load: no matching module '%s'", name); rdbReportCorruptRDB("The RDB file contains module data I can't load: no matching module type '%s'", name);
return NULL; return NULL;
} }
RedisModuleIO io; RedisModuleIO io;
...@@ -2212,7 +2212,8 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key) { ...@@ -2212,7 +2212,8 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key) {
return NULL; return NULL;
} }
if (eof != RDB_MODULE_OPCODE_EOF) { if (eof != RDB_MODULE_OPCODE_EOF) {
rdbReportCorruptRDB("The RDB file contains module data for the module '%s' that is not terminated by the proper module value EOF marker", name); rdbReportCorruptRDB("The RDB file contains module data for the module '%s' that is not terminated by "
"the proper module value EOF marker", moduleTypeModuleName(mt));
if (ptr) { if (ptr) {
o = createModuleObject(mt,ptr); /* creating just in order to easily destroy */ o = createModuleObject(mt,ptr); /* creating just in order to easily destroy */
decrRefCount(o); decrRefCount(o);
...@@ -2222,8 +2223,9 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key) { ...@@ -2222,8 +2223,9 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key) {
} }
if (ptr == NULL) { if (ptr == NULL) {
moduleTypeNameByID(name,moduleid); rdbReportCorruptRDB("The RDB file contains module data for the module type '%s', that the responsible "
rdbReportCorruptRDB("The RDB file contains module data for the module type '%s', that the responsible module is not able to load. Check for modules log above for additional clues.", name); "module is not able to load. Check for modules log above for additional clues.",
moduleTypeModuleName(mt));
return NULL; return NULL;
} }
o = createModuleObject(mt,ptr); o = createModuleObject(mt,ptr);
...@@ -2729,7 +2731,7 @@ void backgroundSaveDoneHandler(int exitcode, int bysignal) { ...@@ -2729,7 +2731,7 @@ void backgroundSaveDoneHandler(int exitcode, int bysignal) {
* the cleanup needed. */ * the cleanup needed. */
void killRDBChild(void) { void killRDBChild(void) {
kill(server.child_pid, SIGUSR1); kill(server.child_pid, SIGUSR1);
/* Because we are not using here wait4 (like we have in killAppendOnlyChild /* Because we are not using here waitpid (like we have in killAppendOnlyChild
* and TerminateModuleForkChild), all the cleanup operations is done by * and TerminateModuleForkChild), all the cleanup operations is done by
* checkChildrenDone, that later will find that the process killed. * checkChildrenDone, that later will find that the process killed.
* This includes: * This includes:
......
...@@ -102,7 +102,6 @@ static struct config { ...@@ -102,7 +102,6 @@ static struct config {
int showerrors; int showerrors;
long long start; long long start;
long long totlatency; long long totlatency;
long long *latency;
const char *title; const char *title;
list *clients; list *clients;
int quiet; int quiet;
...@@ -1656,7 +1655,10 @@ int showThroughput(struct aeEventLoop *eventLoop, long long id, void *clientData ...@@ -1656,7 +1655,10 @@ int showThroughput(struct aeEventLoop *eventLoop, long long id, void *clientData
const float instantaneous_rps = (float)(requests_finished-previous_requests_finished)/instantaneous_dt; const float instantaneous_rps = (float)(requests_finished-previous_requests_finished)/instantaneous_dt;
config.previous_tick = current_tick; config.previous_tick = current_tick;
atomicSet(config.previous_requests_finished,requests_finished); atomicSet(config.previous_requests_finished,requests_finished);
config.last_printed_bytes = printf("%s: rps=%.1f (overall: %.1f) avg_msec=%.3f (overall: %.3f)\r", config.title, instantaneous_rps, rps, hdr_mean(config.current_sec_latency_histogram)/1000.0f, hdr_mean(config.latency_histogram)/1000.0f); int printed_bytes = printf("%s: rps=%.1f (overall: %.1f) avg_msec=%.3f (overall: %.3f)\r", config.title, instantaneous_rps, rps, hdr_mean(config.current_sec_latency_histogram)/1000.0f, hdr_mean(config.latency_histogram)/1000.0f);
if (printed_bytes > config.last_printed_bytes){
config.last_printed_bytes = printed_bytes;
}
hdr_reset(config.current_sec_latency_histogram); hdr_reset(config.current_sec_latency_histogram);
fflush(stdout); fflush(stdout);
return 250; /* every 250ms */ return 250; /* every 250ms */
......
...@@ -192,6 +192,7 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) { ...@@ -192,6 +192,7 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) {
int closefile = (fp == NULL); int closefile = (fp == NULL);
if (fp == NULL && (fp = fopen(rdbfilename,"r")) == NULL) return 1; if (fp == NULL && (fp = fopen(rdbfilename,"r")) == NULL) return 1;
startLoadingFile(fp, rdbfilename, RDBFLAGS_NONE);
rioInitWithFile(&rdb,fp); rioInitWithFile(&rdb,fp);
rdbstate.rio = &rdb; rdbstate.rio = &rdb;
rdb.update_cksum = rdbLoadProgressCallback; rdb.update_cksum = rdbLoadProgressCallback;
...@@ -208,7 +209,6 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) { ...@@ -208,7 +209,6 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) {
} }
expiretime = -1; expiretime = -1;
startLoadingFile(fp, rdbfilename, RDBFLAGS_NONE);
while(1) { while(1) {
robj *key, *val; robj *key, *val;
......
...@@ -227,7 +227,7 @@ static struct config { ...@@ -227,7 +227,7 @@ static struct config {
int scan_mode; int scan_mode;
int intrinsic_latency_mode; int intrinsic_latency_mode;
int intrinsic_latency_duration; int intrinsic_latency_duration;
char *pattern; sds pattern;
char *rdb_filename; char *rdb_filename;
int bigkeys; int bigkeys;
int memkeys; int memkeys;
...@@ -237,6 +237,7 @@ static struct config { ...@@ -237,6 +237,7 @@ static struct config {
char *auth; char *auth;
int askpass; int askpass;
char *user; char *user;
int quoted_input; /* Force input args to be treated as quoted strings */
int output; /* output mode, see OUTPUT_* defines */ int output; /* output mode, see OUTPUT_* defines */
int push_output; /* Should we display spontaneous PUSH replies */ int push_output; /* Should we display spontaneous PUSH replies */
sds mb_delim; sds mb_delim;
...@@ -405,15 +406,17 @@ static void parseRedisUri(const char *uri) { ...@@ -405,15 +406,17 @@ static void parseRedisUri(const char *uri) {
if (!strncasecmp(tlsscheme, curr, strlen(tlsscheme))) { if (!strncasecmp(tlsscheme, curr, strlen(tlsscheme))) {
#ifdef USE_OPENSSL #ifdef USE_OPENSSL
config.tls = 1; config.tls = 1;
curr += strlen(tlsscheme);
#else #else
fprintf(stderr,"rediss:// is only supported when redis-cli is compiled with OpenSSL\n"); fprintf(stderr,"rediss:// is only supported when redis-cli is compiled with OpenSSL\n");
exit(1); exit(1);
#endif #endif
} else if (strncasecmp(scheme, curr, strlen(scheme))) { } else if (!strncasecmp(scheme, curr, strlen(scheme))) {
curr += strlen(scheme);
} else {
fprintf(stderr,"Invalid URI scheme\n"); fprintf(stderr,"Invalid URI scheme\n");
exit(1); exit(1);
} }
curr += strlen(scheme);
if (curr == end) return; if (curr == end) return;
/* Extract user info. */ /* Extract user info. */
...@@ -763,6 +766,23 @@ static void freeHintsCallback(void *ptr) { ...@@ -763,6 +766,23 @@ static void freeHintsCallback(void *ptr) {
* Networking / parsing * Networking / parsing
*--------------------------------------------------------------------------- */ *--------------------------------------------------------------------------- */
/* Unquote a null-terminated string and return it as a binary-safe sds. */
static sds unquoteCString(char *str) {
int count;
sds *unquoted = sdssplitargs(str, &count);
sds res = NULL;
if (unquoted && count == 1) {
res = unquoted[0];
unquoted[0] = NULL;
}
if (unquoted)
sdsfreesplitres(unquoted, count);
return res;
}
/* Send AUTH command to the server */ /* Send AUTH command to the server */
static int cliAuth(redisContext *ctx, char *user, char *auth) { static int cliAuth(redisContext *ctx, char *user, char *auth) {
redisReply *reply; redisReply *reply;
...@@ -1533,6 +1553,8 @@ static int parseOptions(int argc, char **argv) { ...@@ -1533,6 +1553,8 @@ static int parseOptions(int argc, char **argv) {
config.output = OUTPUT_RAW; config.output = OUTPUT_RAW;
} else if (!strcmp(argv[i],"--no-raw")) { } else if (!strcmp(argv[i],"--no-raw")) {
config.output = OUTPUT_STANDARD; config.output = OUTPUT_STANDARD;
} else if (!strcmp(argv[i],"--quoted-input")) {
config.quoted_input = 1;
} else if (!strcmp(argv[i],"--csv")) { } else if (!strcmp(argv[i],"--csv")) {
config.output = OUTPUT_CSV; config.output = OUTPUT_CSV;
} else if (!strcmp(argv[i],"--latency")) { } else if (!strcmp(argv[i],"--latency")) {
...@@ -1557,7 +1579,15 @@ static int parseOptions(int argc, char **argv) { ...@@ -1557,7 +1579,15 @@ static int parseOptions(int argc, char **argv) {
} else if (!strcmp(argv[i],"--scan")) { } else if (!strcmp(argv[i],"--scan")) {
config.scan_mode = 1; config.scan_mode = 1;
} else if (!strcmp(argv[i],"--pattern") && !lastarg) { } else if (!strcmp(argv[i],"--pattern") && !lastarg) {
config.pattern = argv[++i]; sdsfree(config.pattern);
config.pattern = sdsnew(argv[++i]);
} else if (!strcmp(argv[i],"--quoted-pattern") && !lastarg) {
sdsfree(config.pattern);
config.pattern = unquoteCString(argv[++i]);
if (!config.pattern) {
fprintf(stderr,"Invalid quoted string specified for --quoted-pattern.\n");
exit(1);
}
} else if (!strcmp(argv[i],"--intrinsic-latency") && !lastarg) { } else if (!strcmp(argv[i],"--intrinsic-latency") && !lastarg) {
config.intrinsic_latency_mode = 1; config.intrinsic_latency_mode = 1;
config.intrinsic_latency_duration = atoi(argv[++i]); config.intrinsic_latency_duration = atoi(argv[++i]);
...@@ -1841,6 +1871,7 @@ static void usage(void) { ...@@ -1841,6 +1871,7 @@ static void usage(void) {
" --raw Use raw formatting for replies (default when STDOUT is\n" " --raw Use raw formatting for replies (default when STDOUT is\n"
" not a tty).\n" " not a tty).\n"
" --no-raw Force formatted output even when STDOUT is not a tty.\n" " --no-raw Force formatted output even when STDOUT is not a tty.\n"
" --quoted-input Force input to be handled as quoted strings.\n"
" --csv Output in CSV format.\n" " --csv Output in CSV format.\n"
" --show-pushes <yn> Whether to print RESP3 PUSH messages. Enabled by default when\n" " --show-pushes <yn> Whether to print RESP3 PUSH messages. Enabled by default when\n"
" STDOUT is a tty but can be overriden with --show-pushes no.\n" " STDOUT is a tty but can be overriden with --show-pushes no.\n"
...@@ -1876,6 +1907,8 @@ static void usage(void) { ...@@ -1876,6 +1907,8 @@ static void usage(void) {
" --scan List all keys using the SCAN command.\n" " --scan List all keys using the SCAN command.\n"
" --pattern <pat> Keys pattern when using the --scan, --bigkeys or --hotkeys\n" " --pattern <pat> Keys pattern when using the --scan, --bigkeys or --hotkeys\n"
" options (default: *).\n" " options (default: *).\n"
" --quoted-pattern <pat> Same as --pattern, but the specified string can be\n"
" quoted, in order to pass an otherwise non binary-safe string.\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"
" The test will run for the specified amount of seconds.\n" " The test will run for the specified amount of seconds.\n"
" --eval <file> Send an EVAL command using the Lua script at <file>.\n" " --eval <file> Send an EVAL command using the Lua script at <file>.\n"
...@@ -1901,6 +1934,7 @@ static void usage(void) { ...@@ -1901,6 +1934,7 @@ static void usage(void) {
" redis-cli get mypasswd\n" " redis-cli get mypasswd\n"
" redis-cli -r 100 lpush mylist x\n" " redis-cli -r 100 lpush mylist x\n"
" redis-cli -r 100 -i 1 info | grep used_memory_human:\n" " redis-cli -r 100 -i 1 info | grep used_memory_human:\n"
" redis-cli --quoted-input set '\"null-\\x00-separated\"' value\n"
" redis-cli --eval myscript.lua key1 key2 , arg1 arg2 arg3\n" " redis-cli --eval myscript.lua key1 key2 , arg1 arg2 arg3\n"
" redis-cli --scan --pattern '*:12345*'\n" " redis-cli --scan --pattern '*:12345*'\n"
"\n" "\n"
...@@ -1930,22 +1964,28 @@ static int confirmWithYes(char *msg, int ignore_force) { ...@@ -1930,22 +1964,28 @@ static int confirmWithYes(char *msg, int ignore_force) {
return (nread != 0 && !strcmp("yes", buf)); return (nread != 0 && !strcmp("yes", buf));
} }
/* Turn the plain C strings into Sds strings */ /* Create an sds array from argv, either as-is or by dequoting every
static char **convertToSds(int count, char** args) { * element. When quoted is non-zero, may return a NULL to indicate an
int j; * invalid quoted string.
char **sds = zmalloc(sizeof(char*)*count); */
static sds *getSdsArrayFromArgv(int argc, char **argv, int quoted) {
for(j = 0; j < count; j++) sds *res = sds_malloc(sizeof(sds) * argc);
sds[j] = sdsnew(args[j]);
for (int j = 0; j < argc; j++) {
return sds; if (quoted) {
} sds unquoted = unquoteCString(argv[j]);
if (!unquoted) {
while (--j >= 0) sdsfree(res[j]);
sds_free(res);
return NULL;
}
res[j] = unquoted;
} else {
res[j] = sdsnew(argv[j]);
}
}
static void freeConvertedSds(int count, char **sds) { return res;
int j;
for (j = 0; j < count; j++)
sdsfree(sds[j]);
zfree(sds);
} }
static int issueCommandRepeat(int argc, char **argv, long repeat) { static int issueCommandRepeat(int argc, char **argv, long repeat) {
...@@ -2178,17 +2218,19 @@ static void repl(void) { ...@@ -2178,17 +2218,19 @@ static void repl(void) {
static int noninteractive(int argc, char **argv) { static int noninteractive(int argc, char **argv) {
int retval = 0; int retval = 0;
sds *sds_args = getSdsArrayFromArgv(argc, argv, config.quoted_input);
argv = convertToSds(argc, argv); if (!sds_args) {
printf("Invalid quoted string\n");
return 1;
}
if (config.stdinarg) { if (config.stdinarg) {
argv = zrealloc(argv, (argc+1)*sizeof(char*)); sds_args = sds_realloc(sds_args, (argc + 1) * sizeof(sds));
argv[argc] = readArgFromStdin(); sds_args[argc] = readArgFromStdin();
retval = issueCommand(argc+1, argv); argc++;
sdsfree(argv[argc]);
} else {
retval = issueCommand(argc, argv);
} }
freeConvertedSds(argc, argv);
retval = issueCommand(argc, sds_args);
sdsfreesplitres(sds_args, argc);
return retval; return retval;
} }
...@@ -2913,8 +2955,12 @@ static int clusterManagerGetAntiAffinityScore(clusterManagerNodeArray *ipnodes, ...@@ -2913,8 +2955,12 @@ static int clusterManagerGetAntiAffinityScore(clusterManagerNodeArray *ipnodes,
else types = sdsempty(); else types = sdsempty();
/* Master type 'm' is always set as the first character of the /* Master type 'm' is always set as the first character of the
* types string. */ * types string. */
if (!node->replicate) types = sdscatprintf(types, "m%s", types); if (node->replicate) types = sdscat(types, "s");
else types = sdscat(types, "s"); else {
sds s = sdscatsds(sdsnew("m"), types);
sdsfree(types);
types = s;
}
dictReplace(related, key, types); dictReplace(related, key, types);
} }
/* Now it's trivial to check, for each related group having the /* Now it's trivial to check, for each related group having the
...@@ -7156,7 +7202,10 @@ static void getRDB(clusterManagerNode *node) { ...@@ -7156,7 +7202,10 @@ static void getRDB(clusterManagerNode *node) {
redisFree(s); /* Close the connection ASAP as fsync() may take time. */ redisFree(s); /* Close the connection ASAP as fsync() may take time. */
if (node) if (node)
node->context = NULL; node->context = NULL;
fsync(fd); if (fsync(fd) == -1) {
fprintf(stderr,"Fail to fsync '%s': %s\n", filename, strerror(errno));
exit(1);
}
close(fd); close(fd);
if (node) { if (node) {
sdsfree(filename); sdsfree(filename);
...@@ -7332,8 +7381,8 @@ static redisReply *sendScan(unsigned long long *it) { ...@@ -7332,8 +7381,8 @@ static redisReply *sendScan(unsigned long long *it) {
redisReply *reply; redisReply *reply;
if (config.pattern) if (config.pattern)
reply = redisCommand(context,"SCAN %llu MATCH %s", reply = redisCommand(context, "SCAN %llu MATCH %b",
*it,config.pattern); *it, config.pattern, sdslen(config.pattern));
else else
reply = redisCommand(context,"SCAN %llu",*it); reply = redisCommand(context,"SCAN %llu",*it);
...@@ -7368,8 +7417,14 @@ static int getDbSize(void) { ...@@ -7368,8 +7417,14 @@ static int getDbSize(void) {
reply = redisCommand(context, "DBSIZE"); reply = redisCommand(context, "DBSIZE");
if(reply == NULL || reply->type != REDIS_REPLY_INTEGER) { if (reply == NULL) {
fprintf(stderr, "Couldn't determine DBSIZE!\n"); fprintf(stderr, "\nI/O error\n");
exit(1);
} else if (reply->type == REDIS_REPLY_ERROR) {
fprintf(stderr, "Couldn't determine DBSIZE: %s\n", reply->str);
exit(1);
} else if (reply->type != REDIS_REPLY_INTEGER) {
fprintf(stderr, "Non INTEGER response from DBSIZE!\n");
exit(1); exit(1);
} }
...@@ -7945,24 +8000,17 @@ static void scanMode(void) { ...@@ -7945,24 +8000,17 @@ static void scanMode(void) {
unsigned long long cur = 0; unsigned long long cur = 0;
do { do {
if (config.pattern) reply = sendScan(&cur);
reply = redisCommand(context,"SCAN %llu MATCH %s", for (unsigned int j = 0; j < reply->element[1]->elements; j++) {
cur,config.pattern); if (config.output == OUTPUT_STANDARD) {
else sds out = sdscatrepr(sdsempty(), reply->element[1]->element[j]->str,
reply = redisCommand(context,"SCAN %llu",cur); reply->element[1]->element[j]->len);
if (reply == NULL) { printf("%s\n", out);
printf("I/O error\n"); sdsfree(out);
exit(1);
} else if (reply->type == REDIS_REPLY_ERROR) {
printf("ERROR: %s\n", reply->str);
exit(1);
} else { } else {
unsigned int j;
cur = strtoull(reply->element[0]->str,NULL,10);
for (j = 0; j < reply->element[1]->elements; j++)
printf("%s\n", reply->element[1]->element[j]->str); printf("%s\n", reply->element[1]->element[j]->str);
} }
}
freeReplyObject(reply); freeReplyObject(reply);
} while(cur != 0); } while(cur != 0);
......
...@@ -160,13 +160,14 @@ This flag should not be used directly by the module. ...@@ -160,13 +160,14 @@ This flag should not be used directly by the module.
#define REDISMODULE_NOTIFY_STREAM (1<<10) /* t */ #define REDISMODULE_NOTIFY_STREAM (1<<10) /* t */
#define REDISMODULE_NOTIFY_KEY_MISS (1<<11) /* m (Note: This one is excluded from REDISMODULE_NOTIFY_ALL on purpose) */ #define REDISMODULE_NOTIFY_KEY_MISS (1<<11) /* m (Note: This one is excluded from REDISMODULE_NOTIFY_ALL on purpose) */
#define REDISMODULE_NOTIFY_LOADED (1<<12) /* module only key space notification, indicate a key loaded from rdb */ #define REDISMODULE_NOTIFY_LOADED (1<<12) /* module only key space notification, indicate a key loaded from rdb */
#define REDISMODULE_NOTIFY_MODULE (1<<13) /* d, module key space notification */
/* Next notification flag, must be updated when adding new flags above! /* Next notification flag, must be updated when adding new flags above!
This flag should not be used directly by the module. This flag should not be used directly by the module.
* Use RedisModule_GetKeyspaceNotificationFlagsAll instead. */ * Use RedisModule_GetKeyspaceNotificationFlagsAll instead. */
#define _REDISMODULE_NOTIFY_NEXT (1<<13) #define _REDISMODULE_NOTIFY_NEXT (1<<14)
#define REDISMODULE_NOTIFY_ALL (REDISMODULE_NOTIFY_GENERIC | REDISMODULE_NOTIFY_STRING | REDISMODULE_NOTIFY_LIST | REDISMODULE_NOTIFY_SET | REDISMODULE_NOTIFY_HASH | REDISMODULE_NOTIFY_ZSET | REDISMODULE_NOTIFY_EXPIRED | REDISMODULE_NOTIFY_EVICTED | REDISMODULE_NOTIFY_STREAM) /* A */ #define REDISMODULE_NOTIFY_ALL (REDISMODULE_NOTIFY_GENERIC | REDISMODULE_NOTIFY_STRING | REDISMODULE_NOTIFY_LIST | REDISMODULE_NOTIFY_SET | REDISMODULE_NOTIFY_HASH | REDISMODULE_NOTIFY_ZSET | REDISMODULE_NOTIFY_EXPIRED | REDISMODULE_NOTIFY_EVICTED | REDISMODULE_NOTIFY_STREAM | REDISMODULE_NOTIFY_MODULE) /* A */
/* A special pointer that we can use between the core and the module to signal /* A special pointer that we can use between the core and the module to signal
* field deletion, and that is impossible to be a valid pointer. */ * field deletion, and that is impossible to be a valid pointer. */
...@@ -193,6 +194,12 @@ This flag should not be used directly by the module. ...@@ -193,6 +194,12 @@ This flag should not be used directly by the module.
#define REDISMODULE_NOT_USED(V) ((void) V) #define REDISMODULE_NOT_USED(V) ((void) V)
/* Logging level strings */
#define REDISMODULE_LOGLEVEL_DEBUG "debug"
#define REDISMODULE_LOGLEVEL_VERBOSE "verbose"
#define REDISMODULE_LOGLEVEL_NOTICE "notice"
#define REDISMODULE_LOGLEVEL_WARNING "warning"
/* Bit flags for aux_save_triggers and the aux_load and aux_save callbacks */ /* Bit flags for aux_save_triggers and the aux_load and aux_save callbacks */
#define REDISMODULE_AUX_BEFORE_RDB (1<<0) #define REDISMODULE_AUX_BEFORE_RDB (1<<0)
#define REDISMODULE_AUX_AFTER_RDB (1<<1) #define REDISMODULE_AUX_AFTER_RDB (1<<1)
...@@ -635,6 +642,8 @@ REDISMODULE_API char * (*RedisModule_StringDMA)(RedisModuleKey *key, size_t *len ...@@ -635,6 +642,8 @@ REDISMODULE_API char * (*RedisModule_StringDMA)(RedisModuleKey *key, size_t *len
REDISMODULE_API int (*RedisModule_StringTruncate)(RedisModuleKey *key, size_t newlen) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_StringTruncate)(RedisModuleKey *key, size_t newlen) REDISMODULE_ATTR;
REDISMODULE_API mstime_t (*RedisModule_GetExpire)(RedisModuleKey *key) REDISMODULE_ATTR; REDISMODULE_API mstime_t (*RedisModule_GetExpire)(RedisModuleKey *key) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_SetExpire)(RedisModuleKey *key, mstime_t expire) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_SetExpire)(RedisModuleKey *key, mstime_t expire) REDISMODULE_ATTR;
REDISMODULE_API mstime_t (*RedisModule_GetAbsExpire)(RedisModuleKey *key) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_SetAbsExpire)(RedisModuleKey *key, mstime_t expire) REDISMODULE_ATTR;
REDISMODULE_API void (*RedisModule_ResetDataset)(int restart_aof, int async) REDISMODULE_ATTR; REDISMODULE_API void (*RedisModule_ResetDataset)(int restart_aof, int async) REDISMODULE_ATTR;
REDISMODULE_API unsigned long long (*RedisModule_DbSize)(RedisModuleCtx *ctx) REDISMODULE_ATTR; REDISMODULE_API unsigned long long (*RedisModule_DbSize)(RedisModuleCtx *ctx) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleString * (*RedisModule_RandomKey)(RedisModuleCtx *ctx) REDISMODULE_ATTR; REDISMODULE_API RedisModuleString * (*RedisModule_RandomKey)(RedisModuleCtx *ctx) REDISMODULE_ATTR;
...@@ -835,7 +844,7 @@ REDISMODULE_API int (*RedisModule_DefragCursorSet)(RedisModuleDefragCtx *ctx, un ...@@ -835,7 +844,7 @@ REDISMODULE_API int (*RedisModule_DefragCursorSet)(RedisModuleDefragCtx *ctx, un
REDISMODULE_API int (*RedisModule_DefragCursorGet)(RedisModuleDefragCtx *ctx, unsigned long *cursor) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_DefragCursorGet)(RedisModuleDefragCtx *ctx, unsigned long *cursor) REDISMODULE_ATTR;
#endif #endif
#define RedisModule_IsAOFClient(id) ((id) == CLIENT_ID_AOF) #define RedisModule_IsAOFClient(id) ((id) == UINT64_MAX)
/* This is included inline inside each Redis module. */ /* This is included inline inside each Redis module. */
static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int apiver) REDISMODULE_ATTR_UNUSED; static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int apiver) REDISMODULE_ATTR_UNUSED;
...@@ -907,6 +916,8 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -907,6 +916,8 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(StringTruncate); REDISMODULE_GET_API(StringTruncate);
REDISMODULE_GET_API(GetExpire); REDISMODULE_GET_API(GetExpire);
REDISMODULE_GET_API(SetExpire); REDISMODULE_GET_API(SetExpire);
REDISMODULE_GET_API(GetAbsExpire);
REDISMODULE_GET_API(SetAbsExpire);
REDISMODULE_GET_API(ResetDataset); REDISMODULE_GET_API(ResetDataset);
REDISMODULE_GET_API(DbSize); REDISMODULE_GET_API(DbSize);
REDISMODULE_GET_API(RandomKey); REDISMODULE_GET_API(RandomKey);
......
...@@ -892,17 +892,34 @@ void syncCommand(client *c) { ...@@ -892,17 +892,34 @@ void syncCommand(client *c) {
} }
/* REPLCONF <option> <value> <option> <value> ... /* REPLCONF <option> <value> <option> <value> ...
* This command is used by a slave in order to configure the replication * This command is used by a replica in order to configure the replication
* process before starting it with the SYNC command. * process before starting it with the SYNC command.
* This command is also used by a master in order to get the replication
* offset from a replica.
* *
* Currently the only use of this command is to communicate to the master * Currently we support these options:
* what is the listening port of the Slave redis instance, so that the
* master can accurately list slaves and their listening ports in
* the INFO output.
* *
* In the future the same command can be used in order to configure * - listening-port <port>
* the replication to initiate an incremental replication instead of a * - ip-address <ip>
* full resync. */ * What is the listening ip and port of the Replica redis instance, so that
* the master can accurately lists replicas and their listening ports in the
* INFO output.
*
* - capa <eof|psync2>
* What is the capabilities of this instance.
* eof: supports EOF-style RDB transfer for diskless replication.
* psync2: supports PSYNC v2, so understands +CONTINUE <new repl ID>.
*
* - ack <offset>
* Replica informs the master the amount of replication stream that it
* processed so far.
*
* - getack
* Unlike other subcommands, this is used by master to get the replication
* offset from a replica.
*
* - rdb-only
* Only wants RDB snapshot without replication buffer. */
void replconfCommand(client *c) { void replconfCommand(client *c) {
int j; int j;
...@@ -1136,6 +1153,8 @@ void rdbPipeWriteHandlerConnRemoved(struct connection *conn) { ...@@ -1136,6 +1153,8 @@ void rdbPipeWriteHandlerConnRemoved(struct connection *conn) {
if (!connHasWriteHandler(conn)) if (!connHasWriteHandler(conn))
return; return;
connSetWriteHandler(conn, NULL); connSetWriteHandler(conn, NULL);
client *slave = connGetPrivateData(conn);
slave->repl_last_partial_write = 0;
server.rdb_pipe_numconns_writing--; server.rdb_pipe_numconns_writing--;
/* if there are no more writes for now for this conn, or write error: */ /* if there are no more writes for now for this conn, or write error: */
if (server.rdb_pipe_numconns_writing == 0) { if (server.rdb_pipe_numconns_writing == 0) {
...@@ -1163,9 +1182,11 @@ void rdbPipeWriteHandler(struct connection *conn) { ...@@ -1163,9 +1182,11 @@ void rdbPipeWriteHandler(struct connection *conn) {
} else { } else {
slave->repldboff += nwritten; slave->repldboff += nwritten;
atomicIncr(server.stat_net_output_bytes, nwritten); atomicIncr(server.stat_net_output_bytes, nwritten);
if (slave->repldboff < server.rdb_pipe_bufflen) if (slave->repldboff < server.rdb_pipe_bufflen) {
slave->repl_last_partial_write = server.unixtime;
return; /* more data to write.. */ return; /* more data to write.. */
} }
}
rdbPipeWriteHandlerConnRemoved(conn); rdbPipeWriteHandlerConnRemoved(conn);
} }
...@@ -1245,6 +1266,7 @@ void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData, ...@@ -1245,6 +1266,7 @@ void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData,
/* If we were unable to write all the data to one of the replicas, /* If we were unable to write all the data to one of the replicas,
* setup write handler (and disable pipe read handler, below) */ * setup write handler (and disable pipe read handler, below) */
if (nwritten != server.rdb_pipe_bufflen) { if (nwritten != server.rdb_pipe_bufflen) {
slave->repl_last_partial_write = server.unixtime;
server.rdb_pipe_numconns_writing++; server.rdb_pipe_numconns_writing++;
connSetWriteHandler(conn, rdbPipeWriteHandler); connSetWriteHandler(conn, rdbPipeWriteHandler);
} }
...@@ -1873,8 +1895,7 @@ void readSyncBulkPayload(connection *conn) { ...@@ -1873,8 +1895,7 @@ void readSyncBulkPayload(connection *conn) {
serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Finished with success"); serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Finished with success");
if (server.supervised_mode == SUPERVISED_SYSTEMD) { if (server.supervised_mode == SUPERVISED_SYSTEMD) {
redisCommunicateSystemd("STATUS=MASTER <-> REPLICA sync: Finished with success. Ready to accept connections.\n"); redisCommunicateSystemd("STATUS=MASTER <-> REPLICA sync: Finished with success. Ready to accept connections in read-write mode.\n");
redisCommunicateSystemd("READY=1\n");
} }
/* Send the initial ACK immediately to put this replica in online state. */ /* Send the initial ACK immediately to put this replica in online state. */
...@@ -2434,8 +2455,7 @@ void syncWithMaster(connection *conn) { ...@@ -2434,8 +2455,7 @@ void syncWithMaster(connection *conn) {
if (psync_result == PSYNC_CONTINUE) { if (psync_result == PSYNC_CONTINUE) {
serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Master accepted a Partial Resynchronization."); serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Master accepted a Partial Resynchronization.");
if (server.supervised_mode == SUPERVISED_SYSTEMD) { if (server.supervised_mode == SUPERVISED_SYSTEMD) {
redisCommunicateSystemd("STATUS=MASTER <-> REPLICA sync: Partial Resynchronization accepted. Ready to accept connections.\n"); redisCommunicateSystemd("STATUS=MASTER <-> REPLICA sync: Partial Resynchronization accepted. Ready to accept connections in read-write mode.\n");
redisCommunicateSystemd("READY=1\n");
} }
return; return;
} }
...@@ -2684,6 +2704,9 @@ void replicationUnsetMaster(void) { ...@@ -2684,6 +2704,9 @@ void replicationUnsetMaster(void) {
* failover if slaves do not connect immediately. */ * failover if slaves do not connect immediately. */
server.repl_no_slaves_since = server.unixtime; server.repl_no_slaves_since = server.unixtime;
/* Reset down time so it'll be ready for when we turn into replica again. */
server.repl_down_since = 0;
/* Fire the role change modules event. */ /* Fire the role change modules event. */
moduleFireServerEvent(REDISMODULE_EVENT_REPLICATION_ROLE_CHANGED, moduleFireServerEvent(REDISMODULE_EVENT_REPLICATION_ROLE_CHANGED,
REDISMODULE_EVENT_REPLROLECHANGED_NOW_MASTER, REDISMODULE_EVENT_REPLROLECHANGED_NOW_MASTER,
...@@ -2758,7 +2781,7 @@ void replicaofCommand(client *c) { ...@@ -2758,7 +2781,7 @@ void replicaofCommand(client *c) {
if ((getLongFromObjectOrReply(c, c->argv[2], &port, NULL) != C_OK)) if ((getLongFromObjectOrReply(c, c->argv[2], &port, NULL) != C_OK))
return; return;
/* Check if we are already attached to the specified slave */ /* Check if we are already attached to the specified master */
if (server.masterhost && !strcasecmp(server.masterhost,c->argv[1]->ptr) if (server.masterhost && !strcasecmp(server.masterhost,c->argv[1]->ptr)
&& server.masterport == port) { && server.masterport == port) {
serverLog(LL_NOTICE,"REPLICAOF would result into synchronization " serverLog(LL_NOTICE,"REPLICAOF would result into synchronization "
...@@ -3375,13 +3398,28 @@ void replicationCron(void) { ...@@ -3375,13 +3398,28 @@ void replicationCron(void) {
while((ln = listNext(&li))) { while((ln = listNext(&li))) {
client *slave = ln->value; client *slave = ln->value;
if (slave->replstate != SLAVE_STATE_ONLINE) continue; if (slave->replstate == SLAVE_STATE_ONLINE) {
if (slave->flags & CLIENT_PRE_PSYNC) continue; if (slave->flags & CLIENT_PRE_PSYNC)
if ((server.unixtime - slave->repl_ack_time) > server.repl_timeout) continue;
if ((server.unixtime - slave->repl_ack_time) > server.repl_timeout) {
serverLog(LL_WARNING, "Disconnecting timedout replica (streaming sync): %s",
replicationGetSlaveName(slave));
freeClient(slave);
continue;
}
}
/* We consider disconnecting only diskless replicas because disk-based replicas aren't fed
* by the fork child so if a disk-based replica is stuck it doesn't prevent the fork child
* from terminating. */
if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_END && server.rdb_child_type == RDB_CHILD_TYPE_SOCKET) {
if (slave->repl_last_partial_write != 0 &&
(server.unixtime - slave->repl_last_partial_write) > server.repl_timeout)
{ {
serverLog(LL_WARNING, "Disconnecting timedout replica: %s", serverLog(LL_WARNING, "Disconnecting timedout replica (full sync): %s",
replicationGetSlaveName(slave)); replicationGetSlaveName(slave));
freeClient(slave); freeClient(slave);
continue;
}
} }
} }
} }
...@@ -3546,7 +3584,7 @@ void abortFailover(const char *err) { ...@@ -3546,7 +3584,7 @@ void abortFailover(const char *err) {
} }
/* /*
* FAILOVER [TO <HOST> <IP> [FORCE]] [ABORT] [TIMEOUT <timeout>] * FAILOVER [TO <HOST> <PORT> [FORCE]] [ABORT] [TIMEOUT <timeout>]
* *
* This command will coordinate a failover between the master and one * This command will coordinate a failover between the master and one
* of its replicas. The happy path contains the following steps: * of its replicas. The happy path contains the following steps:
...@@ -3649,7 +3687,7 @@ void failoverCommand(client *c) { ...@@ -3649,7 +3687,7 @@ void failoverCommand(client *c) {
client *replica = findReplica(host, port); client *replica = findReplica(host, port);
if (replica == NULL) { if (replica == NULL) {
addReplyError(c,"FAILOVER target HOST and IP is not " addReplyError(c,"FAILOVER target HOST and PORT is not "
"a replica."); "a replica.");
return; return;
} }
......
...@@ -117,7 +117,7 @@ static size_t rioFileWrite(rio *r, const void *buf, size_t len) { ...@@ -117,7 +117,7 @@ static size_t rioFileWrite(rio *r, const void *buf, size_t len) {
r->io.file.buffered >= r->io.file.autosync) r->io.file.buffered >= r->io.file.autosync)
{ {
fflush(r->io.file.fp); fflush(r->io.file.fp);
redis_fsync(fileno(r->io.file.fp)); if (redis_fsync(fileno(r->io.file.fp)) == -1) return 0;
r->io.file.buffered = 0; r->io.file.buffered = 0;
} }
return retval; return retval;
...@@ -160,7 +160,7 @@ void rioInitWithFile(rio *r, FILE *fp) { ...@@ -160,7 +160,7 @@ void rioInitWithFile(rio *r, FILE *fp) {
} }
/* ------------------- Connection implementation ------------------- /* ------------------- Connection implementation -------------------
* We use this RIO implemetnation when reading an RDB file directly from * We use this RIO implementation when reading an RDB file directly from
* the connection to the memory via rdbLoadRio(), thus this implementation * the connection to the memory via rdbLoadRio(), thus this implementation
* only implements reading from a connection that is, normally, * only implements reading from a connection that is, normally,
* just a socket. */ * just a socket. */
...@@ -262,7 +262,7 @@ void rioInitWithConn(rio *r, connection *conn, size_t read_limit) { ...@@ -262,7 +262,7 @@ void rioInitWithConn(rio *r, connection *conn, size_t read_limit) {
sdsclear(r->io.conn.buf); sdsclear(r->io.conn.buf);
} }
/* Release the RIO tream. Optionally returns the unread buffered data /* Release the RIO stream. Optionally returns the unread buffered data
* when the SDS pointer 'remaining' is passed. */ * when the SDS pointer 'remaining' is passed. */
void rioFreeConn(rio *r, sds *remaining) { void rioFreeConn(rio *r, sds *remaining) {
if (remaining && (size_t)r->io.conn.pos < sdslen(r->io.conn.buf)) { if (remaining && (size_t)r->io.conn.pos < sdslen(r->io.conn.buf)) {
......
...@@ -604,9 +604,7 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) { ...@@ -604,9 +604,7 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) {
/* Check the ACLs. */ /* Check the ACLs. */
int acl_errpos; int acl_errpos;
int acl_retval = ACLCheckCommandPerm(c,&acl_errpos); int acl_retval = ACLCheckAllPerm(c,&acl_errpos);
if (acl_retval == ACL_OK && c->cmd->proc == publishCommand)
acl_retval = ACLCheckPubsubPerm(c,1,1,0,&acl_errpos);
if (acl_retval != ACL_OK) { if (acl_retval != ACL_OK) {
addACLLogEntry(c,acl_retval,acl_errpos,NULL); addACLLogEntry(c,acl_retval,acl_errpos,NULL);
switch (acl_retval) { switch (acl_retval) {
...@@ -1453,6 +1451,14 @@ void luaMaskCountHook(lua_State *lua, lua_Debug *ar) { ...@@ -1453,6 +1451,14 @@ void luaMaskCountHook(lua_State *lua, lua_Debug *ar) {
if (server.lua_timedout) processEventsWhileBlocked(); if (server.lua_timedout) processEventsWhileBlocked();
if (server.lua_kill) { if (server.lua_kill) {
serverLog(LL_WARNING,"Lua script killed by user with SCRIPT KILL."); serverLog(LL_WARNING,"Lua script killed by user with SCRIPT KILL.");
/*
* Set the hook to invoke all the time so the user
         * will not be able to catch the error with pcall and invoke
         * pcall again which will prevent the script from ever been killed
*/
lua_sethook(lua, luaMaskCountHook, LUA_MASKLINE, 0);
lua_pushstring(lua,"Script killed by user with SCRIPT KILL..."); lua_pushstring(lua,"Script killed by user with SCRIPT KILL...");
lua_error(lua); lua_error(lua);
} }
...@@ -1498,7 +1504,6 @@ void evalGenericCommand(client *c, int evalsha) { ...@@ -1498,7 +1504,6 @@ void evalGenericCommand(client *c, int evalsha) {
server.lua_replicate_commands = server.lua_always_replicate_commands; server.lua_replicate_commands = server.lua_always_replicate_commands;
server.lua_multi_emitted = 0; server.lua_multi_emitted = 0;
server.lua_repl = PROPAGATE_AOF|PROPAGATE_REPL; server.lua_repl = PROPAGATE_AOF|PROPAGATE_REPL;
server.in_eval = 1;
/* Get the number of arguments that are keys */ /* Get the number of arguments that are keys */
if (getLongLongFromObjectOrReply(c,c->argv[2],&numkeys,NULL) != C_OK) if (getLongLongFromObjectOrReply(c,c->argv[2],&numkeys,NULL) != C_OK)
...@@ -1570,6 +1575,7 @@ void evalGenericCommand(client *c, int evalsha) { ...@@ -1570,6 +1575,7 @@ void evalGenericCommand(client *c, int evalsha) {
* *
* If we are debugging, we set instead a "line" hook so that the * If we are debugging, we set instead a "line" hook so that the
* debugger is call-back at every line executed by the script. */ * debugger is call-back at every line executed by the script. */
server.in_eval = 1;
server.lua_caller = c; server.lua_caller = c;
server.lua_cur_script = funcname + 2; server.lua_cur_script = funcname + 2;
server.lua_time_start = mstime(); server.lua_time_start = mstime();
...@@ -1602,6 +1608,7 @@ void evalGenericCommand(client *c, int evalsha) { ...@@ -1602,6 +1608,7 @@ void evalGenericCommand(client *c, int evalsha) {
if (server.masterhost && server.master) if (server.masterhost && server.master)
queueClientForReprocessing(server.master); queueClientForReprocessing(server.master);
} }
server.in_eval = 0;
server.lua_caller = NULL; server.lua_caller = NULL;
server.lua_cur_script = NULL; server.lua_cur_script = NULL;
...@@ -1678,8 +1685,6 @@ void evalGenericCommand(client *c, int evalsha) { ...@@ -1678,8 +1685,6 @@ void evalGenericCommand(client *c, int evalsha) {
forceCommandPropagation(c,PROPAGATE_REPL|PROPAGATE_AOF); forceCommandPropagation(c,PROPAGATE_REPL|PROPAGATE_AOF);
} }
} }
server.in_eval = 0;
} }
void evalCommand(client *c) { void evalCommand(client *c) {
......
...@@ -1234,9 +1234,10 @@ static sds sdsTestTemplateCallback(sds varname, void *arg) { ...@@ -1234,9 +1234,10 @@ static sds sdsTestTemplateCallback(sds varname, void *arg) {
else return NULL; else return NULL;
} }
int sdsTest(int argc, char **argv) { int sdsTest(int argc, char **argv, int accurate) {
UNUSED(argc); UNUSED(argc);
UNUSED(argv); UNUSED(argv);
UNUSED(accurate);
{ {
sds x = sdsnew("foo"), y; sds x = sdsnew("foo"), y;
......
...@@ -277,7 +277,7 @@ void *sds_realloc(void *ptr, size_t size); ...@@ -277,7 +277,7 @@ void *sds_realloc(void *ptr, size_t size);
void sds_free(void *ptr); void sds_free(void *ptr);
#ifdef REDIS_TEST #ifdef REDIS_TEST
int sdsTest(int argc, char *argv[]); int sdsTest(int argc, char *argv[], int accurate);
#endif #endif
#endif #endif
...@@ -215,6 +215,7 @@ typedef struct sentinelRedisInstance { ...@@ -215,6 +215,7 @@ typedef struct sentinelRedisInstance {
/* Slave specific. */ /* Slave specific. */
mstime_t master_link_down_time; /* Slave replication link down time. */ mstime_t master_link_down_time; /* Slave replication link down time. */
int slave_priority; /* Slave priority according to its INFO output. */ int slave_priority; /* Slave priority according to its INFO output. */
int replica_announced; /* Replica announcing according to its INFO output. */
mstime_t slave_reconf_sent_time; /* Time at which we sent SLAVE OF <new> */ mstime_t slave_reconf_sent_time; /* Time at which we sent SLAVE OF <new> */
struct sentinelRedisInstance *master; /* Master instance if it's slave. */ struct sentinelRedisInstance *master; /* Master instance if it's slave. */
char *slave_master_host; /* Master host as reported by INFO */ char *slave_master_host; /* Master host as reported by INFO */
...@@ -548,14 +549,12 @@ void initSentinel(void) { ...@@ -548,14 +549,12 @@ void initSentinel(void) {
server.sentinel_config = NULL; server.sentinel_config = NULL;
} }
/* This function gets called when the server is in Sentinel mode, started, /* This function is for checking whether sentinel config file has been set,
* loaded the configuration, and is ready for normal operations. */ * also checking whether we have write permissions. */
void sentinelIsRunning(void) { void sentinelCheckConfigFile(void) {
int j;
if (server.configfile == NULL) { if (server.configfile == NULL) {
serverLog(LL_WARNING, serverLog(LL_WARNING,
"Sentinel started without a config file. Exiting..."); "Sentinel needs config file on disk to save state. Exiting...");
exit(1); exit(1);
} else if (access(server.configfile,W_OK) == -1) { } else if (access(server.configfile,W_OK) == -1) {
serverLog(LL_WARNING, serverLog(LL_WARNING,
...@@ -563,6 +562,12 @@ void sentinelIsRunning(void) { ...@@ -563,6 +562,12 @@ void sentinelIsRunning(void) {
server.configfile,strerror(errno)); server.configfile,strerror(errno));
exit(1); exit(1);
} }
}
/* This function gets called when the server is in Sentinel mode, started,
* loaded the configuration, and is ready for normal operations. */
void sentinelIsRunning(void) {
int j;
/* If this Sentinel has yet no ID set in the configuration file, we /* If this Sentinel has yet no ID set in the configuration file, we
* pick a random one and persist the config on disk. From now on this * pick a random one and persist the config on disk. From now on this
...@@ -870,6 +875,7 @@ void sentinelRunPendingScripts(void) { ...@@ -870,6 +875,7 @@ void sentinelRunPendingScripts(void) {
sj->pid = 0; sj->pid = 0;
} else if (pid == 0) { } else if (pid == 0) {
/* Child */ /* Child */
tlsCleanup();
execve(sj->argv[0],sj->argv,environ); execve(sj->argv[0],sj->argv,environ);
/* If we are here an error occurred. */ /* If we are here an error occurred. */
_exit(2); /* Don't retry execution. */ _exit(2); /* Don't retry execution. */
...@@ -903,7 +909,7 @@ void sentinelCollectTerminatedScripts(void) { ...@@ -903,7 +909,7 @@ void sentinelCollectTerminatedScripts(void) {
int statloc; int statloc;
pid_t pid; pid_t pid;
while ((pid = wait3(&statloc,WNOHANG,NULL)) > 0) { while ((pid = waitpid(-1, &statloc, WNOHANG)) > 0) {
int exitcode = WEXITSTATUS(statloc); int exitcode = WEXITSTATUS(statloc);
int bysignal = 0; int bysignal = 0;
listNode *ln; listNode *ln;
...@@ -915,7 +921,7 @@ void sentinelCollectTerminatedScripts(void) { ...@@ -915,7 +921,7 @@ void sentinelCollectTerminatedScripts(void) {
ln = sentinelGetScriptListNodeByPid(pid); ln = sentinelGetScriptListNodeByPid(pid);
if (ln == NULL) { if (ln == NULL) {
serverLog(LL_WARNING,"wait3() returned a pid (%ld) we can't find in our scripts execution queue!", (long)pid); serverLog(LL_WARNING,"waitpid() returned a pid (%ld) we can't find in our scripts execution queue!", (long)pid);
continue; continue;
} }
sj = ln->value; sj = ln->value;
...@@ -1334,6 +1340,7 @@ sentinelRedisInstance *createSentinelRedisInstance(char *name, int flags, char * ...@@ -1334,6 +1340,7 @@ sentinelRedisInstance *createSentinelRedisInstance(char *name, int flags, char *
ri->auth_pass = NULL; ri->auth_pass = NULL;
ri->auth_user = NULL; ri->auth_user = NULL;
ri->slave_priority = SENTINEL_DEFAULT_SLAVE_PRIORITY; ri->slave_priority = SENTINEL_DEFAULT_SLAVE_PRIORITY;
ri->replica_announced = 1;
ri->slave_reconf_sent_time = 0; ri->slave_reconf_sent_time = 0;
ri->slave_master_host = NULL; ri->slave_master_host = NULL;
ri->slave_master_port = 0; ri->slave_master_port = 0;
...@@ -1719,19 +1726,12 @@ void sentinelPropagateDownAfterPeriod(sentinelRedisInstance *master) { ...@@ -1719,19 +1726,12 @@ void sentinelPropagateDownAfterPeriod(sentinelRedisInstance *master) {
} }
} }
char *sentinelGetInstanceTypeString(sentinelRedisInstance *ri) {
if (ri->flags & SRI_MASTER) return "master";
else if (ri->flags & SRI_SLAVE) return "slave";
else if (ri->flags & SRI_SENTINEL) return "sentinel";
else return "unknown";
}
/* This function is used in order to send commands to Redis instances: the /* This function is used in order to send commands to Redis instances: the
* commands we send from Sentinel may be renamed, a common case is a master * commands we send from Sentinel may be renamed, a common case is a master
* with CONFIG and SLAVEOF commands renamed for security concerns. In that * with CONFIG and SLAVEOF commands renamed for security concerns. In that
* case we check the ri->renamed_command table (or if the instance is a slave, * case we check the ri->renamed_command table (or if the instance is a slave,
* we check the one of the master), and map the command that we should send * we check the one of the master), and map the command that we should send
* to the set of renamed commads. However, if the command was not renamed, * to the set of renamed commands. However, if the command was not renamed,
* we just return "command" itself. */ * we just return "command" itself. */
char *sentinelInstanceMapCommand(sentinelRedisInstance *ri, char *command) { char *sentinelInstanceMapCommand(sentinelRedisInstance *ri, char *command) {
sds sc = sdsnew(command); sds sc = sdsnew(command);
...@@ -2045,12 +2045,12 @@ const char *sentinelHandleConfiguration(char **argv, int argc) { ...@@ -2045,12 +2045,12 @@ const char *sentinelHandleConfiguration(char **argv, int argc) {
} else if (!strcasecmp(argv[0],"resolve-hostnames") && argc == 2) { } else if (!strcasecmp(argv[0],"resolve-hostnames") && argc == 2) {
/* resolve-hostnames <yes|no> */ /* resolve-hostnames <yes|no> */
if ((sentinel.resolve_hostnames = yesnotoi(argv[1])) == -1) { if ((sentinel.resolve_hostnames = yesnotoi(argv[1])) == -1) {
return "Please specify yes or not for the resolve-hostnames option."; return "Please specify yes or no for the resolve-hostnames option.";
} }
} else if (!strcasecmp(argv[0],"announce-hostnames") && argc == 2) { } else if (!strcasecmp(argv[0],"announce-hostnames") && argc == 2) {
/* announce-hostnames <yes|no> */ /* announce-hostnames <yes|no> */
if ((sentinel.announce_hostnames = yesnotoi(argv[1])) == -1) { if ((sentinel.announce_hostnames = yesnotoi(argv[1])) == -1) {
return "Please specify yes or not for the announce-hostnames option."; return "Please specify yes or no for the announce-hostnames option.";
} }
} else { } else {
return "Unrecognized sentinel configuration statement."; return "Unrecognized sentinel configuration statement.";
...@@ -2320,8 +2320,8 @@ void sentinelFlushConfig(void) { ...@@ -2320,8 +2320,8 @@ void sentinelFlushConfig(void) {
return; return;
werr: werr:
if (fd != -1) close(fd);
serverLog(LL_WARNING,"WARNING: Sentinel was not able to save the new configuration on disk!!!: %s", strerror(errno)); serverLog(LL_WARNING,"WARNING: Sentinel was not able to save the new configuration on disk!!!: %s", strerror(errno));
if (fd != -1) close(fd);
} }
/* ====================== hiredis connection handling ======================= */ /* ====================== hiredis connection handling ======================= */
...@@ -2422,8 +2422,10 @@ void sentinelReconnectInstance(sentinelRedisInstance *ri) { ...@@ -2422,8 +2422,10 @@ void sentinelReconnectInstance(sentinelRedisInstance *ri) {
/* Commands connection. */ /* Commands connection. */
if (link->cc == NULL) { if (link->cc == NULL) {
link->cc = redisAsyncConnectBind(ri->addr->ip,ri->addr->port,NET_FIRST_BIND_ADDR); link->cc = redisAsyncConnectBind(ri->addr->ip,ri->addr->port,NET_FIRST_BIND_ADDR);
if (!link->cc->err) anetCloexec(link->cc->c.fd); if (link->cc && !link->cc->err) anetCloexec(link->cc->c.fd);
if (!link->cc->err && server.tls_replication && if (!link->cc) {
sentinelEvent(LL_DEBUG,"-cmd-link-reconnection",ri,"%@ #Failed to establish connection");
} else if (!link->cc->err && server.tls_replication &&
(instanceLinkNegotiateTLS(link->cc) == C_ERR)) { (instanceLinkNegotiateTLS(link->cc) == C_ERR)) {
sentinelEvent(LL_DEBUG,"-cmd-link-reconnection",ri,"%@ #Failed to initialize TLS"); sentinelEvent(LL_DEBUG,"-cmd-link-reconnection",ri,"%@ #Failed to initialize TLS");
instanceLinkCloseConnection(link,link->cc); instanceLinkCloseConnection(link,link->cc);
...@@ -2450,8 +2452,10 @@ void sentinelReconnectInstance(sentinelRedisInstance *ri) { ...@@ -2450,8 +2452,10 @@ void sentinelReconnectInstance(sentinelRedisInstance *ri) {
/* Pub / Sub */ /* Pub / Sub */
if ((ri->flags & (SRI_MASTER|SRI_SLAVE)) && link->pc == NULL) { if ((ri->flags & (SRI_MASTER|SRI_SLAVE)) && link->pc == NULL) {
link->pc = redisAsyncConnectBind(ri->addr->ip,ri->addr->port,NET_FIRST_BIND_ADDR); link->pc = redisAsyncConnectBind(ri->addr->ip,ri->addr->port,NET_FIRST_BIND_ADDR);
if (!link->pc->err) anetCloexec(link->pc->c.fd); if (link->pc && !link->pc->err) anetCloexec(link->pc->c.fd);
if (!link->pc->err && server.tls_replication && if (!link->pc) {
sentinelEvent(LL_DEBUG,"-pubsub-link-reconnection",ri,"%@ #Failed to establish connection");
} else if (!link->pc->err && server.tls_replication &&
(instanceLinkNegotiateTLS(link->pc) == C_ERR)) { (instanceLinkNegotiateTLS(link->pc) == C_ERR)) {
sentinelEvent(LL_DEBUG,"-pubsub-link-reconnection",ri,"%@ #Failed to initialize TLS"); sentinelEvent(LL_DEBUG,"-pubsub-link-reconnection",ri,"%@ #Failed to initialize TLS");
} else if (link->pc->err) { } else if (link->pc->err) {
...@@ -2624,6 +2628,10 @@ void sentinelRefreshInstanceInfo(sentinelRedisInstance *ri, const char *info) { ...@@ -2624,6 +2628,10 @@ void sentinelRefreshInstanceInfo(sentinelRedisInstance *ri, const char *info) {
/* slave_repl_offset:<offset> */ /* slave_repl_offset:<offset> */
if (sdslen(l) >= 18 && !memcmp(l,"slave_repl_offset:",18)) if (sdslen(l) >= 18 && !memcmp(l,"slave_repl_offset:",18))
ri->slave_repl_offset = strtoull(l+18,NULL,10); ri->slave_repl_offset = strtoull(l+18,NULL,10);
/* replica_announced:<announcement> */
if (sdslen(l) >= 18 && !memcmp(l,"replica_announced:",18))
ri->replica_announced = atoi(l+18);
} }
} }
ri->info_refresh = mstime(); ri->info_refresh = mstime();
...@@ -2644,8 +2652,7 @@ void sentinelRefreshInstanceInfo(sentinelRedisInstance *ri, const char *info) { ...@@ -2644,8 +2652,7 @@ void sentinelRefreshInstanceInfo(sentinelRedisInstance *ri, const char *info) {
((ri->flags & (SRI_MASTER|SRI_SLAVE)) == role) ? ((ri->flags & (SRI_MASTER|SRI_SLAVE)) == role) ?
"+role-change" : "-role-change", "+role-change" : "-role-change",
ri, "%@ new reported role is %s", ri, "%@ new reported role is %s",
role == SRI_MASTER ? "master" : "slave", role == SRI_MASTER ? "master" : "slave");
ri->flags & SRI_MASTER ? "master" : "slave");
} }
/* None of the following conditions are processed when in tilt mode, so /* None of the following conditions are processed when in tilt mode, so
...@@ -3291,6 +3298,8 @@ void addReplySentinelRedisInstance(client *c, sentinelRedisInstance *ri) { ...@@ -3291,6 +3298,8 @@ void addReplySentinelRedisInstance(client *c, sentinelRedisInstance *ri) {
if (ri->flags & SRI_RECONF_SENT) flags = sdscat(flags,"reconf_sent,"); if (ri->flags & SRI_RECONF_SENT) flags = sdscat(flags,"reconf_sent,");
if (ri->flags & SRI_RECONF_INPROG) flags = sdscat(flags,"reconf_inprog,"); if (ri->flags & SRI_RECONF_INPROG) flags = sdscat(flags,"reconf_inprog,");
if (ri->flags & SRI_RECONF_DONE) flags = sdscat(flags,"reconf_done,"); if (ri->flags & SRI_RECONF_DONE) flags = sdscat(flags,"reconf_done,");
if (ri->flags & SRI_FORCE_FAILOVER) flags = sdscat(flags,"force_failover,");
if (ri->flags & SRI_SCRIPT_KILL_SENT) flags = sdscat(flags,"script_kill_sent,");
if (sdslen(flags) != 0) sdsrange(flags,0,-2); /* remove last "," */ if (sdslen(flags) != 0) sdsrange(flags,0,-2); /* remove last "," */
addReplyBulkCString(c,flags); addReplyBulkCString(c,flags);
...@@ -3343,7 +3352,8 @@ void addReplySentinelRedisInstance(client *c, sentinelRedisInstance *ri) { ...@@ -3343,7 +3352,8 @@ void addReplySentinelRedisInstance(client *c, sentinelRedisInstance *ri) {
/* Masters and Slaves */ /* Masters and Slaves */
if (ri->flags & (SRI_MASTER|SRI_SLAVE)) { if (ri->flags & (SRI_MASTER|SRI_SLAVE)) {
addReplyBulkCString(c,"info-refresh"); addReplyBulkCString(c,"info-refresh");
addReplyBulkLongLong(c,mstime() - ri->info_refresh); addReplyBulkLongLong(c,
ri->info_refresh ? (mstime() - ri->info_refresh) : 0);
fields++; fields++;
addReplyBulkCString(c,"role-reported"); addReplyBulkCString(c,"role-reported");
...@@ -3423,6 +3433,10 @@ void addReplySentinelRedisInstance(client *c, sentinelRedisInstance *ri) { ...@@ -3423,6 +3433,10 @@ void addReplySentinelRedisInstance(client *c, sentinelRedisInstance *ri) {
addReplyBulkCString(c,"slave-repl-offset"); addReplyBulkCString(c,"slave-repl-offset");
addReplyBulkLongLong(c,ri->slave_repl_offset); addReplyBulkLongLong(c,ri->slave_repl_offset);
fields++; fields++;
addReplyBulkCString(c,"replica-announced");
addReplyBulkLongLong(c,ri->replica_announced);
fields++;
} }
/* Only sentinels */ /* Only sentinels */
...@@ -3448,15 +3462,20 @@ void addReplySentinelRedisInstance(client *c, sentinelRedisInstance *ri) { ...@@ -3448,15 +3462,20 @@ void addReplySentinelRedisInstance(client *c, sentinelRedisInstance *ri) {
void addReplyDictOfRedisInstances(client *c, dict *instances) { void addReplyDictOfRedisInstances(client *c, dict *instances) {
dictIterator *di; dictIterator *di;
dictEntry *de; dictEntry *de;
long slaves = 0;
void *replylen = addReplyDeferredLen(c);
di = dictGetIterator(instances); di = dictGetIterator(instances);
addReplyArrayLen(c,dictSize(instances));
while((de = dictNext(di)) != NULL) { while((de = dictNext(di)) != NULL) {
sentinelRedisInstance *ri = dictGetVal(de); sentinelRedisInstance *ri = dictGetVal(de);
/* don't announce unannounced replicas */
if (ri->flags & SRI_SLAVE && !ri->replica_announced) continue;
addReplySentinelRedisInstance(c,ri); addReplySentinelRedisInstance(c,ri);
slaves++;
} }
dictReleaseIterator(di); dictReleaseIterator(di);
setDeferredArrayLen(c, replylen, slaves);
} }
/* Lookup the named master into sentinel.masters. /* Lookup the named master into sentinel.masters.
...@@ -3710,17 +3729,7 @@ NULL ...@@ -3710,17 +3729,7 @@ NULL
ri = createSentinelRedisInstance(c->argv[2]->ptr,SRI_MASTER, ri = createSentinelRedisInstance(c->argv[2]->ptr,SRI_MASTER,
c->argv[3]->ptr,port,quorum,NULL); c->argv[3]->ptr,port,quorum,NULL);
if (ri == NULL) { if (ri == NULL) {
switch(errno) { addReplyError(c,sentinelCheckCreateInstanceErrors(SRI_MASTER));
case EBUSY:
addReplyError(c,"Duplicated master name");
break;
case EINVAL:
addReplyError(c,"Invalid port number");
break;
default:
addReplyError(c,"Unspecified error adding the instance");
break;
}
} else { } else {
sentinelFlushConfig(); sentinelFlushConfig();
sentinelEvent(LL_WARNING,"+monitor",ri,"%@ quorum %d",ri->quorum); sentinelEvent(LL_WARNING,"+monitor",ri,"%@ quorum %d",ri->quorum);
...@@ -3820,7 +3829,8 @@ NULL ...@@ -3820,7 +3829,8 @@ NULL
addReplyBulkCBuffer(c,ri->name,strlen(ri->name)); addReplyBulkCBuffer(c,ri->name,strlen(ri->name));
addReplyArrayLen(c,dictSize(ri->slaves) + 1); /* +1 for self */ addReplyArrayLen(c,dictSize(ri->slaves) + 1); /* +1 for self */
addReplyArrayLen(c,2); addReplyArrayLen(c,2);
addReplyLongLong(c, now - ri->info_refresh); addReplyLongLong(c,
ri->info_refresh ? (now - ri->info_refresh) : 0);
if (ri->info) if (ri->info)
addReplyBulkCBuffer(c,ri->info,sdslen(ri->info)); addReplyBulkCBuffer(c,ri->info,sdslen(ri->info));
else else
...@@ -3832,7 +3842,8 @@ NULL ...@@ -3832,7 +3842,8 @@ NULL
while ((sde = dictNext(sdi)) != NULL) { while ((sde = dictNext(sdi)) != NULL) {
sentinelRedisInstance *sri = dictGetVal(sde); sentinelRedisInstance *sri = dictGetVal(sde);
addReplyArrayLen(c,2); addReplyArrayLen(c,2);
addReplyLongLong(c, now - sri->info_refresh); addReplyLongLong(c,
ri->info_refresh ? (now - sri->info_refresh) : 0);
if (sri->info) if (sri->info)
addReplyBulkCBuffer(c,sri->info,sdslen(sri->info)); addReplyBulkCBuffer(c,sri->info,sdslen(sri->info));
else else
......
This diff is collapsed.
This diff is collapsed.
...@@ -201,7 +201,7 @@ void SHA1Final(unsigned char digest[20], SHA1_CTX* context) ...@@ -201,7 +201,7 @@ void SHA1Final(unsigned char digest[20], SHA1_CTX* context)
#define BUFSIZE 4096 #define BUFSIZE 4096
#define UNUSED(x) (void)(x) #define UNUSED(x) (void)(x)
int sha1Test(int argc, char **argv) int sha1Test(int argc, char **argv, int accurate)
{ {
SHA1_CTX ctx; SHA1_CTX ctx;
unsigned char hash[20], buf[BUFSIZE]; unsigned char hash[20], buf[BUFSIZE];
...@@ -209,6 +209,7 @@ int sha1Test(int argc, char **argv) ...@@ -209,6 +209,7 @@ int sha1Test(int argc, char **argv)
UNUSED(argc); UNUSED(argc);
UNUSED(argv); UNUSED(argv);
UNUSED(accurate);
for(i=0;i<BUFSIZE;i++) for(i=0;i<BUFSIZE;i++)
buf[i] = i; buf[i] = i;
......
...@@ -19,6 +19,6 @@ void SHA1Update(SHA1_CTX* context, const unsigned char* data, uint32_t len); ...@@ -19,6 +19,6 @@ void SHA1Update(SHA1_CTX* context, const unsigned char* data, uint32_t len);
void SHA1Final(unsigned char digest[20], SHA1_CTX* context); void SHA1Final(unsigned char digest[20], SHA1_CTX* context);
#ifdef REDIS_TEST #ifdef REDIS_TEST
int sha1Test(int argc, char **argv); int sha1Test(int argc, char **argv, int accurate);
#endif #endif
#endif #endif
...@@ -572,7 +572,7 @@ static int _hashZiplistEntryValidation(unsigned char *p, void *userdata) { ...@@ -572,7 +572,7 @@ static int _hashZiplistEntryValidation(unsigned char *p, void *userdata) {
return 1; return 1;
} }
/* Validate the integrity of the data stracture. /* Validate the integrity of the data structure.
* when `deep` is 0, only the integrity of the header is validated. * when `deep` is 0, only the integrity of the header is validated.
* when `deep` is 1, we scan all the entries one by one. */ * when `deep` is 1, we scan all the entries one by one. */
int hashZiplistValidateIntegrity(unsigned char *zl, size_t size, int deep) { int hashZiplistValidateIntegrity(unsigned char *zl, size_t size, int deep) {
......
...@@ -592,20 +592,14 @@ void lposCommand(client *c) { ...@@ -592,20 +592,14 @@ void lposCommand(client *c) {
} }
} else if (!strcasecmp(opt,"COUNT") && moreargs) { } else if (!strcasecmp(opt,"COUNT") && moreargs) {
j++; j++;
if (getLongFromObjectOrReply(c, c->argv[j], &count, NULL) != C_OK) if (getPositiveLongFromObjectOrReply(c, c->argv[j], &count,
"COUNT can't be negative") != C_OK)
return; return;
if (count < 0) {
addReplyError(c,"COUNT can't be negative");
return;
}
} else if (!strcasecmp(opt,"MAXLEN") && moreargs) { } else if (!strcasecmp(opt,"MAXLEN") && moreargs) {
j++; j++;
if (getLongFromObjectOrReply(c, c->argv[j], &maxlen, NULL) != C_OK) if (getPositiveLongFromObjectOrReply(c, c->argv[j], &maxlen,
"MAXLEN can't be negative") != C_OK)
return; return;
if (maxlen < 0) {
addReplyError(c,"MAXLEN can't be negative");
return;
}
} else { } else {
addReplyErrorObject(c,shared.syntaxerr); addReplyErrorObject(c,shared.syntaxerr);
return; return;
......
...@@ -861,7 +861,7 @@ int64_t streamTrimByID(stream *s, streamID minid, int approx) { ...@@ -861,7 +861,7 @@ int64_t streamTrimByID(stream *s, streamID minid, int approx) {
return streamTrim(s, &args); return streamTrim(s, &args);
} }
/* Parse the arguements of XADD/XTRIM. /* Parse the arguments of XADD/XTRIM.
* *
* See streamAddTrimArgs for more details about the arguments handled. * See streamAddTrimArgs for more details about the arguments handled.
* *
...@@ -1313,7 +1313,8 @@ void streamLastValidID(stream *s, streamID *maxid) ...@@ -1313,7 +1313,8 @@ void streamLastValidID(stream *s, streamID *maxid)
streamIterator si; streamIterator si;
streamIteratorStart(&si,s,NULL,NULL,1); streamIteratorStart(&si,s,NULL,NULL,1);
int64_t numfields; int64_t numfields;
streamIteratorGetID(&si,maxid,&numfields); if (!streamIteratorGetID(&si,maxid,&numfields) && s->length)
serverPanic("Corrupt stream, length is %llu, but no max id", (unsigned long long)s->length);
streamIteratorStop(&si); streamIteratorStop(&si);
} }
...@@ -3050,12 +3051,8 @@ void xautoclaimCommand(client *c) { ...@@ -3050,12 +3051,8 @@ void xautoclaimCommand(client *c) {
int moreargs = (c->argc-1) - j; /* Number of additional arguments. */ int moreargs = (c->argc-1) - j; /* Number of additional arguments. */
char *opt = c->argv[j]->ptr; char *opt = c->argv[j]->ptr;
if (!strcasecmp(opt,"COUNT") && moreargs) { if (!strcasecmp(opt,"COUNT") && moreargs) {
if (getPositiveLongFromObjectOrReply(c,c->argv[j+1],&count,NULL) != C_OK) if (getRangeLongFromObjectOrReply(c,c->argv[j+1],1,LONG_MAX,&count,"COUNT must be > 0") != C_OK)
return; return;
if (count == 0) {
addReplyError(c,"COUNT must be > 0");
return;
}
j++; j++;
} else if (!strcasecmp(opt,"JUSTID")) { } else if (!strcasecmp(opt,"JUSTID")) {
justid = 1; justid = 1;
...@@ -3120,6 +3117,8 @@ void xautoclaimCommand(client *c) { ...@@ -3120,6 +3117,8 @@ void xautoclaimCommand(client *c) {
/* Update the consumer and idle time. */ /* Update the consumer and idle time. */
nack->delivery_time = now; nack->delivery_time = now;
/* Increment the delivery attempts counter unless JUSTID option provided */
if (!justid)
nack->delivery_count++; nack->delivery_count++;
if (nack->consumer != consumer) { if (nack->consumer != consumer) {
...@@ -3148,6 +3147,9 @@ void xautoclaimCommand(client *c) { ...@@ -3148,6 +3147,9 @@ void xautoclaimCommand(client *c) {
server.dirty++; server.dirty++;
} }
/* We need to return the next entry as a cursor for the next XAUTOCLAIM call */
raxNext(&ri);
streamID endid; streamID endid;
if (raxEOF(&ri)) { if (raxEOF(&ri)) {
endid.ms = endid.seq = 0; endid.ms = endid.seq = 0;
...@@ -3545,8 +3547,8 @@ NULL ...@@ -3545,8 +3547,8 @@ NULL
} }
} }
/* Validate the integrity stream listpack entries stracture. Both in term of a /* Validate the integrity stream listpack entries structure. Both in term of a
* valid listpack, but also that the stracture of the entires matches a valid * valid listpack, but also that the structure of the entires matches a valid
* stream. return 1 if valid 0 if not valid. */ * stream. return 1 if valid 0 if not valid. */
int streamValidateListpackIntegrity(unsigned char *lp, size_t size, int deep) { int streamValidateListpackIntegrity(unsigned char *lp, size_t size, int deep) {
int valid_record; int valid_record;
......
...@@ -723,7 +723,7 @@ void stralgoCommand(client *c) { ...@@ -723,7 +723,7 @@ void stralgoCommand(client *c) {
} }
} }
/* STRALGO <algo> [IDX] [MINMATCHLEN <len>] [WITHMATCHLEN] /* STRALGO <algo> [IDX] [LEN] [MINMATCHLEN <len>] [WITHMATCHLEN]
* STRINGS <string> <string> | KEYS <keya> <keyb> * STRINGS <string> <string> | KEYS <keya> <keyb>
*/ */
void stralgoLCS(client *c) { void stralgoLCS(client *c) {
......
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