Unverified Commit 61297585 authored by Salvatore Sanfilippo's avatar Salvatore Sanfilippo Committed by GitHub
Browse files

Merge branch 'unstable' into modules_fork

parents 83e87bac fddc4757
...@@ -138,6 +138,7 @@ int clusterLoadConfig(char *filename) { ...@@ -138,6 +138,7 @@ int clusterLoadConfig(char *filename) {
/* Handle the special "vars" line. Don't pretend it is the last /* Handle the special "vars" line. Don't pretend it is the last
* line even if it actually is when generated by Redis. */ * line even if it actually is when generated by Redis. */
if (strcasecmp(argv[0],"vars") == 0) { if (strcasecmp(argv[0],"vars") == 0) {
if (!(argc % 2)) goto fmterr;
for (j = 1; j < argc; j += 2) { for (j = 1; j < argc; j += 2) {
if (strcasecmp(argv[j],"currentEpoch") == 0) { if (strcasecmp(argv[j],"currentEpoch") == 0) {
server.cluster->currentEpoch = server.cluster->currentEpoch =
...@@ -4251,12 +4252,9 @@ NULL ...@@ -4251,12 +4252,9 @@ NULL
} }
} else if (!strcasecmp(c->argv[1]->ptr,"nodes") && c->argc == 2) { } else if (!strcasecmp(c->argv[1]->ptr,"nodes") && c->argc == 2) {
/* CLUSTER NODES */ /* CLUSTER NODES */
robj *o; sds nodes = clusterGenNodesDescription(0);
sds ci = clusterGenNodesDescription(0); addReplyVerbatim(c,nodes,sdslen(nodes),"txt");
sdsfree(nodes);
o = createObject(OBJ_STRING,ci);
addReplyBulk(c,o);
decrRefCount(o);
} else if (!strcasecmp(c->argv[1]->ptr,"myid") && c->argc == 2) { } else if (!strcasecmp(c->argv[1]->ptr,"myid") && c->argc == 2) {
/* CLUSTER MYID */ /* CLUSTER MYID */
addReplyBulkCBuffer(c,myself->name, CLUSTER_NAMELEN); addReplyBulkCBuffer(c,myself->name, CLUSTER_NAMELEN);
...@@ -4498,10 +4496,8 @@ NULL ...@@ -4498,10 +4496,8 @@ NULL
"cluster_stats_messages_received:%lld\r\n", tot_msg_received); "cluster_stats_messages_received:%lld\r\n", tot_msg_received);
/* Produce the reply protocol. */ /* Produce the reply protocol. */
addReplySds(c,sdscatprintf(sdsempty(),"$%lu\r\n", addReplyVerbatim(c,info,sdslen(info),"txt");
(unsigned long)sdslen(info))); sdsfree(info);
addReplySds(c,info);
addReply(c,shared.crlf);
} else if (!strcasecmp(c->argv[1]->ptr,"saveconfig") && c->argc == 2) { } else if (!strcasecmp(c->argv[1]->ptr,"saveconfig") && c->argc == 2) {
int retval = clusterSaveConfig(1); int retval = clusterSaveConfig(1);
...@@ -4832,7 +4828,7 @@ int verifyDumpPayload(unsigned char *p, size_t len) { ...@@ -4832,7 +4828,7 @@ int verifyDumpPayload(unsigned char *p, size_t len) {
* DUMP is actually not used by Redis Cluster but it is the obvious * DUMP is actually not used by Redis Cluster but it is the obvious
* complement of RESTORE and can be useful for different applications. */ * complement of RESTORE and can be useful for different applications. */
void dumpCommand(client *c) { void dumpCommand(client *c) {
robj *o, *dumpobj; robj *o;
rio payload; rio payload;
/* Check if the key is here. */ /* Check if the key is here. */
...@@ -4845,9 +4841,7 @@ void dumpCommand(client *c) { ...@@ -4845,9 +4841,7 @@ void dumpCommand(client *c) {
createDumpPayload(&payload,o,c->argv[1]); createDumpPayload(&payload,o,c->argv[1]);
/* Transfer to the client */ /* Transfer to the client */
dumpobj = createObject(OBJ_STRING,payload.io.buffer.ptr); addReplyBulkSds(c,payload.io.buffer.ptr);
addReplyBulk(c,dumpobj);
decrRefCount(dumpobj);
return; return;
} }
......
...@@ -672,6 +672,10 @@ void loadServerConfigFromString(char *config) { ...@@ -672,6 +672,10 @@ void loadServerConfigFromString(char *config) {
server.lua_time_limit = strtoll(argv[1],NULL,10); server.lua_time_limit = strtoll(argv[1],NULL,10);
} else if (!strcasecmp(argv[0],"lua-replicate-commands") && argc == 2) { } else if (!strcasecmp(argv[0],"lua-replicate-commands") && argc == 2) {
server.lua_always_replicate_commands = yesnotoi(argv[1]); server.lua_always_replicate_commands = yesnotoi(argv[1]);
if (server.lua_always_replicate_commands == -1) {
err = "argument must be 'yes' or 'no'";
goto loaderr;
}
} else if (!strcasecmp(argv[0],"slowlog-log-slower-than") && } else if (!strcasecmp(argv[0],"slowlog-log-slower-than") &&
argc == 2) argc == 2)
{ {
...@@ -686,6 +690,17 @@ void loadServerConfigFromString(char *config) { ...@@ -686,6 +690,17 @@ void loadServerConfigFromString(char *config) {
} }
} else if (!strcasecmp(argv[0],"slowlog-max-len") && argc == 2) { } else if (!strcasecmp(argv[0],"slowlog-max-len") && argc == 2) {
server.slowlog_max_len = strtoll(argv[1],NULL,10); server.slowlog_max_len = strtoll(argv[1],NULL,10);
} else if (!strcasecmp(argv[0],"tracking-table-max-fill") &&
argc == 2)
{
server.tracking_table_max_fill = strtoll(argv[1],NULL,10);
if (server.tracking_table_max_fill > 100 ||
server.tracking_table_max_fill < 0)
{
err = "The tracking table fill percentage must be an "
"integer between 0 and 100";
goto loaderr;
}
} else if (!strcasecmp(argv[0],"client-output-buffer-limit") && } else if (!strcasecmp(argv[0],"client-output-buffer-limit") &&
argc == 5) argc == 5)
{ {
...@@ -1133,6 +1148,8 @@ void configSetCommand(client *c) { ...@@ -1133,6 +1148,8 @@ void configSetCommand(client *c) {
"slowlog-max-len",ll,0,LONG_MAX) { "slowlog-max-len",ll,0,LONG_MAX) {
/* Cast to unsigned. */ /* Cast to unsigned. */
server.slowlog_max_len = (unsigned long)ll; server.slowlog_max_len = (unsigned long)ll;
} config_set_numerical_field(
"tracking-table-max-fill",server.tracking_table_max_fill,0,100) {
} config_set_numerical_field( } config_set_numerical_field(
"latency-monitor-threshold",server.latency_monitor_threshold,0,LLONG_MAX){ "latency-monitor-threshold",server.latency_monitor_threshold,0,LLONG_MAX){
} config_set_numerical_field( } config_set_numerical_field(
...@@ -1338,8 +1355,8 @@ void configGetCommand(client *c) { ...@@ -1338,8 +1355,8 @@ void configGetCommand(client *c) {
server.slowlog_log_slower_than); server.slowlog_log_slower_than);
config_get_numerical_field("latency-monitor-threshold", config_get_numerical_field("latency-monitor-threshold",
server.latency_monitor_threshold); server.latency_monitor_threshold);
config_get_numerical_field("slowlog-max-len", config_get_numerical_field("slowlog-max-len", server.slowlog_max_len);
server.slowlog_max_len); config_get_numerical_field("tracking-table-max-fill", server.tracking_table_max_fill);
config_get_numerical_field("port",server.port); config_get_numerical_field("port",server.port);
config_get_numerical_field("cluster-announce-port",server.cluster_announce_port); config_get_numerical_field("cluster-announce-port",server.cluster_announce_port);
config_get_numerical_field("cluster-announce-bus-port",server.cluster_announce_bus_port); config_get_numerical_field("cluster-announce-bus-port",server.cluster_announce_bus_port);
...@@ -1470,12 +1487,10 @@ void configGetCommand(client *c) { ...@@ -1470,12 +1487,10 @@ void configGetCommand(client *c) {
matches++; matches++;
} }
if (stringmatch(pattern,"notify-keyspace-events",1)) { if (stringmatch(pattern,"notify-keyspace-events",1)) {
robj *flagsobj = createObject(OBJ_STRING, sds flags = keyspaceEventsFlagsToString(server.notify_keyspace_events);
keyspaceEventsFlagsToString(server.notify_keyspace_events));
addReplyBulkCString(c,"notify-keyspace-events"); addReplyBulkCString(c,"notify-keyspace-events");
addReplyBulk(c,flagsobj); addReplyBulkSds(c,flags);
decrRefCount(flagsobj);
matches++; matches++;
} }
if (stringmatch(pattern,"bind",1)) { if (stringmatch(pattern,"bind",1)) {
...@@ -2167,6 +2182,7 @@ int rewriteConfig(char *path) { ...@@ -2167,6 +2182,7 @@ int rewriteConfig(char *path) {
rewriteConfigNumericalOption(state,"slowlog-log-slower-than",server.slowlog_log_slower_than,CONFIG_DEFAULT_SLOWLOG_LOG_SLOWER_THAN); rewriteConfigNumericalOption(state,"slowlog-log-slower-than",server.slowlog_log_slower_than,CONFIG_DEFAULT_SLOWLOG_LOG_SLOWER_THAN);
rewriteConfigNumericalOption(state,"latency-monitor-threshold",server.latency_monitor_threshold,CONFIG_DEFAULT_LATENCY_MONITOR_THRESHOLD); rewriteConfigNumericalOption(state,"latency-monitor-threshold",server.latency_monitor_threshold,CONFIG_DEFAULT_LATENCY_MONITOR_THRESHOLD);
rewriteConfigNumericalOption(state,"slowlog-max-len",server.slowlog_max_len,CONFIG_DEFAULT_SLOWLOG_MAX_LEN); rewriteConfigNumericalOption(state,"slowlog-max-len",server.slowlog_max_len,CONFIG_DEFAULT_SLOWLOG_MAX_LEN);
rewriteConfigNumericalOption(state,"tracking-table-max-fill",server.tracking_table_max_fill,CONFIG_DEFAULT_TRACKING_TABLE_MAX_FILL);
rewriteConfigNotifykeyspaceeventsOption(state); rewriteConfigNotifykeyspaceeventsOption(state);
rewriteConfigNumericalOption(state,"hash-max-ziplist-entries",server.hash_max_ziplist_entries,OBJ_HASH_MAX_ZIPLIST_ENTRIES); rewriteConfigNumericalOption(state,"hash-max-ziplist-entries",server.hash_max_ziplist_entries,OBJ_HASH_MAX_ZIPLIST_ENTRIES);
rewriteConfigNumericalOption(state,"hash-max-ziplist-value",server.hash_max_ziplist_value,OBJ_HASH_MAX_ZIPLIST_VALUE); rewriteConfigNumericalOption(state,"hash-max-ziplist-value",server.hash_max_ziplist_value,OBJ_HASH_MAX_ZIPLIST_VALUE);
......
...@@ -350,6 +350,11 @@ long long emptyDbGeneric(redisDb *dbarray, int dbnum, int flags, void(callback)( ...@@ -350,6 +350,11 @@ long long emptyDbGeneric(redisDb *dbarray, int dbnum, int flags, void(callback)(
return -1; return -1;
} }
/* Make sure the WATCHed keys are affected by the FLUSH* commands.
* Note that we need to call the function while the keys are still
* there. */
signalFlushedDb(dbnum);
int startdb, enddb; int startdb, enddb;
if (dbnum == -1) { if (dbnum == -1) {
startdb = 0; startdb = 0;
...@@ -409,11 +414,12 @@ long long dbTotalServerKeyCount() { ...@@ -409,11 +414,12 @@ long long dbTotalServerKeyCount() {
void signalModifiedKey(redisDb *db, robj *key) { void signalModifiedKey(redisDb *db, robj *key) {
touchWatchedKey(db,key); touchWatchedKey(db,key);
if (server.tracking_clients) trackingInvalidateKey(key); trackingInvalidateKey(key);
} }
void signalFlushedDb(int dbid) { void signalFlushedDb(int dbid) {
touchWatchedKeysOnFlush(dbid); touchWatchedKeysOnFlush(dbid);
trackingInvalidateKeysOnFlush(dbid);
} }
/*----------------------------------------------------------------------------- /*-----------------------------------------------------------------------------
...@@ -449,7 +455,6 @@ void flushdbCommand(client *c) { ...@@ -449,7 +455,6 @@ void flushdbCommand(client *c) {
int flags; int flags;
if (getFlushCommandFlags(c,&flags) == C_ERR) return; if (getFlushCommandFlags(c,&flags) == C_ERR) return;
signalFlushedDb(c->db->id);
server.dirty += emptyDb(c->db->id,flags,NULL); server.dirty += emptyDb(c->db->id,flags,NULL);
addReply(c,shared.ok); addReply(c,shared.ok);
} }
...@@ -461,7 +466,6 @@ void flushallCommand(client *c) { ...@@ -461,7 +466,6 @@ void flushallCommand(client *c) {
int flags; int flags;
if (getFlushCommandFlags(c,&flags) == C_ERR) return; if (getFlushCommandFlags(c,&flags) == C_ERR) return;
signalFlushedDb(-1);
server.dirty += emptyDb(-1,flags,NULL); server.dirty += emptyDb(-1,flags,NULL);
addReply(c,shared.ok); addReply(c,shared.ok);
if (server.rdb_child_pid != -1) killRDBChild(); if (server.rdb_child_pid != -1) killRDBChild();
......
...@@ -638,7 +638,8 @@ NULL ...@@ -638,7 +638,8 @@ NULL
dictGetStats(buf,sizeof(buf),server.db[dbid].expires); dictGetStats(buf,sizeof(buf),server.db[dbid].expires);
stats = sdscat(stats,buf); stats = sdscat(stats,buf);
addReplyBulkSds(c,stats); addReplyVerbatim(c,stats,sdslen(stats),"txt");
sdsfree(stats);
} else if (!strcasecmp(c->argv[1]->ptr,"htstats-key") && c->argc == 3) { } else if (!strcasecmp(c->argv[1]->ptr,"htstats-key") && c->argc == 3) {
robj *o; robj *o;
dict *ht = NULL; dict *ht = NULL;
...@@ -665,7 +666,7 @@ NULL ...@@ -665,7 +666,7 @@ NULL
} else { } else {
char buf[4096]; char buf[4096];
dictGetStats(buf,sizeof(buf),ht); dictGetStats(buf,sizeof(buf),ht);
addReplyBulkCString(c,buf); addReplyVerbatim(c,buf,strlen(buf),"txt");
} }
} else if (!strcasecmp(c->argv[1]->ptr,"change-repl-id") && c->argc == 2) { } else if (!strcasecmp(c->argv[1]->ptr,"change-repl-id") && c->argc == 2) {
serverLog(LL_WARNING,"Changing replication IDs after receiving DEBUG change-repl-id"); serverLog(LL_WARNING,"Changing replication IDs after receiving DEBUG change-repl-id");
...@@ -1110,6 +1111,33 @@ void logRegisters(ucontext_t *uc) { ...@@ -1110,6 +1111,33 @@ void logRegisters(ucontext_t *uc) {
(unsigned long) uc->uc_mcontext.mc_cs (unsigned long) uc->uc_mcontext.mc_cs
); );
logStackContent((void**)uc->uc_mcontext.mc_rsp); logStackContent((void**)uc->uc_mcontext.mc_rsp);
#elif defined(__aarch64__) /* Linux AArch64 */
serverLog(LL_WARNING,
"\n"
"X18:%016lx X19:%016lx\nX20:%016lx X21:%016lx\n"
"X22:%016lx X23:%016lx\nX24:%016lx X25:%016lx\n"
"X26:%016lx X27:%016lx\nX28:%016lx X29:%016lx\n"
"X30:%016lx\n"
"pc:%016lx sp:%016lx\npstate:%016lx fault_address:%016lx\n",
(unsigned long) uc->uc_mcontext.regs[18],
(unsigned long) uc->uc_mcontext.regs[19],
(unsigned long) uc->uc_mcontext.regs[20],
(unsigned long) uc->uc_mcontext.regs[21],
(unsigned long) uc->uc_mcontext.regs[22],
(unsigned long) uc->uc_mcontext.regs[23],
(unsigned long) uc->uc_mcontext.regs[24],
(unsigned long) uc->uc_mcontext.regs[25],
(unsigned long) uc->uc_mcontext.regs[26],
(unsigned long) uc->uc_mcontext.regs[27],
(unsigned long) uc->uc_mcontext.regs[28],
(unsigned long) uc->uc_mcontext.regs[29],
(unsigned long) uc->uc_mcontext.regs[30],
(unsigned long) uc->uc_mcontext.pc,
(unsigned long) uc->uc_mcontext.sp,
(unsigned long) uc->uc_mcontext.pstate,
(unsigned long) uc->uc_mcontext.fault_address
);
logStackContent((void**)uc->uc_mcontext.sp);
#else #else
serverLog(LL_WARNING, serverLog(LL_WARNING,
" Dumping of registers not supported for this OS/arch"); " Dumping of registers not supported for this OS/arch");
......
...@@ -64,7 +64,7 @@ int activeExpireCycleTryExpire(redisDb *db, dictEntry *de, long long now) { ...@@ -64,7 +64,7 @@ int activeExpireCycleTryExpire(redisDb *db, dictEntry *de, long long now) {
dbSyncDelete(db,keyobj); dbSyncDelete(db,keyobj);
notifyKeyspaceEvent(NOTIFY_EXPIRED, notifyKeyspaceEvent(NOTIFY_EXPIRED,
"expired",keyobj,db->id); "expired",keyobj,db->id);
if (server.tracking_clients) trackingInvalidateKey(keyobj); trackingInvalidateKey(keyobj);
decrRefCount(keyobj); decrRefCount(keyobj);
server.stat_expiredkeys++; server.stat_expiredkeys++;
return 1; return 1;
......
...@@ -700,7 +700,7 @@ int hllSparseSet(robj *o, long index, uint8_t count) { ...@@ -700,7 +700,7 @@ int hllSparseSet(robj *o, long index, uint8_t count) {
p += oplen; p += oplen;
first += span; first += span;
} }
if (span == 0) return -1; /* Invalid format. */ if (span == 0 || p >= end) return -1; /* Invalid format. */
next = HLL_SPARSE_IS_XZERO(p) ? p+2 : p+1; next = HLL_SPARSE_IS_XZERO(p) ? p+2 : p+1;
if (next >= end) next = NULL; if (next >= end) next = NULL;
...@@ -1242,7 +1242,7 @@ void pfcountCommand(client *c) { ...@@ -1242,7 +1242,7 @@ void pfcountCommand(client *c) {
if (o == NULL) continue; /* Assume empty HLL for non existing var.*/ if (o == NULL) continue; /* Assume empty HLL for non existing var.*/
if (isHLLObjectOrReply(c,o) != C_OK) return; if (isHLLObjectOrReply(c,o) != C_OK) return;
/* Merge with this HLL with our 'max' HHL by setting max[i] /* Merge with this HLL with our 'max' HLL by setting max[i]
* to MAX(max[i],hll[i]). */ * to MAX(max[i],hll[i]). */
if (hllMerge(registers,o) == C_ERR) { if (hllMerge(registers,o) == C_ERR) {
addReplySds(c,sdsnew(invalid_hll_err)); addReplySds(c,sdsnew(invalid_hll_err));
...@@ -1329,7 +1329,7 @@ void pfmergeCommand(client *c) { ...@@ -1329,7 +1329,7 @@ void pfmergeCommand(client *c) {
hdr = o->ptr; hdr = o->ptr;
if (hdr->encoding == HLL_DENSE) use_dense = 1; if (hdr->encoding == HLL_DENSE) use_dense = 1;
/* Merge with this HLL with our 'max' HHL by setting max[i] /* Merge with this HLL with our 'max' HLL by setting max[i]
* to MAX(max[i],hll[i]). */ * to MAX(max[i],hll[i]). */
if (hllMerge(max,o) == C_ERR) { if (hllMerge(max,o) == C_ERR) {
addReplySds(c,sdsnew(invalid_hll_err)); addReplySds(c,sdsnew(invalid_hll_err));
......
...@@ -599,7 +599,7 @@ NULL ...@@ -599,7 +599,7 @@ NULL
event = dictGetKey(de); event = dictGetKey(de);
graph = latencyCommandGenSparkeline(event,ts); graph = latencyCommandGenSparkeline(event,ts);
addReplyBulkCString(c,graph); addReplyVerbatim(c,graph,sdslen(graph),"txt");
sdsfree(graph); sdsfree(graph);
} else if (!strcasecmp(c->argv[1]->ptr,"latest") && c->argc == 2) { } else if (!strcasecmp(c->argv[1]->ptr,"latest") && c->argc == 2) {
/* LATENCY LATEST */ /* LATENCY LATEST */
...@@ -608,7 +608,7 @@ NULL ...@@ -608,7 +608,7 @@ NULL
/* LATENCY DOCTOR */ /* LATENCY DOCTOR */
sds report = createLatencyReport(); sds report = createLatencyReport();
addReplyBulkCBuffer(c,report,sdslen(report)); addReplyVerbatim(c,report,sdslen(report),"txt");
sdsfree(report); sdsfree(report);
} else if (!strcasecmp(c->argv[1]->ptr,"reset") && c->argc >= 2) { } else if (!strcasecmp(c->argv[1]->ptr,"reset") && c->argc >= 2) {
/* LATENCY RESET */ /* LATENCY RESET */
......
...@@ -43,7 +43,8 @@ void lolwutUnstableCommand(client *c) { ...@@ -43,7 +43,8 @@ void lolwutUnstableCommand(client *c) {
sds rendered = sdsnew("Redis ver. "); sds rendered = sdsnew("Redis ver. ");
rendered = sdscat(rendered,REDIS_VERSION); rendered = sdscat(rendered,REDIS_VERSION);
rendered = sdscatlen(rendered,"\n",1); rendered = sdscatlen(rendered,"\n",1);
addReplyBulkSds(c,rendered); addReplyVerbatim(c,rendered,sdslen(rendered),"txt");
sdsfree(rendered);
} }
void lolwutCommand(client *c) { void lolwutCommand(client *c) {
......
...@@ -277,6 +277,7 @@ void lolwut5Command(client *c) { ...@@ -277,6 +277,7 @@ void lolwut5Command(client *c) {
"\nGeorg Nees - schotter, plotter on paper, 1968. Redis ver. "); "\nGeorg Nees - schotter, plotter on paper, 1968. Redis ver. ");
rendered = sdscat(rendered,REDIS_VERSION); rendered = sdscat(rendered,REDIS_VERSION);
rendered = sdscatlen(rendered,"\n",1); rendered = sdscatlen(rendered,"\n",1);
addReplyBulkSds(c,rendered); addReplyVerbatim(c,rendered,sdslen(rendered),"txt");
sdsfree(rendered);
lwFreeCanvas(canvas); lwFreeCanvas(canvas);
} }
...@@ -29,6 +29,7 @@ ...@@ -29,6 +29,7 @@
#include "server.h" #include "server.h"
#include "cluster.h" #include "cluster.h"
#include "rdb.h"
#include <dlfcn.h> #include <dlfcn.h>
#include <wait.h> #include <wait.h>
...@@ -52,6 +53,7 @@ struct RedisModule { ...@@ -52,6 +53,7 @@ struct RedisModule {
list *using; /* List of modules we use some APIs of. */ list *using; /* List of modules we use some APIs of. */
list *filters; /* List of filters the module has registered. */ list *filters; /* List of filters the module has registered. */
int in_call; /* RM_Call() nesting level */ int in_call; /* RM_Call() nesting level */
int options; /* Module options and capabilities. */
}; };
typedef struct RedisModule RedisModule; typedef struct RedisModule RedisModule;
...@@ -780,6 +782,19 @@ long long RM_Milliseconds(void) { ...@@ -780,6 +782,19 @@ long long RM_Milliseconds(void) {
return mstime(); return mstime();
} }
/* Set flags defining capabilities or behavior bit flags.
*
* REDISMODULE_OPTIONS_HANDLE_IO_ERRORS:
* Generally, modules don't need to bother with this, as the process will just
* terminate if a read error happens, however, setting this flag would allow
* repl-diskless-load to work if enabled.
* The module should use RedisModule_IsIOError after reads, before using the
* data that was read, and in case of error, propagate it upwards, and also be
* able to release the partially populated value and all it's allocations. */
void RM_SetModuleOptions(RedisModuleCtx *ctx, int options) {
ctx->module->options = options;
}
/* -------------------------------------------------------------------------- /* --------------------------------------------------------------------------
* Automatic memory management for modules * Automatic memory management for modules
* -------------------------------------------------------------------------- */ * -------------------------------------------------------------------------- */
...@@ -2397,7 +2412,7 @@ int RM_HashSet(RedisModuleKey *key, int flags, ...) { ...@@ -2397,7 +2412,7 @@ int RM_HashSet(RedisModuleKey *key, int flags, ...) {
* *
* REDISMODULE_HASH_EXISTS: instead of setting the value of the field * REDISMODULE_HASH_EXISTS: instead of setting the value of the field
* expecting a RedisModuleString pointer to pointer, the function just * expecting a RedisModuleString pointer to pointer, the function just
* reports if the field esists or not and expects an integer pointer * reports if the field exists or not and expects an integer pointer
* as the second element of each pair. * as the second element of each pair.
* *
* Example of REDISMODULE_HASH_CFIELD: * Example of REDISMODULE_HASH_CFIELD:
...@@ -3087,6 +3102,11 @@ moduleType *RM_CreateDataType(RedisModuleCtx *ctx, const char *name, int encver, ...@@ -3087,6 +3102,11 @@ moduleType *RM_CreateDataType(RedisModuleCtx *ctx, const char *name, int encver,
moduleTypeMemUsageFunc mem_usage; moduleTypeMemUsageFunc mem_usage;
moduleTypeDigestFunc digest; moduleTypeDigestFunc digest;
moduleTypeFreeFunc free; moduleTypeFreeFunc free;
struct {
moduleTypeAuxLoadFunc aux_load;
moduleTypeAuxSaveFunc aux_save;
int aux_save_triggers;
} v2;
} *tms = (struct typemethods*) typemethods_ptr; } *tms = (struct typemethods*) typemethods_ptr;
moduleType *mt = zcalloc(sizeof(*mt)); moduleType *mt = zcalloc(sizeof(*mt));
...@@ -3098,6 +3118,11 @@ moduleType *RM_CreateDataType(RedisModuleCtx *ctx, const char *name, int encver, ...@@ -3098,6 +3118,11 @@ moduleType *RM_CreateDataType(RedisModuleCtx *ctx, const char *name, int encver,
mt->mem_usage = tms->mem_usage; mt->mem_usage = tms->mem_usage;
mt->digest = tms->digest; mt->digest = tms->digest;
mt->free = tms->free; mt->free = tms->free;
if (tms->version >= 2) {
mt->aux_load = tms->v2.aux_load;
mt->aux_save = tms->v2.aux_save;
mt->aux_save_triggers = tms->v2.aux_save_triggers;
}
memcpy(mt->name,name,sizeof(mt->name)); memcpy(mt->name,name,sizeof(mt->name));
listAddNodeTail(ctx->module->types,mt); listAddNodeTail(ctx->module->types,mt);
return mt; return mt;
...@@ -3148,9 +3173,14 @@ void *RM_ModuleTypeGetValue(RedisModuleKey *key) { ...@@ -3148,9 +3173,14 @@ void *RM_ModuleTypeGetValue(RedisModuleKey *key) {
* RDB loading and saving functions * RDB loading and saving functions
* -------------------------------------------------------------------------- */ * -------------------------------------------------------------------------- */
/* Called when there is a load error in the context of a module. This cannot /* Called when there is a load error in the context of a module. On some
* be recovered like for the built-in types. */ * modules this cannot be recovered, but if the module declared capability
* to handle errors, we'll raise a flag rather than exiting. */
void moduleRDBLoadError(RedisModuleIO *io) { void moduleRDBLoadError(RedisModuleIO *io) {
if (io->ctx->module->options & REDISMODULE_OPTIONS_HANDLE_IO_ERRORS) {
io->error = 1;
return;
}
serverLog(LL_WARNING, serverLog(LL_WARNING,
"Error loading data from RDB (short read or EOF). " "Error loading data from RDB (short read or EOF). "
"Read performed by module '%s' about type '%s' " "Read performed by module '%s' about type '%s' "
...@@ -3161,6 +3191,33 @@ void moduleRDBLoadError(RedisModuleIO *io) { ...@@ -3161,6 +3191,33 @@ void moduleRDBLoadError(RedisModuleIO *io) {
exit(1); exit(1);
} }
/* Returns 0 if there's at least one registered data type that did not declare
* REDISMODULE_OPTIONS_HANDLE_IO_ERRORS, in which case diskless loading should
* be avoided since it could cause data loss. */
int moduleAllDatatypesHandleErrors() {
dictIterator *di = dictGetIterator(modules);
dictEntry *de;
while ((de = dictNext(di)) != NULL) {
struct RedisModule *module = dictGetVal(de);
if (listLength(module->types) &&
!(module->options & REDISMODULE_OPTIONS_HANDLE_IO_ERRORS))
{
dictReleaseIterator(di);
return 0;
}
}
dictReleaseIterator(di);
return 1;
}
/* Returns true if any previous IO API failed.
* for Load* APIs the REDISMODULE_OPTIONS_HANDLE_IO_ERRORS flag must be set with
* RediModule_SetModuleOptions first. */
int RM_IsIOError(RedisModuleIO *io) {
return io->error;
}
/* Save an unsigned 64 bit value into the RDB file. This function should only /* Save an unsigned 64 bit value into the RDB file. This function should only
* be called in the context of the rdb_save method of modules implementing new * be called in the context of the rdb_save method of modules implementing new
* data types. */ * data types. */
...@@ -3184,6 +3241,7 @@ saveerr: ...@@ -3184,6 +3241,7 @@ saveerr:
* be called in the context of the rdb_load method of modules implementing * be called in the context of the rdb_load method of modules implementing
* new data types. */ * new data types. */
uint64_t RM_LoadUnsigned(RedisModuleIO *io) { uint64_t RM_LoadUnsigned(RedisModuleIO *io) {
if (io->error) return 0;
if (io->ver == 2) { if (io->ver == 2) {
uint64_t opcode = rdbLoadLen(io->rio,NULL); uint64_t opcode = rdbLoadLen(io->rio,NULL);
if (opcode != RDB_MODULE_OPCODE_UINT) goto loaderr; if (opcode != RDB_MODULE_OPCODE_UINT) goto loaderr;
...@@ -3195,7 +3253,7 @@ uint64_t RM_LoadUnsigned(RedisModuleIO *io) { ...@@ -3195,7 +3253,7 @@ uint64_t RM_LoadUnsigned(RedisModuleIO *io) {
loaderr: loaderr:
moduleRDBLoadError(io); moduleRDBLoadError(io);
return 0; /* Never reached. */ return 0;
} }
/* Like RedisModule_SaveUnsigned() but for signed 64 bit values. */ /* Like RedisModule_SaveUnsigned() but for signed 64 bit values. */
...@@ -3254,6 +3312,7 @@ saveerr: ...@@ -3254,6 +3312,7 @@ saveerr:
/* Implements RM_LoadString() and RM_LoadStringBuffer() */ /* Implements RM_LoadString() and RM_LoadStringBuffer() */
void *moduleLoadString(RedisModuleIO *io, int plain, size_t *lenptr) { void *moduleLoadString(RedisModuleIO *io, int plain, size_t *lenptr) {
if (io->error) return NULL;
if (io->ver == 2) { if (io->ver == 2) {
uint64_t opcode = rdbLoadLen(io->rio,NULL); uint64_t opcode = rdbLoadLen(io->rio,NULL);
if (opcode != RDB_MODULE_OPCODE_STRING) goto loaderr; if (opcode != RDB_MODULE_OPCODE_STRING) goto loaderr;
...@@ -3265,7 +3324,7 @@ void *moduleLoadString(RedisModuleIO *io, int plain, size_t *lenptr) { ...@@ -3265,7 +3324,7 @@ void *moduleLoadString(RedisModuleIO *io, int plain, size_t *lenptr) {
loaderr: loaderr:
moduleRDBLoadError(io); moduleRDBLoadError(io);
return NULL; /* Never reached. */ return NULL;
} }
/* In the context of the rdb_load method of a module data type, loads a string /* In the context of the rdb_load method of a module data type, loads a string
...@@ -3286,7 +3345,7 @@ RedisModuleString *RM_LoadString(RedisModuleIO *io) { ...@@ -3286,7 +3345,7 @@ RedisModuleString *RM_LoadString(RedisModuleIO *io) {
* RedisModule_Realloc() or RedisModule_Free(). * RedisModule_Realloc() or RedisModule_Free().
* *
* The size of the string is stored at '*lenptr' if not NULL. * The size of the string is stored at '*lenptr' if not NULL.
* The returned string is not automatically NULL termianted, it is loaded * The returned string is not automatically NULL terminated, it is loaded
* exactly as it was stored inisde the RDB file. */ * exactly as it was stored inisde the RDB file. */
char *RM_LoadStringBuffer(RedisModuleIO *io, size_t *lenptr) { char *RM_LoadStringBuffer(RedisModuleIO *io, size_t *lenptr) {
return moduleLoadString(io,1,lenptr); return moduleLoadString(io,1,lenptr);
...@@ -3314,6 +3373,7 @@ saveerr: ...@@ -3314,6 +3373,7 @@ saveerr:
/* In the context of the rdb_save method of a module data type, loads back the /* In the context of the rdb_save method of a module data type, loads back the
* double value saved by RedisModule_SaveDouble(). */ * double value saved by RedisModule_SaveDouble(). */
double RM_LoadDouble(RedisModuleIO *io) { double RM_LoadDouble(RedisModuleIO *io) {
if (io->error) return 0;
if (io->ver == 2) { if (io->ver == 2) {
uint64_t opcode = rdbLoadLen(io->rio,NULL); uint64_t opcode = rdbLoadLen(io->rio,NULL);
if (opcode != RDB_MODULE_OPCODE_DOUBLE) goto loaderr; if (opcode != RDB_MODULE_OPCODE_DOUBLE) goto loaderr;
...@@ -3325,7 +3385,7 @@ double RM_LoadDouble(RedisModuleIO *io) { ...@@ -3325,7 +3385,7 @@ double RM_LoadDouble(RedisModuleIO *io) {
loaderr: loaderr:
moduleRDBLoadError(io); moduleRDBLoadError(io);
return 0; /* Never reached. */ return 0;
} }
/* In the context of the rdb_save method of a module data type, saves a float /* In the context of the rdb_save method of a module data type, saves a float
...@@ -3350,6 +3410,7 @@ saveerr: ...@@ -3350,6 +3410,7 @@ saveerr:
/* In the context of the rdb_save method of a module data type, loads back the /* In the context of the rdb_save method of a module data type, loads back the
* float value saved by RedisModule_SaveFloat(). */ * float value saved by RedisModule_SaveFloat(). */
float RM_LoadFloat(RedisModuleIO *io) { float RM_LoadFloat(RedisModuleIO *io) {
if (io->error) return 0;
if (io->ver == 2) { if (io->ver == 2) {
uint64_t opcode = rdbLoadLen(io->rio,NULL); uint64_t opcode = rdbLoadLen(io->rio,NULL);
if (opcode != RDB_MODULE_OPCODE_FLOAT) goto loaderr; if (opcode != RDB_MODULE_OPCODE_FLOAT) goto loaderr;
...@@ -3361,7 +3422,37 @@ float RM_LoadFloat(RedisModuleIO *io) { ...@@ -3361,7 +3422,37 @@ float RM_LoadFloat(RedisModuleIO *io) {
loaderr: loaderr:
moduleRDBLoadError(io); moduleRDBLoadError(io);
return 0; /* Never reached. */ return 0;
}
/* Iterate over modules, and trigger rdb aux saving for the ones modules types
* who asked for it. */
ssize_t rdbSaveModulesAux(rio *rdb, int when) {
size_t total_written = 0;
dictIterator *di = dictGetIterator(modules);
dictEntry *de;
while ((de = dictNext(di)) != NULL) {
struct RedisModule *module = dictGetVal(de);
listIter li;
listNode *ln;
listRewind(module->types,&li);
while((ln = listNext(&li))) {
moduleType *mt = ln->value;
if (!mt->aux_save || !(mt->aux_save_triggers & when))
continue;
ssize_t ret = rdbSaveSingleModuleAux(rdb, when, mt);
if (ret==-1) {
dictReleaseIterator(di);
return -1;
}
total_written += ret;
}
}
dictReleaseIterator(di);
return total_written;
} }
/* -------------------------------------------------------------------------- /* --------------------------------------------------------------------------
...@@ -3524,7 +3615,7 @@ void RM_LogRaw(RedisModule *module, const char *levelstr, const char *fmt, va_li ...@@ -3524,7 +3615,7 @@ void RM_LogRaw(RedisModule *module, const char *levelstr, const char *fmt, va_li
if (level < server.verbosity) return; if (level < server.verbosity) return;
name_len = snprintf(msg, sizeof(msg),"<%s> ", module->name); name_len = snprintf(msg, sizeof(msg),"<%s> ", module? module->name: "module");
vsnprintf(msg + name_len, sizeof(msg) - name_len, fmt, ap); vsnprintf(msg + name_len, sizeof(msg) - name_len, fmt, ap);
serverLogRaw(level,msg); serverLogRaw(level,msg);
} }
...@@ -3542,13 +3633,15 @@ void RM_LogRaw(RedisModule *module, const char *levelstr, const char *fmt, va_li ...@@ -3542,13 +3633,15 @@ void RM_LogRaw(RedisModule *module, const char *levelstr, const char *fmt, va_li
* 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 limit 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.
*
* The ctx argument may be NULL if cannot be provided in the context of the
* caller for instance threads or callbacks, in which case a generic "module"
* will be used instead of the module name.
*/ */
void RM_Log(RedisModuleCtx *ctx, const char *levelstr, const char *fmt, ...) { void RM_Log(RedisModuleCtx *ctx, const char *levelstr, const char *fmt, ...) {
if (!ctx->module) return; /* Can only log if module is initialized */
va_list ap; va_list ap;
va_start(ap, fmt); va_start(ap, fmt);
RM_LogRaw(ctx->module,levelstr,fmt,ap); RM_LogRaw(ctx? ctx->module: NULL,levelstr,fmt,ap);
va_end(ap); va_end(ap);
} }
...@@ -3564,6 +3657,15 @@ void RM_LogIOError(RedisModuleIO *io, const char *levelstr, const char *fmt, ... ...@@ -3564,6 +3657,15 @@ void RM_LogIOError(RedisModuleIO *io, const char *levelstr, const char *fmt, ...
va_end(ap); va_end(ap);
} }
/* Redis-like assert function.
*
* A failed assertion will shut down the server and produce logging information
* that looks identical to information generated by Redis itself.
*/
void RM__Assert(const char *estr, const char *file, int line) {
_serverAssert(estr, file, line);
}
/* -------------------------------------------------------------------------- /* --------------------------------------------------------------------------
* Blocking clients from modules * Blocking clients from modules
* -------------------------------------------------------------------------- */ * -------------------------------------------------------------------------- */
...@@ -5362,6 +5464,62 @@ void addReplyLoadedModules(client *c) { ...@@ -5362,6 +5464,62 @@ void addReplyLoadedModules(client *c) {
dictReleaseIterator(di); dictReleaseIterator(di);
} }
/* Helper for genModulesInfoString(): given a list of modules, return
* am SDS string in the form "[modulename|modulename2|...]" */
sds genModulesInfoStringRenderModulesList(list *l) {
listIter li;
listNode *ln;
listRewind(l,&li);
sds output = sdsnew("[");
while((ln = listNext(&li))) {
RedisModule *module = ln->value;
output = sdscat(output,module->name);
}
output = sdstrim(output,"|");
output = sdscat(output,"]");
return output;
}
/* Helper for genModulesInfoString(): render module options as an SDS string. */
sds genModulesInfoStringRenderModuleOptions(struct RedisModule *module) {
sds output = sdsnew("[");
if (module->options & REDISMODULE_OPTIONS_HANDLE_IO_ERRORS)
output = sdscat(output,"handle-io-errors|");
output = sdstrim(output,"|");
output = sdscat(output,"]");
return output;
}
/* Helper function for the INFO command: adds loaded modules as to info's
* output.
*
* After the call, the passed sds info string is no longer valid and all the
* references must be substituted with the new pointer returned by the call. */
sds genModulesInfoString(sds info) {
dictIterator *di = dictGetIterator(modules);
dictEntry *de;
while ((de = dictNext(di)) != NULL) {
sds name = dictGetKey(de);
struct RedisModule *module = dictGetVal(de);
sds usedby = genModulesInfoStringRenderModulesList(module->usedby);
sds using = genModulesInfoStringRenderModulesList(module->using);
sds options = genModulesInfoStringRenderModuleOptions(module);
info = sdscatprintf(info,
"module:name=%s,ver=%d,api=%d,filters=%d,"
"usedby=%s,using=%s,options=%s\r\n",
name, module->ver, module->apiver,
(int)listLength(module->filters), usedby, using, options);
sdsfree(usedby);
sdsfree(using);
sdsfree(options);
}
dictReleaseIterator(di);
return info;
}
/* Redis MODULE command. /* Redis MODULE command.
* *
* MODULE LOAD <path> [args...] */ * MODULE LOAD <path> [args...] */
...@@ -5447,6 +5605,7 @@ void moduleRegisterCoreAPI(void) { ...@@ -5447,6 +5605,7 @@ void moduleRegisterCoreAPI(void) {
REGISTER_API(ReplySetArrayLength); REGISTER_API(ReplySetArrayLength);
REGISTER_API(ReplyWithString); REGISTER_API(ReplyWithString);
REGISTER_API(ReplyWithStringBuffer); REGISTER_API(ReplyWithStringBuffer);
REGISTER_API(ReplyWithCString);
REGISTER_API(ReplyWithNull); REGISTER_API(ReplyWithNull);
REGISTER_API(ReplyWithCallReply); REGISTER_API(ReplyWithCallReply);
REGISTER_API(ReplyWithDouble); REGISTER_API(ReplyWithDouble);
...@@ -5509,6 +5668,8 @@ void moduleRegisterCoreAPI(void) { ...@@ -5509,6 +5668,8 @@ void moduleRegisterCoreAPI(void) {
REGISTER_API(ModuleTypeSetValue); REGISTER_API(ModuleTypeSetValue);
REGISTER_API(ModuleTypeGetType); REGISTER_API(ModuleTypeGetType);
REGISTER_API(ModuleTypeGetValue); REGISTER_API(ModuleTypeGetValue);
REGISTER_API(IsIOError);
REGISTER_API(SetModuleOptions);
REGISTER_API(SaveUnsigned); REGISTER_API(SaveUnsigned);
REGISTER_API(LoadUnsigned); REGISTER_API(LoadUnsigned);
REGISTER_API(SaveSigned); REGISTER_API(SaveSigned);
...@@ -5524,6 +5685,7 @@ void moduleRegisterCoreAPI(void) { ...@@ -5524,6 +5685,7 @@ void moduleRegisterCoreAPI(void) {
REGISTER_API(EmitAOF); REGISTER_API(EmitAOF);
REGISTER_API(Log); REGISTER_API(Log);
REGISTER_API(LogIOError); REGISTER_API(LogIOError);
REGISTER_API(_Assert);
REGISTER_API(StringAppendBuffer); REGISTER_API(StringAppendBuffer);
REGISTER_API(RetainString); REGISTER_API(RetainString);
REGISTER_API(StringCompare); REGISTER_API(StringCompare);
......
...@@ -175,7 +175,19 @@ void execCommand(client *c) { ...@@ -175,7 +175,19 @@ void execCommand(client *c) {
must_propagate = 1; must_propagate = 1;
} }
call(c,server.loading ? CMD_CALL_NONE : CMD_CALL_FULL); int acl_retval = ACLCheckCommandPerm(c);
if (acl_retval != ACL_OK) {
addReplyErrorFormat(c,
"-NOPERM ACLs rules changed between the moment the "
"transaction was accumulated and the EXEC call. "
"This command is no longer allowed for the "
"following reason: %s",
(acl_retval == ACL_DENIED_CMD) ?
"no permission to execute the command or subcommand" :
"no permission to touch the specified keys");
} else {
call(c,server.loading ? CMD_CALL_NONE : CMD_CALL_FULL);
}
/* Commands may alter argc/argv, restore mstate. */ /* Commands may alter argc/argv, restore mstate. */
c->mstate.commands[j].argc = c->argc; c->mstate.commands[j].argc = c->argc;
......
...@@ -1990,7 +1990,7 @@ NULL ...@@ -1990,7 +1990,7 @@ NULL
return; return;
} }
sds o = getAllClientsInfoString(type); sds o = getAllClientsInfoString(type);
addReplyBulkCBuffer(c,o,sdslen(o)); addReplyVerbatim(c,o,sdslen(o),"txt");
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) {
/* CLIENT REPLY ON|OFF|SKIP */ /* CLIENT REPLY ON|OFF|SKIP */
...@@ -2468,17 +2468,27 @@ void flushSlavesOutputBuffers(void) { ...@@ -2468,17 +2468,27 @@ void flushSlavesOutputBuffers(void) {
listRewind(server.slaves,&li); listRewind(server.slaves,&li);
while((ln = listNext(&li))) { while((ln = listNext(&li))) {
client *slave = listNodeValue(ln); client *slave = listNodeValue(ln);
int events; int events = aeGetFileEvents(server.el,slave->fd);
int can_receive_writes = (events & AE_WRITABLE) ||
/* Note that the following will not flush output buffers of slaves (slave->flags & CLIENT_PENDING_WRITE);
* in STATE_ONLINE but having put_online_on_ack set to true: in this
* case the writable event is never installed, since the purpose /* We don't want to send the pending data to the replica in a few
* of put_online_on_ack is to postpone the moment it is installed. * cases:
* This is what we want since slaves in this state should not receive *
* writes before the first ACK. */ * 1. For some reason there is neither the write handler installed
events = aeGetFileEvents(server.el,slave->fd); * nor the client is flagged as to have pending writes: for some
if (events & AE_WRITABLE && * reason this replica may not be set to receive data. This is
slave->replstate == SLAVE_STATE_ONLINE && * just for the sake of defensive programming.
*
* 2. The put_online_on_ack flag is true. To know why we don't want
* to send data to the replica in this case, please grep for the
* flag for this flag.
*
* 3. Obviously if the slave is not ONLINE.
*/
if (slave->replstate == SLAVE_STATE_ONLINE &&
can_receive_writes &&
!slave->repl_put_online_on_ack &&
clientHasPendingReplies(slave)) clientHasPendingReplies(slave))
{ {
writeToClient(slave->fd,slave,0); writeToClient(slave->fd,slave,0);
......
...@@ -467,10 +467,15 @@ robj *tryObjectEncoding(robj *o) { ...@@ -467,10 +467,15 @@ robj *tryObjectEncoding(robj *o) {
incrRefCount(shared.integers[value]); incrRefCount(shared.integers[value]);
return shared.integers[value]; return shared.integers[value];
} else { } else {
if (o->encoding == OBJ_ENCODING_RAW) sdsfree(o->ptr); if (o->encoding == OBJ_ENCODING_RAW) {
o->encoding = OBJ_ENCODING_INT; sdsfree(o->ptr);
o->ptr = (void*) value; o->encoding = OBJ_ENCODING_INT;
return o; o->ptr = (void*) value;
return o;
} else if (o->encoding == OBJ_ENCODING_EMBSTR) {
decrRefCount(o);
return createStringObjectFromLongLongForValue(value);
}
} }
} }
...@@ -1435,13 +1440,15 @@ NULL ...@@ -1435,13 +1440,15 @@ NULL
#if defined(USE_JEMALLOC) #if defined(USE_JEMALLOC)
sds info = sdsempty(); sds info = sdsempty();
je_malloc_stats_print(inputCatSds, &info, NULL); je_malloc_stats_print(inputCatSds, &info, NULL);
addReplyBulkSds(c, info); addReplyVerbatim(c,info,sdslen(info),"txt");
sdsfree(info);
#else #else
addReplyBulkCString(c,"Stats not supported for the current allocator"); addReplyBulkCString(c,"Stats not supported for the current allocator");
#endif #endif
} else if (!strcasecmp(c->argv[1]->ptr,"doctor") && c->argc == 2) { } else if (!strcasecmp(c->argv[1]->ptr,"doctor") && c->argc == 2) {
sds report = getMemoryDoctorReport(); sds report = getMemoryDoctorReport();
addReplyBulkSds(c,report); addReplyVerbatim(c,report,sdslen(report),"txt");
sdsfree(report);
} else if (!strcasecmp(c->argv[1]->ptr,"purge") && c->argc == 2) { } else if (!strcasecmp(c->argv[1]->ptr,"purge") && c->argc == 2) {
#if defined(USE_JEMALLOC) #if defined(USE_JEMALLOC)
char tmp[32]; char tmp[32];
......
...@@ -42,31 +42,35 @@ ...@@ -42,31 +42,35 @@
#include <sys/stat.h> #include <sys/stat.h>
#include <sys/param.h> #include <sys/param.h>
#define rdbExitReportCorruptRDB(...) rdbCheckThenExit(__LINE__,__VA_ARGS__) /* This macro is called when the internal RDB stracture is corrupt */
#define rdbExitReportCorruptRDB(...) rdbReportError(1, __LINE__,__VA_ARGS__)
/* This macro is called when RDB read failed (possibly a short read) */
#define rdbReportReadError(...) rdbReportError(0, __LINE__,__VA_ARGS__)
char* rdbFileBeingLoaded = NULL; /* used for rdb checking on read error */ char* rdbFileBeingLoaded = NULL; /* used for rdb checking on read error */
extern int rdbCheckMode; extern int rdbCheckMode;
void rdbCheckError(const char *fmt, ...); void rdbCheckError(const char *fmt, ...);
void rdbCheckSetError(const char *fmt, ...); void rdbCheckSetError(const char *fmt, ...);
void rdbCheckThenExit(int linenum, char *reason, ...) { void rdbReportError(int corruption_error, int linenum, char *reason, ...) {
va_list ap; va_list ap;
char msg[1024]; char msg[1024];
int len; int len;
len = snprintf(msg,sizeof(msg), len = snprintf(msg,sizeof(msg),
"Internal error in RDB reading function at rdb.c:%d -> ", linenum); "Internal error in RDB reading offset %llu, function at rdb.c:%d -> ",
(unsigned long long)server.loading_loaded_bytes, linenum);
va_start(ap,reason); va_start(ap,reason);
vsnprintf(msg+len,sizeof(msg)-len,reason,ap); vsnprintf(msg+len,sizeof(msg)-len,reason,ap);
va_end(ap); va_end(ap);
if (!rdbCheckMode) { if (!rdbCheckMode) {
serverLog(LL_WARNING, "%s", msg); if (rdbFileBeingLoaded || corruption_error) {
if (rdbFileBeingLoaded) { serverLog(LL_WARNING, "%s", msg);
char *argv[2] = {"",rdbFileBeingLoaded}; char *argv[2] = {"",rdbFileBeingLoaded};
redis_check_rdb_main(2,argv,NULL); redis_check_rdb_main(2,argv,NULL);
} else { } else {
serverLog(LL_WARNING, "Failure loading rdb format from socket, assuming connection error, resuming operation."); serverLog(LL_WARNING, "%s. Failure loading rdb format from socket, assuming connection error, resuming operation.", msg);
return; return;
} }
} else { } else {
...@@ -82,18 +86,6 @@ static int rdbWriteRaw(rio *rdb, void *p, size_t len) { ...@@ -82,18 +86,6 @@ static int rdbWriteRaw(rio *rdb, void *p, size_t len) {
return len; return len;
} }
/* This is just a wrapper for the low level function rioRead() that will
* automatically abort if it is not possible to read the specified amount
* of bytes. */
void rdbLoadRaw(rio *rdb, void *buf, uint64_t len) {
if (rioRead(rdb,buf,len) == 0) {
rdbExitReportCorruptRDB(
"Impossible to read %llu bytes in rdbLoadRaw()",
(unsigned long long) len);
return; /* Not reached. */
}
}
int rdbSaveType(rio *rdb, unsigned char type) { int rdbSaveType(rio *rdb, unsigned char type) {
return rdbWriteRaw(rdb,&type,1); return rdbWriteRaw(rdb,&type,1);
} }
...@@ -109,10 +101,12 @@ int rdbLoadType(rio *rdb) { ...@@ -109,10 +101,12 @@ int rdbLoadType(rio *rdb) {
/* This is only used to load old databases stored with the RDB_OPCODE_EXPIRETIME /* This is only used to load old databases stored with the RDB_OPCODE_EXPIRETIME
* opcode. New versions of Redis store using the RDB_OPCODE_EXPIRETIME_MS * opcode. New versions of Redis store using the RDB_OPCODE_EXPIRETIME_MS
* opcode. */ * opcode. On error -1 is returned, however this could be a valid time, so
* to check for loading errors the caller should call rioGetReadError() after
* calling this function. */
time_t rdbLoadTime(rio *rdb) { time_t rdbLoadTime(rio *rdb) {
int32_t t32; int32_t t32;
rdbLoadRaw(rdb,&t32,4); if (rioRead(rdb,&t32,4) == 0) return -1;
return (time_t)t32; return (time_t)t32;
} }
...@@ -132,10 +126,14 @@ int rdbSaveMillisecondTime(rio *rdb, long long t) { ...@@ -132,10 +126,14 @@ int rdbSaveMillisecondTime(rio *rdb, long long t) {
* after upgrading to Redis version 5 they will no longer be able to load their * after upgrading to Redis version 5 they will no longer be able to load their
* own old RDB files. Because of that, we instead fix the function only for new * own old RDB files. Because of that, we instead fix the function only for new
* RDB versions, and load older RDB versions as we used to do in the past, * RDB versions, and load older RDB versions as we used to do in the past,
* allowing big endian systems to load their own old RDB files. */ * allowing big endian systems to load their own old RDB files.
*
* On I/O error the function returns LLONG_MAX, however if this is also a
* valid stored value, the caller should use rioGetReadError() to check for
* errors after calling this function. */
long long rdbLoadMillisecondTime(rio *rdb, int rdbver) { long long rdbLoadMillisecondTime(rio *rdb, int rdbver) {
int64_t t64; int64_t t64;
rdbLoadRaw(rdb,&t64,8); if (rioRead(rdb,&t64,8) == 0) return LLONG_MAX;
if (rdbver >= 9) /* Check the top comment of this function. */ if (rdbver >= 9) /* Check the top comment of this function. */
memrev64ifbe(&t64); /* Convert in big endian if the system is BE. */ memrev64ifbe(&t64); /* Convert in big endian if the system is BE. */
return (long long)t64; return (long long)t64;
...@@ -262,7 +260,7 @@ int rdbEncodeInteger(long long value, unsigned char *enc) { ...@@ -262,7 +260,7 @@ int rdbEncodeInteger(long long value, unsigned char *enc) {
/* Loads an integer-encoded object with the specified encoding type "enctype". /* Loads an integer-encoded object with the specified encoding type "enctype".
* The returned value changes according to the flags, see * The returned value changes according to the flags, see
* rdbGenerincLoadStringObject() for more info. */ * rdbGenericLoadStringObject() for more info. */
void *rdbLoadIntegerObject(rio *rdb, int enctype, int flags, size_t *lenptr) { void *rdbLoadIntegerObject(rio *rdb, int enctype, int flags, size_t *lenptr) {
int plain = flags & RDB_LOAD_PLAIN; int plain = flags & RDB_LOAD_PLAIN;
int sds = flags & RDB_LOAD_SDS; int sds = flags & RDB_LOAD_SDS;
...@@ -284,8 +282,8 @@ void *rdbLoadIntegerObject(rio *rdb, int enctype, int flags, size_t *lenptr) { ...@@ -284,8 +282,8 @@ void *rdbLoadIntegerObject(rio *rdb, int enctype, int flags, size_t *lenptr) {
v = enc[0]|(enc[1]<<8)|(enc[2]<<16)|(enc[3]<<24); v = enc[0]|(enc[1]<<8)|(enc[2]<<16)|(enc[3]<<24);
val = (int32_t)v; val = (int32_t)v;
} else { } else {
val = 0; /* anti-warning */
rdbExitReportCorruptRDB("Unknown RDB integer encoding type %d",enctype); rdbExitReportCorruptRDB("Unknown RDB integer encoding type %d",enctype);
return NULL; /* Never reached. */
} }
if (plain || sds) { if (plain || sds) {
char buf[LONG_STR_SIZE], *p; char buf[LONG_STR_SIZE], *p;
...@@ -388,8 +386,7 @@ void *rdbLoadLzfStringObject(rio *rdb, int flags, size_t *lenptr) { ...@@ -388,8 +386,7 @@ void *rdbLoadLzfStringObject(rio *rdb, int flags, size_t *lenptr) {
/* Load the compressed representation and uncompress it to target. */ /* Load the compressed representation and uncompress it to target. */
if (rioRead(rdb,c,clen) == 0) goto err; if (rioRead(rdb,c,clen) == 0) goto err;
if (lzf_decompress(c,clen,val,len) == 0) { if (lzf_decompress(c,clen,val,len) == 0) {
if (rdbCheckMode) rdbCheckSetError("Invalid LZF compressed string"); rdbExitReportCorruptRDB("Invalid LZF compressed string");
goto err;
} }
zfree(c); zfree(c);
...@@ -503,6 +500,7 @@ void *rdbGenericLoadStringObject(rio *rdb, int flags, size_t *lenptr) { ...@@ -503,6 +500,7 @@ void *rdbGenericLoadStringObject(rio *rdb, int flags, size_t *lenptr) {
return rdbLoadLzfStringObject(rdb,flags,lenptr); return rdbLoadLzfStringObject(rdb,flags,lenptr);
default: default:
rdbExitReportCorruptRDB("Unknown RDB string encoding type %d",len); rdbExitReportCorruptRDB("Unknown RDB string encoding type %d",len);
return NULL; /* Never reached. */
} }
} }
...@@ -973,7 +971,6 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key) { ...@@ -973,7 +971,6 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key) {
RedisModuleIO io; RedisModuleIO io;
moduleValue *mv = o->ptr; moduleValue *mv = o->ptr;
moduleType *mt = mv->type; moduleType *mt = mv->type;
moduleInitIOContext(io,mt,rdb,key);
/* Write the "module" identifier as prefix, so that we'll be able /* Write the "module" identifier as prefix, so that we'll be able
* to call the right module during loading. */ * to call the right module during loading. */
...@@ -982,10 +979,13 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key) { ...@@ -982,10 +979,13 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key) {
io.bytes += retval; io.bytes += retval;
/* Then write the module-specific representation + EOF marker. */ /* Then write the module-specific representation + EOF marker. */
moduleInitIOContext(io,mt,rdb,key);
mt->rdb_save(&io,mv->value); mt->rdb_save(&io,mv->value);
retval = rdbSaveLen(rdb,RDB_MODULE_OPCODE_EOF); retval = rdbSaveLen(rdb,RDB_MODULE_OPCODE_EOF);
if (retval == -1) return -1; if (retval == -1)
io.bytes += retval; io.error = 1;
else
io.bytes += retval;
if (io.ctx) { if (io.ctx) {
moduleFreeContext(io.ctx); moduleFreeContext(io.ctx);
...@@ -1103,6 +1103,45 @@ int rdbSaveInfoAuxFields(rio *rdb, int flags, rdbSaveInfo *rsi) { ...@@ -1103,6 +1103,45 @@ int rdbSaveInfoAuxFields(rio *rdb, int flags, rdbSaveInfo *rsi) {
return 1; return 1;
} }
ssize_t rdbSaveSingleModuleAux(rio *rdb, int when, moduleType *mt) {
/* Save a module-specific aux value. */
RedisModuleIO io;
int retval = rdbSaveType(rdb, RDB_OPCODE_MODULE_AUX);
/* Write the "module" identifier as prefix, so that we'll be able
* to call the right module during loading. */
retval = rdbSaveLen(rdb,mt->id);
if (retval == -1) return -1;
io.bytes += retval;
/* write the 'when' so that we can provide it on loading. add a UINT opcode
* for backwards compatibility, everything after the MT needs to be prefixed
* by an opcode. */
retval = rdbSaveLen(rdb,RDB_MODULE_OPCODE_UINT);
if (retval == -1) return -1;
io.bytes += retval;
retval = rdbSaveLen(rdb,when);
if (retval == -1) return -1;
io.bytes += retval;
/* Then write the module-specific representation + EOF marker. */
moduleInitIOContext(io,mt,rdb,NULL);
mt->aux_save(&io,when);
retval = rdbSaveLen(rdb,RDB_MODULE_OPCODE_EOF);
if (retval == -1)
io.error = 1;
else
io.bytes += retval;
if (io.ctx) {
moduleFreeContext(io.ctx);
zfree(io.ctx);
}
if (io.error)
return -1;
return io.bytes;
}
/* Produces a dump of the database in RDB format sending it to the specified /* Produces a dump of the database in RDB format sending it to the specified
* Redis I/O channel. On success C_OK is returned, otherwise C_ERR * Redis I/O channel. On success C_OK is returned, otherwise C_ERR
* is returned and part of the output, or all the output, can be * is returned and part of the output, or all the output, can be
...@@ -1124,6 +1163,7 @@ int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi) { ...@@ -1124,6 +1163,7 @@ int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi) {
snprintf(magic,sizeof(magic),"REDIS%04d",RDB_VERSION); snprintf(magic,sizeof(magic),"REDIS%04d",RDB_VERSION);
if (rdbWriteRaw(rdb,magic,9) == -1) goto werr; if (rdbWriteRaw(rdb,magic,9) == -1) goto werr;
if (rdbSaveInfoAuxFields(rdb,flags,rsi) == -1) goto werr; if (rdbSaveInfoAuxFields(rdb,flags,rsi) == -1) goto werr;
if (rdbSaveModulesAux(rdb, REDISMODULE_AUX_BEFORE_RDB) == -1) goto werr;
for (j = 0; j < server.dbnum; j++) { for (j = 0; j < server.dbnum; j++) {
redisDb *db = server.db+j; redisDb *db = server.db+j;
...@@ -1185,6 +1225,8 @@ int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi) { ...@@ -1185,6 +1225,8 @@ int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi) {
di = NULL; /* So that we don't release it again on error. */ di = NULL; /* So that we don't release it again on error. */
} }
if (rdbSaveModulesAux(rdb, REDISMODULE_AUX_AFTER_RDB) == -1) goto werr;
/* EOF opcode */ /* EOF opcode */
if (rdbSaveType(rdb,RDB_OPCODE_EOF) == -1) goto werr; if (rdbSaveType(rdb,RDB_OPCODE_EOF) == -1) goto werr;
...@@ -1628,6 +1670,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) { ...@@ -1628,6 +1670,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) {
hashTypeConvert(o, OBJ_ENCODING_HT); hashTypeConvert(o, OBJ_ENCODING_HT);
break; break;
default: default:
/* totally unreachable */
rdbExitReportCorruptRDB("Unknown RDB encoding type %d",rdbtype); rdbExitReportCorruptRDB("Unknown RDB encoding type %d",rdbtype);
break; break;
} }
...@@ -1635,6 +1678,11 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) { ...@@ -1635,6 +1678,11 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) {
o = createStreamObject(); o = createStreamObject();
stream *s = o->ptr; stream *s = o->ptr;
uint64_t listpacks = rdbLoadLen(rdb,NULL); uint64_t listpacks = rdbLoadLen(rdb,NULL);
if (listpacks == RDB_LENERR) {
rdbReportReadError("Stream listpacks len loading failed.");
decrRefCount(o);
return NULL;
}
while(listpacks--) { while(listpacks--) {
/* Get the master ID, the one we'll use as key of the radix tree /* Get the master ID, the one we'll use as key of the radix tree
...@@ -1642,7 +1690,9 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) { ...@@ -1642,7 +1690,9 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) {
* relatively to this ID. */ * relatively to this ID. */
sds nodekey = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS,NULL); sds nodekey = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS,NULL);
if (nodekey == NULL) { if (nodekey == NULL) {
rdbExitReportCorruptRDB("Stream master ID loading failed: invalid encoding or I/O error."); rdbReportReadError("Stream master ID loading failed: invalid encoding or I/O error.");
decrRefCount(o);
return NULL;
} }
if (sdslen(nodekey) != sizeof(streamID)) { if (sdslen(nodekey) != sizeof(streamID)) {
rdbExitReportCorruptRDB("Stream node key entry is not the " rdbExitReportCorruptRDB("Stream node key entry is not the "
...@@ -1652,7 +1702,12 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) { ...@@ -1652,7 +1702,12 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) {
/* Load the listpack. */ /* Load the listpack. */
unsigned char *lp = unsigned char *lp =
rdbGenericLoadStringObject(rdb,RDB_LOAD_PLAIN,NULL); rdbGenericLoadStringObject(rdb,RDB_LOAD_PLAIN,NULL);
if (lp == NULL) return NULL; if (lp == NULL) {
rdbReportReadError("Stream listpacks loading failed.");
sdsfree(nodekey);
decrRefCount(o);
return NULL;
}
unsigned char *first = lpFirst(lp); unsigned char *first = lpFirst(lp);
if (first == NULL) { if (first == NULL) {
/* Serialized listpacks should never be empty, since on /* Serialized listpacks should never be empty, since on
...@@ -1670,12 +1725,24 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) { ...@@ -1670,12 +1725,24 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) {
} }
/* Load total number of items inside the stream. */ /* Load total number of items inside the stream. */
s->length = rdbLoadLen(rdb,NULL); s->length = rdbLoadLen(rdb,NULL);
/* Load the last entry ID. */ /* Load the last entry ID. */
s->last_id.ms = rdbLoadLen(rdb,NULL); s->last_id.ms = rdbLoadLen(rdb,NULL);
s->last_id.seq = rdbLoadLen(rdb,NULL); s->last_id.seq = rdbLoadLen(rdb,NULL);
if (rioGetReadError(rdb)) {
rdbReportReadError("Stream object metadata loading failed.");
decrRefCount(o);
return NULL;
}
/* Consumer groups loading */ /* Consumer groups loading */
size_t cgroups_count = rdbLoadLen(rdb,NULL); uint64_t cgroups_count = rdbLoadLen(rdb,NULL);
if (cgroups_count == RDB_LENERR) {
rdbReportReadError("Stream cgroup count loading failed.");
decrRefCount(o);
return NULL;
}
while(cgroups_count--) { while(cgroups_count--) {
/* Get the consumer group name and ID. We can then create the /* Get the consumer group name and ID. We can then create the
* consumer group ASAP and populate its structure as * consumer group ASAP and populate its structure as
...@@ -1683,11 +1750,21 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) { ...@@ -1683,11 +1750,21 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) {
streamID cg_id; streamID cg_id;
sds cgname = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS,NULL); sds cgname = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS,NULL);
if (cgname == NULL) { if (cgname == NULL) {
rdbExitReportCorruptRDB( rdbReportReadError(
"Error reading the consumer group name from Stream"); "Error reading the consumer group name from Stream");
decrRefCount(o);
return NULL;
} }
cg_id.ms = rdbLoadLen(rdb,NULL); cg_id.ms = rdbLoadLen(rdb,NULL);
cg_id.seq = rdbLoadLen(rdb,NULL); cg_id.seq = rdbLoadLen(rdb,NULL);
if (rioGetReadError(rdb)) {
rdbReportReadError("Stream cgroup ID loading failed.");
sdsfree(cgname);
decrRefCount(o);
return NULL;
}
streamCG *cgroup = streamCreateCG(s,cgname,sdslen(cgname),&cg_id); streamCG *cgroup = streamCreateCG(s,cgname,sdslen(cgname),&cg_id);
if (cgroup == NULL) if (cgroup == NULL)
rdbExitReportCorruptRDB("Duplicated consumer group name %s", rdbExitReportCorruptRDB("Duplicated consumer group name %s",
...@@ -1699,13 +1776,28 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) { ...@@ -1699,13 +1776,28 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) {
* owner, since consumers for this group and their messages will * owner, since consumers for this group and their messages will
* be read as a next step. So for now leave them not resolved * be read as a next step. So for now leave them not resolved
* and later populate it. */ * and later populate it. */
size_t pel_size = rdbLoadLen(rdb,NULL); uint64_t pel_size = rdbLoadLen(rdb,NULL);
if (pel_size == RDB_LENERR) {
rdbReportReadError("Stream PEL size loading failed.");
decrRefCount(o);
return NULL;
}
while(pel_size--) { while(pel_size--) {
unsigned char rawid[sizeof(streamID)]; unsigned char rawid[sizeof(streamID)];
rdbLoadRaw(rdb,rawid,sizeof(rawid)); if (rioRead(rdb,rawid,sizeof(rawid)) == 0) {
rdbReportReadError("Stream PEL ID loading failed.");
decrRefCount(o);
return NULL;
}
streamNACK *nack = streamCreateNACK(NULL); streamNACK *nack = streamCreateNACK(NULL);
nack->delivery_time = rdbLoadMillisecondTime(rdb,RDB_VERSION); nack->delivery_time = rdbLoadMillisecondTime(rdb,RDB_VERSION);
nack->delivery_count = rdbLoadLen(rdb,NULL); nack->delivery_count = rdbLoadLen(rdb,NULL);
if (rioGetReadError(rdb)) {
rdbReportReadError("Stream PEL NACK loading failed.");
decrRefCount(o);
streamFreeNACK(nack);
return NULL;
}
if (!raxInsert(cgroup->pel,rawid,sizeof(rawid),nack,NULL)) if (!raxInsert(cgroup->pel,rawid,sizeof(rawid),nack,NULL))
rdbExitReportCorruptRDB("Duplicated gobal PEL entry " rdbExitReportCorruptRDB("Duplicated gobal PEL entry "
"loading stream consumer group"); "loading stream consumer group");
...@@ -1713,24 +1805,47 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) { ...@@ -1713,24 +1805,47 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) {
/* Now that we loaded our global PEL, we need to load the /* Now that we loaded our global PEL, we need to load the
* consumers and their local PELs. */ * consumers and their local PELs. */
size_t consumers_num = rdbLoadLen(rdb,NULL); uint64_t consumers_num = rdbLoadLen(rdb,NULL);
if (consumers_num == RDB_LENERR) {
rdbReportReadError("Stream consumers num loading failed.");
decrRefCount(o);
return NULL;
}
while(consumers_num--) { while(consumers_num--) {
sds cname = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS,NULL); sds cname = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS,NULL);
if (cname == NULL) { if (cname == NULL) {
rdbExitReportCorruptRDB( rdbReportReadError(
"Error reading the consumer name from Stream group"); "Error reading the consumer name from Stream group.");
decrRefCount(o);
return NULL;
} }
streamConsumer *consumer = streamLookupConsumer(cgroup,cname, streamConsumer *consumer = streamLookupConsumer(cgroup,cname,
1); 1);
sdsfree(cname); sdsfree(cname);
consumer->seen_time = rdbLoadMillisecondTime(rdb,RDB_VERSION); consumer->seen_time = rdbLoadMillisecondTime(rdb,RDB_VERSION);
if (rioGetReadError(rdb)) {
rdbReportReadError("Stream short read reading seen time.");
decrRefCount(o);
return NULL;
}
/* Load the PEL about entries owned by this specific /* Load the PEL about entries owned by this specific
* consumer. */ * consumer. */
pel_size = rdbLoadLen(rdb,NULL); pel_size = rdbLoadLen(rdb,NULL);
if (pel_size == RDB_LENERR) {
rdbReportReadError(
"Stream consumer PEL num loading failed.");
decrRefCount(o);
return NULL;
}
while(pel_size--) { while(pel_size--) {
unsigned char rawid[sizeof(streamID)]; unsigned char rawid[sizeof(streamID)];
rdbLoadRaw(rdb,rawid,sizeof(rawid)); if (rioRead(rdb,rawid,sizeof(rawid)) == 0) {
rdbReportReadError(
"Stream short read reading PEL streamID.");
decrRefCount(o);
return NULL;
}
streamNACK *nack = raxFind(cgroup->pel,rawid,sizeof(rawid)); streamNACK *nack = raxFind(cgroup->pel,rawid,sizeof(rawid));
if (nack == raxNotFound) if (nack == raxNotFound)
rdbExitReportCorruptRDB("Consumer entry not found in " rdbExitReportCorruptRDB("Consumer entry not found in "
...@@ -1749,6 +1864,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) { ...@@ -1749,6 +1864,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) {
} }
} else if (rdbtype == RDB_TYPE_MODULE || rdbtype == RDB_TYPE_MODULE_2) { } else if (rdbtype == RDB_TYPE_MODULE || rdbtype == RDB_TYPE_MODULE_2) {
uint64_t moduleid = rdbLoadLen(rdb,NULL); uint64_t moduleid = rdbLoadLen(rdb,NULL);
if (rioGetReadError(rdb)) return NULL;
moduleType *mt = moduleTypeLookupModuleByID(moduleid); moduleType *mt = moduleTypeLookupModuleByID(moduleid);
char name[10]; char name[10];
...@@ -1776,6 +1892,11 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) { ...@@ -1776,6 +1892,11 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) {
/* Module v2 serialization has an EOF mark at the end. */ /* Module v2 serialization has an EOF mark at the end. */
if (io.ver == 2) { if (io.ver == 2) {
uint64_t eof = rdbLoadLen(rdb,NULL); uint64_t eof = rdbLoadLen(rdb,NULL);
if (eof == RDB_LENERR) {
o = createModuleObject(mt,ptr); /* creating just in order to easily destroy */
decrRefCount(o);
return NULL;
}
if (eof != RDB_MODULE_OPCODE_EOF) { if (eof != RDB_MODULE_OPCODE_EOF) {
serverLog(LL_WARNING,"The RDB file contains module data for the module '%s' that is not terminated by the proper module value EOF marker", name); serverLog(LL_WARNING,"The RDB file contains module data for the module '%s' that is not terminated by the proper module value EOF marker", name);
exit(1); exit(1);
...@@ -1789,7 +1910,8 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) { ...@@ -1789,7 +1910,8 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, robj *key) {
} }
o = createModuleObject(mt,ptr); o = createModuleObject(mt,ptr);
} else { } else {
rdbExitReportCorruptRDB("Unknown RDB encoding type %d",rdbtype); rdbReportReadError("Unknown RDB encoding type %d",rdbtype);
return NULL;
} }
return o; return o;
} }
...@@ -1888,11 +2010,13 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) { ...@@ -1888,11 +2010,13 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) {
* load the actual type, and continue. */ * load the actual type, and continue. */
expiretime = rdbLoadTime(rdb); expiretime = rdbLoadTime(rdb);
expiretime *= 1000; expiretime *= 1000;
if (rioGetReadError(rdb)) goto eoferr;
continue; /* Read next opcode. */ continue; /* Read next opcode. */
} else if (type == RDB_OPCODE_EXPIRETIME_MS) { } else if (type == RDB_OPCODE_EXPIRETIME_MS) {
/* EXPIRETIME_MS: milliseconds precision expire times introduced /* EXPIRETIME_MS: milliseconds precision expire times introduced
* with RDB v3. Like EXPIRETIME but no with more precision. */ * with RDB v3. Like EXPIRETIME but no with more precision. */
expiretime = rdbLoadMillisecondTime(rdb,rdbver); expiretime = rdbLoadMillisecondTime(rdb,rdbver);
if (rioGetReadError(rdb)) goto eoferr;
continue; /* Read next opcode. */ continue; /* Read next opcode. */
} else if (type == RDB_OPCODE_FREQ) { } else if (type == RDB_OPCODE_FREQ) {
/* FREQ: LFU frequency. */ /* FREQ: LFU frequency. */
...@@ -1993,15 +2117,15 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) { ...@@ -1993,15 +2117,15 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) {
decrRefCount(auxval); decrRefCount(auxval);
continue; /* Read type again. */ continue; /* Read type again. */
} else if (type == RDB_OPCODE_MODULE_AUX) { } else if (type == RDB_OPCODE_MODULE_AUX) {
/* This is just for compatibility with the future: we have plans /* Load module data that is not related to the Redis key space.
* to add the ability for modules to store anything in the RDB * Such data can be potentially be stored both before and after the
* file, like data that is not related to the Redis key space. * RDB keys-values section. */
* Such data will potentially be stored both before and after the
* RDB keys-values section. For this reason since RDB version 9,
* we have the ability to read a MODULE_AUX opcode followed by an
* identifier of the module, and a serialized value in "MODULE V2"
* format. */
uint64_t moduleid = rdbLoadLen(rdb,NULL); uint64_t moduleid = rdbLoadLen(rdb,NULL);
int when_opcode = rdbLoadLen(rdb,NULL);
int when = rdbLoadLen(rdb,NULL);
if (rioGetReadError(rdb)) goto eoferr;
if (when_opcode != RDB_MODULE_OPCODE_UINT)
rdbReportReadError("bad when_opcode");
moduleType *mt = moduleTypeLookupModuleByID(moduleid); moduleType *mt = moduleTypeLookupModuleByID(moduleid);
char name[10]; char name[10];
moduleTypeNameByID(name,moduleid); moduleTypeNameByID(name,moduleid);
...@@ -2011,14 +2135,37 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) { ...@@ -2011,14 +2135,37 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) {
serverLog(LL_WARNING,"The RDB file contains AUX module data I can't load: no matching module '%s'", name); serverLog(LL_WARNING,"The RDB file contains AUX module data I can't load: no matching module '%s'", name);
exit(1); exit(1);
} else if (!rdbCheckMode && mt != NULL) { } else if (!rdbCheckMode && mt != NULL) {
/* This version of Redis actually does not know what to do if (!mt->aux_load) {
* with modules AUX data... */ /* Module doesn't support AUX. */
serverLog(LL_WARNING,"The RDB file contains AUX module data I can't load for the module '%s'. Probably you want to use a newer version of Redis which implements aux data callbacks", name); serverLog(LL_WARNING,"The RDB file contains module AUX data, but the module '%s' doesn't seem to support it.", name);
exit(1); exit(1);
}
RedisModuleIO io;
moduleInitIOContext(io,mt,rdb,NULL);
io.ver = 2;
/* Call the rdb_load method of the module providing the 10 bit
* encoding version in the lower 10 bits of the module ID. */
if (mt->aux_load(&io,moduleid&1023, when) || io.error) {
moduleTypeNameByID(name,moduleid);
serverLog(LL_WARNING,"The RDB file contains module AUX data for the module type '%s', that the responsible module is not able to load. Check for modules log above for additional clues.", name);
exit(1);
}
if (io.ctx) {
moduleFreeContext(io.ctx);
zfree(io.ctx);
}
uint64_t eof = rdbLoadLen(rdb,NULL);
if (eof != RDB_MODULE_OPCODE_EOF) {
serverLog(LL_WARNING,"The RDB file contains module AUX data for the module '%s' that is not terminated by the proper module value EOF marker", name);
exit(1);
}
continue;
} else { } else {
/* RDB check mode. */ /* RDB check mode. */
robj *aux = rdbLoadCheckModuleValue(rdb,name); robj *aux = rdbLoadCheckModuleValue(rdb,name);
decrRefCount(aux); decrRefCount(aux);
continue; /* Read next opcode. */
} }
} }
...@@ -2072,10 +2219,15 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) { ...@@ -2072,10 +2219,15 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) {
} }
return C_OK; return C_OK;
eoferr: /* unexpected end of file is handled here with a fatal exit */ /* Unexpected end of file is handled here calling rdbReportReadError():
serverLog(LL_WARNING,"Short read or OOM loading DB. Unrecoverable error, aborting now."); * this will in turn either abort Redis in most cases, or if we are loading
rdbExitReportCorruptRDB("Unexpected EOF reading RDB file"); * the RDB file from a socket during initial SYNC (diskless replica mode),
return C_ERR; /* Just to avoid warning */ * we'll report the error to the caller, so that we can retry. */
eoferr:
serverLog(LL_WARNING,
"Short read or OOM loading DB. Unrecoverable error, aborting now.");
rdbReportReadError("Unexpected EOF reading RDB file");
return C_ERR;
} }
/* Like rdbLoadRio() but takes a filename instead of a rio stream. The /* Like rdbLoadRio() but takes a filename instead of a rio stream. The
......
...@@ -145,6 +145,7 @@ size_t rdbSavedObjectLen(robj *o); ...@@ -145,6 +145,7 @@ size_t rdbSavedObjectLen(robj *o);
robj *rdbLoadObject(int type, rio *rdb, robj *key); robj *rdbLoadObject(int type, rio *rdb, robj *key);
void backgroundSaveDoneHandler(int exitcode, int bysignal); void backgroundSaveDoneHandler(int exitcode, int bysignal);
int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime); int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime);
ssize_t rdbSaveSingleModuleAux(rio *rdb, int when, moduleType *mt);
robj *rdbLoadStringObject(rio *rdb); robj *rdbLoadStringObject(rio *rdb);
ssize_t rdbSaveStringObject(rio *rdb, robj *obj); ssize_t rdbSaveStringObject(rio *rdb, robj *obj);
ssize_t rdbSaveRawString(rio *rdb, unsigned char *s, size_t len); ssize_t rdbSaveRawString(rio *rdb, unsigned char *s, size_t len);
......
...@@ -104,6 +104,7 @@ static struct config { ...@@ -104,6 +104,7 @@ static struct config {
int is_fetching_slots; int is_fetching_slots;
int is_updating_slots; int is_updating_slots;
int slots_last_update; int slots_last_update;
int enable_tracking;
/* Thread mutexes to be used as fallbacks by atomicvar.h */ /* Thread mutexes to be used as fallbacks by atomicvar.h */
pthread_mutex_t requests_issued_mutex; pthread_mutex_t requests_issued_mutex;
pthread_mutex_t requests_finished_mutex; pthread_mutex_t requests_finished_mutex;
...@@ -255,7 +256,7 @@ static redisConfig *getRedisConfig(const char *ip, int port, ...@@ -255,7 +256,7 @@ static redisConfig *getRedisConfig(const char *ip, int port,
goto fail; goto fail;
} }
if(config.auth){ if(config.auth) {
void *authReply = NULL; void *authReply = NULL;
redisAppendCommand(c, "AUTH %s", config.auth); redisAppendCommand(c, "AUTH %s", config.auth);
if (REDIS_OK != redisGetReply(c, &authReply)) goto fail; if (REDIS_OK != redisGetReply(c, &authReply)) goto fail;
...@@ -633,6 +634,14 @@ static client createClient(char *cmd, size_t len, client from, int thread_id) { ...@@ -633,6 +634,14 @@ static client createClient(char *cmd, size_t len, client from, int thread_id) {
c->prefix_pending++; c->prefix_pending++;
} }
if (config.enable_tracking) {
char *buf = NULL;
int len = redisFormatCommand(&buf, "CLIENT TRACKING on");
c->obuf = sdscatlen(c->obuf, buf, len);
free(buf);
c->prefix_pending++;
}
/* If a DB number different than zero is selected, prefix our request /* If a DB number different than zero is selected, prefix our request
* buffer with the SELECT command, that will be discarded the first * buffer with the SELECT command, that will be discarded the first
* time the replies are received, so if the client is reused the * time the replies are received, so if the client is reused the
...@@ -1350,6 +1359,8 @@ int parseOptions(int argc, const char **argv) { ...@@ -1350,6 +1359,8 @@ int parseOptions(int argc, const char **argv) {
} else if (config.num_threads < 0) config.num_threads = 0; } else if (config.num_threads < 0) config.num_threads = 0;
} else if (!strcmp(argv[i],"--cluster")) { } else if (!strcmp(argv[i],"--cluster")) {
config.cluster_mode = 1; config.cluster_mode = 1;
} else if (!strcmp(argv[i],"--enable-tracking")) {
config.enable_tracking = 1;
} else if (!strcmp(argv[i],"--help")) { } else if (!strcmp(argv[i],"--help")) {
exit_status = 0; exit_status = 0;
goto usage; goto usage;
...@@ -1380,6 +1391,7 @@ usage: ...@@ -1380,6 +1391,7 @@ usage:
" --dbnum <db> SELECT the specified db number (default 0)\n" " --dbnum <db> SELECT the specified db number (default 0)\n"
" --threads <num> Enable multi-thread mode.\n" " --threads <num> Enable multi-thread mode.\n"
" --cluster Enable cluster mode.\n" " --cluster Enable cluster mode.\n"
" --enable-tracking Send CLIENT TRACKING on before starting benchmark.\n"
" -k <boolean> 1=keep alive 0=reconnect (default 1)\n" " -k <boolean> 1=keep alive 0=reconnect (default 1)\n"
" -r <keyspacelen> Use random keys for SET/GET/INCR, random values for SADD\n" " -r <keyspacelen> Use random keys for SET/GET/INCR, random values for SADD\n"
" Using this option the benchmark will expand the string __rand_int__\n" " Using this option the benchmark will expand the string __rand_int__\n"
...@@ -1504,6 +1516,7 @@ int main(int argc, const char **argv) { ...@@ -1504,6 +1516,7 @@ int main(int argc, const char **argv) {
config.is_fetching_slots = 0; config.is_fetching_slots = 0;
config.is_updating_slots = 0; config.is_updating_slots = 0;
config.slots_last_update = 0; config.slots_last_update = 0;
config.enable_tracking = 0;
i = parseOptions(argc,argv); i = parseOptions(argc,argv);
argc -= i; argc -= i;
......
...@@ -216,14 +216,16 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) { ...@@ -216,14 +216,16 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) {
/* EXPIRETIME: load an expire associated with the next key /* EXPIRETIME: load an expire associated with the next key
* to load. Note that after loading an expire we need to * to load. Note that after loading an expire we need to
* load the actual type, and continue. */ * load the actual type, and continue. */
if ((expiretime = rdbLoadTime(&rdb)) == -1) goto eoferr; expiretime = rdbLoadTime(&rdb);
expiretime *= 1000; expiretime *= 1000;
if (rioGetReadError(&rdb)) goto eoferr;
continue; /* Read next opcode. */ continue; /* Read next opcode. */
} else if (type == RDB_OPCODE_EXPIRETIME_MS) { } else if (type == RDB_OPCODE_EXPIRETIME_MS) {
/* EXPIRETIME_MS: milliseconds precision expire times introduced /* EXPIRETIME_MS: milliseconds precision expire times introduced
* with RDB v3. Like EXPIRETIME but no with more precision. */ * with RDB v3. Like EXPIRETIME but no with more precision. */
rdbstate.doing = RDB_CHECK_DOING_READ_EXPIRE; rdbstate.doing = RDB_CHECK_DOING_READ_EXPIRE;
if ((expiretime = rdbLoadMillisecondTime(&rdb, rdbver)) == -1) goto eoferr; expiretime = rdbLoadMillisecondTime(&rdb, rdbver);
if (rioGetReadError(&rdb)) goto eoferr;
continue; /* Read next opcode. */ continue; /* Read next opcode. */
} else if (type == RDB_OPCODE_FREQ) { } else if (type == RDB_OPCODE_FREQ) {
/* FREQ: LFU frequency. */ /* FREQ: LFU frequency. */
......
...@@ -218,6 +218,7 @@ static struct config { ...@@ -218,6 +218,7 @@ static struct config {
int hotkeys; int hotkeys;
int stdinarg; /* get last arg from stdin. (-x option) */ int stdinarg; /* get last arg from stdin. (-x option) */
char *auth; char *auth;
char *user;
int output; /* output mode, see OUTPUT_* defines */ int output; /* output mode, see OUTPUT_* defines */
sds mb_delim; sds mb_delim;
char prompt[128]; char prompt[128];
...@@ -230,6 +231,7 @@ static struct config { ...@@ -230,6 +231,7 @@ static struct config {
int verbose; int verbose;
clusterManagerCommand cluster_manager_command; clusterManagerCommand cluster_manager_command;
int no_auth_warning; int no_auth_warning;
int resp3;
} config; } config;
/* User preferences. */ /* User preferences. */
...@@ -728,8 +730,13 @@ static int cliAuth(void) { ...@@ -728,8 +730,13 @@ static int cliAuth(void) {
redisReply *reply; redisReply *reply;
if (config.auth == NULL) return REDIS_OK; if (config.auth == NULL) return REDIS_OK;
reply = redisCommand(context,"AUTH %s",config.auth); if (config.user == NULL)
reply = redisCommand(context,"AUTH %s",config.auth);
else
reply = redisCommand(context,"AUTH %s %s",config.user,config.auth);
if (reply != NULL) { if (reply != NULL) {
if (reply->type == REDIS_REPLY_ERROR)
fprintf(stderr,"Warning: AUTH failed\n");
freeReplyObject(reply); freeReplyObject(reply);
return REDIS_OK; return REDIS_OK;
} }
...@@ -751,6 +758,21 @@ static int cliSelect(void) { ...@@ -751,6 +758,21 @@ static int cliSelect(void) {
return REDIS_ERR; return REDIS_ERR;
} }
/* Select RESP3 mode if redis-cli was started with the -3 option. */
static int cliSwitchProto(void) {
redisReply *reply;
if (config.resp3 == 0) return REDIS_OK;
reply = redisCommand(context,"HELLO 3");
if (reply != NULL) {
int result = REDIS_OK;
if (reply->type == REDIS_REPLY_ERROR) result = REDIS_ERR;
freeReplyObject(reply);
return result;
}
return REDIS_ERR;
}
/* Connect to the server. It is possible to pass certain flags to the function: /* Connect to the server. It is possible to pass certain flags to the function:
* CC_FORCE: The connection is performed even if there is already * CC_FORCE: The connection is performed even if there is already
* a connected socket. * a connected socket.
...@@ -788,11 +810,13 @@ static int cliConnect(int flags) { ...@@ -788,11 +810,13 @@ static int cliConnect(int flags) {
* errors. */ * errors. */
anetKeepAlive(NULL, context->fd, REDIS_CLI_KEEPALIVE_INTERVAL); anetKeepAlive(NULL, context->fd, REDIS_CLI_KEEPALIVE_INTERVAL);
/* Do AUTH and select the right DB. */ /* Do AUTH, select the right DB, switch to RESP3 if needed. */
if (cliAuth() != REDIS_OK) if (cliAuth() != REDIS_OK)
return REDIS_ERR; return REDIS_ERR;
if (cliSelect() != REDIS_OK) if (cliSelect() != REDIS_OK)
return REDIS_ERR; return REDIS_ERR;
if (cliSwitchProto() != REDIS_OK)
return REDIS_ERR;
} }
return REDIS_OK; return REDIS_OK;
} }
...@@ -819,10 +843,17 @@ static sds cliFormatReplyTTY(redisReply *r, char *prefix) { ...@@ -819,10 +843,17 @@ static sds cliFormatReplyTTY(redisReply *r, char *prefix) {
out = sdscatprintf(out,"(double) %s\n",r->str); out = sdscatprintf(out,"(double) %s\n",r->str);
break; break;
case REDIS_REPLY_STRING: case REDIS_REPLY_STRING:
case REDIS_REPLY_VERB:
/* If you are producing output for the standard output we want /* If you are producing output for the standard output we want
* a more interesting output with quoted characters and so forth */ * a more interesting output with quoted characters and so forth,
out = sdscatrepr(out,r->str,r->len); * unless it's a verbatim string type. */
out = sdscat(out,"\n"); if (r->type == REDIS_REPLY_STRING) {
out = sdscatrepr(out,r->str,r->len);
out = sdscat(out,"\n");
} else {
out = sdscatlen(out,r->str,r->len);
out = sdscat(out,"\n");
}
break; break;
case REDIS_REPLY_NIL: case REDIS_REPLY_NIL:
out = sdscat(out,"(nil)\n"); out = sdscat(out,"(nil)\n");
...@@ -961,6 +992,7 @@ static sds cliFormatReplyRaw(redisReply *r) { ...@@ -961,6 +992,7 @@ static sds cliFormatReplyRaw(redisReply *r) {
break; break;
case REDIS_REPLY_STATUS: case REDIS_REPLY_STATUS:
case REDIS_REPLY_STRING: case REDIS_REPLY_STRING:
case REDIS_REPLY_VERB:
if (r->type == REDIS_REPLY_STATUS && config.eval_ldb) { if (r->type == REDIS_REPLY_STATUS && config.eval_ldb) {
/* The Lua debugger replies with arrays of simple (status) /* The Lua debugger replies with arrays of simple (status)
* strings. We colorize the output for more fun if this * strings. We colorize the output for more fun if this
...@@ -980,9 +1012,15 @@ static sds cliFormatReplyRaw(redisReply *r) { ...@@ -980,9 +1012,15 @@ static sds cliFormatReplyRaw(redisReply *r) {
out = sdscatlen(out,r->str,r->len); out = sdscatlen(out,r->str,r->len);
} }
break; break;
case REDIS_REPLY_BOOL:
out = sdscat(out,r->integer ? "(true)" : "(false)");
break;
case REDIS_REPLY_INTEGER: case REDIS_REPLY_INTEGER:
out = sdscatprintf(out,"%lld",r->integer); out = sdscatprintf(out,"%lld",r->integer);
break; break;
case REDIS_REPLY_DOUBLE:
out = sdscatprintf(out,"%s",r->str);
break;
case REDIS_REPLY_ARRAY: case REDIS_REPLY_ARRAY:
for (i = 0; i < r->elements; i++) { for (i = 0; i < r->elements; i++) {
if (i > 0) out = sdscat(out,config.mb_delim); if (i > 0) out = sdscat(out,config.mb_delim);
...@@ -991,6 +1029,19 @@ static sds cliFormatReplyRaw(redisReply *r) { ...@@ -991,6 +1029,19 @@ static sds cliFormatReplyRaw(redisReply *r) {
sdsfree(tmp); sdsfree(tmp);
} }
break; break;
case REDIS_REPLY_MAP:
for (i = 0; i < r->elements; i += 2) {
if (i > 0) out = sdscat(out,config.mb_delim);
tmp = cliFormatReplyRaw(r->element[i]);
out = sdscatlen(out,tmp,sdslen(tmp));
sdsfree(tmp);
out = sdscatlen(out," ",1);
tmp = cliFormatReplyRaw(r->element[i+1]);
out = sdscatlen(out,tmp,sdslen(tmp));
sdsfree(tmp);
}
break;
default: default:
fprintf(stderr,"Unknown reply type: %d\n", r->type); fprintf(stderr,"Unknown reply type: %d\n", r->type);
exit(1); exit(1);
...@@ -1013,13 +1064,21 @@ static sds cliFormatReplyCSV(redisReply *r) { ...@@ -1013,13 +1064,21 @@ static sds cliFormatReplyCSV(redisReply *r) {
case REDIS_REPLY_INTEGER: case REDIS_REPLY_INTEGER:
out = sdscatprintf(out,"%lld",r->integer); out = sdscatprintf(out,"%lld",r->integer);
break; break;
case REDIS_REPLY_DOUBLE:
out = sdscatprintf(out,"%s",r->str);
break;
case REDIS_REPLY_STRING: case REDIS_REPLY_STRING:
case REDIS_REPLY_VERB:
out = sdscatrepr(out,r->str,r->len); out = sdscatrepr(out,r->str,r->len);
break; break;
case REDIS_REPLY_NIL: case REDIS_REPLY_NIL:
out = sdscat(out,"NIL"); out = sdscat(out,"NULL");
break;
case REDIS_REPLY_BOOL:
out = sdscat(out,r->integer ? "true" : "false");
break; break;
case REDIS_REPLY_ARRAY: case REDIS_REPLY_ARRAY:
case REDIS_REPLY_MAP: /* CSV has no map type, just output flat list. */
for (i = 0; i < r->elements; i++) { for (i = 0; i < r->elements; i++) {
sds tmp = cliFormatReplyCSV(r->element[i]); sds tmp = cliFormatReplyCSV(r->element[i]);
out = sdscatlen(out,tmp,sdslen(tmp)); out = sdscatlen(out,tmp,sdslen(tmp));
...@@ -1213,7 +1272,8 @@ static int cliSendCommand(int argc, char **argv, long repeat) { ...@@ -1213,7 +1272,8 @@ static int cliSendCommand(int argc, char **argv, long repeat) {
if (!strcasecmp(command,"select") && argc == 2 && config.last_cmd_type != REDIS_REPLY_ERROR) { if (!strcasecmp(command,"select") && argc == 2 && config.last_cmd_type != REDIS_REPLY_ERROR) {
config.dbnum = atoi(argv[1]); config.dbnum = atoi(argv[1]);
cliRefreshPrompt(); cliRefreshPrompt();
} else if (!strcasecmp(command,"auth") && argc == 2) { } else if (!strcasecmp(command,"auth") && (argc == 2 || argc == 3))
{
cliSelect(); cliSelect();
} }
} }
...@@ -1296,8 +1356,12 @@ static int parseOptions(int argc, char **argv) { ...@@ -1296,8 +1356,12 @@ static int parseOptions(int argc, char **argv) {
config.dbnum = atoi(argv[++i]); config.dbnum = atoi(argv[++i]);
} else if (!strcmp(argv[i], "--no-auth-warning")) { } else if (!strcmp(argv[i], "--no-auth-warning")) {
config.no_auth_warning = 1; config.no_auth_warning = 1;
} else if (!strcmp(argv[i],"-a") && !lastarg) { } else if ((!strcmp(argv[i],"-a") || !strcmp(argv[i],"--pass"))
&& !lastarg)
{
config.auth = argv[++i]; config.auth = argv[++i];
} else if (!strcmp(argv[i],"--user") && !lastarg) {
config.user = argv[++i];
} else if (!strcmp(argv[i],"-u") && !lastarg) { } else if (!strcmp(argv[i],"-u") && !lastarg) {
parseRedisUri(argv[++i]); parseRedisUri(argv[++i]);
} else if (!strcmp(argv[i],"--raw")) { } else if (!strcmp(argv[i],"--raw")) {
...@@ -1439,6 +1503,8 @@ static int parseOptions(int argc, char **argv) { ...@@ -1439,6 +1503,8 @@ static int parseOptions(int argc, char **argv) {
printf("redis-cli %s\n", version); printf("redis-cli %s\n", version);
sdsfree(version); sdsfree(version);
exit(0); exit(0);
} else if (!strcmp(argv[i],"-3")) {
config.resp3 = 1;
} else if (CLUSTER_MANAGER_MODE() && argv[i][0] != '-') { } else if (CLUSTER_MANAGER_MODE() && argv[i][0] != '-') {
if (config.cluster_manager_command.argc == 0) { if (config.cluster_manager_command.argc == 0) {
int j = i + 1; int j = i + 1;
...@@ -1514,11 +1580,14 @@ static void usage(void) { ...@@ -1514,11 +1580,14 @@ static void usage(void) {
" You can also use the " REDIS_CLI_AUTH_ENV " environment\n" " You can also use the " REDIS_CLI_AUTH_ENV " environment\n"
" variable to pass this password more safely\n" " variable to pass this password more safely\n"
" (if both are used, this argument takes predecence).\n" " (if both are used, this argument takes predecence).\n"
" -user <username> Used to send ACL style 'AUTH username pass'. Needs -a.\n"
" -pass <password> Alias of -a for consistency with the new --user option.\n"
" -u <uri> Server URI.\n" " -u <uri> Server URI.\n"
" -r <repeat> Execute specified command N times.\n" " -r <repeat> Execute specified command N times.\n"
" -i <interval> When -r is used, waits <interval> seconds per command.\n" " -i <interval> When -r is used, waits <interval> seconds per command.\n"
" It is possible to specify sub-second times like -i 0.1.\n" " It is possible to specify sub-second times like -i 0.1.\n"
" -n <db> Database number.\n" " -n <db> Database number.\n"
" -3 Start session in RESP3 protocol mode.\n"
" -x Read last argument from STDIN.\n" " -x Read last argument from STDIN.\n"
" -d <delimiter> Multi-bulk delimiter in for raw formatting (default: \\n).\n" " -d <delimiter> Multi-bulk delimiter in for raw formatting (default: \\n).\n"
" -c Enable cluster mode (follow -ASK and -MOVED redirections).\n" " -c Enable cluster mode (follow -ASK and -MOVED redirections).\n"
...@@ -1533,7 +1602,9 @@ static void usage(void) { ...@@ -1533,7 +1602,9 @@ static void usage(void) {
" --csv is specified, or if you redirect the output to a non\n" " --csv is specified, or if you redirect the output to a non\n"
" TTY, it samples the latency for 1 second (you can use\n" " TTY, it samples the latency for 1 second (you can use\n"
" -i to change the interval), then produces a single output\n" " -i to change the interval), then produces a single output\n"
" and exits.\n" " and exits.\n",version);
fprintf(stderr,
" --latency-history Like --latency but tracking latency changes over time.\n" " --latency-history Like --latency but tracking latency changes over time.\n"
" Default time interval is 15 sec. Change it using -i.\n" " Default time interval is 15 sec. Change it using -i.\n"
" --latency-dist Shows latency as a spectrum, requires xterm 256 colors.\n" " --latency-dist Shows latency as a spectrum, requires xterm 256 colors.\n"
...@@ -1568,7 +1639,7 @@ static void usage(void) { ...@@ -1568,7 +1639,7 @@ static void usage(void) {
" --help Output this help and exit.\n" " --help Output this help and exit.\n"
" --version Output version and exit.\n" " --version Output version and exit.\n"
"\n", "\n",
version, REDIS_CLI_DEFAULT_PIPE_TIMEOUT); REDIS_CLI_DEFAULT_PIPE_TIMEOUT);
/* Using another fprintf call to avoid -Woverlength-strings compile warning */ /* Using another fprintf call to avoid -Woverlength-strings compile warning */
fprintf(stderr, fprintf(stderr,
"Cluster Manager Commands:\n" "Cluster Manager Commands:\n"
...@@ -2350,7 +2421,12 @@ static int clusterManagerNodeConnect(clusterManagerNode *node) { ...@@ -2350,7 +2421,12 @@ static int clusterManagerNodeConnect(clusterManagerNode *node) {
* errors. */ * errors. */
anetKeepAlive(NULL, node->context->fd, REDIS_CLI_KEEPALIVE_INTERVAL); anetKeepAlive(NULL, node->context->fd, REDIS_CLI_KEEPALIVE_INTERVAL);
if (config.auth) { if (config.auth) {
redisReply *reply = redisCommand(node->context,"AUTH %s",config.auth); redisReply *reply;
if (config.user == NULL)
reply = redisCommand(node->context,"AUTH %s", config.auth);
else
reply = redisCommand(node->context,"AUTH %s %s",
config.user,config.auth);
int ok = clusterManagerCheckRedisReply(node, reply, NULL); int ok = clusterManagerCheckRedisReply(node, reply, NULL);
if (reply != NULL) freeReplyObject(reply); if (reply != NULL) freeReplyObject(reply);
if (!ok) return 0; if (!ok) return 0;
...@@ -6724,6 +6800,7 @@ static void pipeMode(void) { ...@@ -6724,6 +6800,7 @@ static void pipeMode(void) {
/* Handle the readable state: we can read replies from the server. */ /* Handle the readable state: we can read replies from the server. */
if (mask & AE_READABLE) { if (mask & AE_READABLE) {
ssize_t nread; ssize_t nread;
int read_error = 0;
/* Read from socket and feed the hiredis reader. */ /* Read from socket and feed the hiredis reader. */
do { do {
...@@ -6731,7 +6808,8 @@ static void pipeMode(void) { ...@@ -6731,7 +6808,8 @@ static void pipeMode(void) {
if (nread == -1 && errno != EAGAIN && errno != EINTR) { if (nread == -1 && errno != EAGAIN && errno != EINTR) {
fprintf(stderr, "Error reading from the server: %s\n", fprintf(stderr, "Error reading from the server: %s\n",
strerror(errno)); strerror(errno));
exit(1); read_error = 1;
break;
} }
if (nread > 0) { if (nread > 0) {
redisReaderFeed(reader,ibuf,nread); redisReaderFeed(reader,ibuf,nread);
...@@ -6764,6 +6842,11 @@ static void pipeMode(void) { ...@@ -6764,6 +6842,11 @@ static void pipeMode(void) {
freeReplyObject(reply); freeReplyObject(reply);
} }
} while(reply); } while(reply);
/* Abort on read errors. We abort here because it is important
* to consume replies even after a read error: this way we can
* show a potential problem to the user. */
if (read_error) exit(1);
} }
/* Handle the writable state: we can send protocol to the server. */ /* Handle the writable state: we can send protocol to the server. */
...@@ -7671,6 +7754,7 @@ int main(int argc, char **argv) { ...@@ -7671,6 +7754,7 @@ int main(int argc, char **argv) {
config.hotkeys = 0; config.hotkeys = 0;
config.stdinarg = 0; config.stdinarg = 0;
config.auth = NULL; config.auth = NULL;
config.user = NULL;
config.eval = NULL; config.eval = NULL;
config.eval_ldb = 0; config.eval_ldb = 0;
config.eval_ldb_end = 0; config.eval_ldb_end = 0;
......
...@@ -129,6 +129,10 @@ ...@@ -129,6 +129,10 @@
#define REDISMODULE_NOT_USED(V) ((void) V) #define REDISMODULE_NOT_USED(V) ((void) V)
/* Bit flags for aux_save_triggers and the aux_load and aux_save callbacks */
#define REDISMODULE_AUX_BEFORE_RDB (1<<0)
#define REDISMODULE_AUX_AFTER_RDB (1<<1)
/* This type represents a timer handle, and is returned when a timer is /* This type represents a timer handle, and is returned when a timer is
* registered and used in order to invalidate a timer. It's just a 64 bit * registered and used in order to invalidate a timer. It's just a 64 bit
* number, because this is how each timer is represented inside the radix tree * number, because this is how each timer is represented inside the radix tree
...@@ -140,6 +144,9 @@ typedef uint64_t RedisModuleTimerID; ...@@ -140,6 +144,9 @@ typedef uint64_t RedisModuleTimerID;
/* Do filter RedisModule_Call() commands initiated by module itself. */ /* Do filter RedisModule_Call() commands initiated by module itself. */
#define REDISMODULE_CMDFILTER_NOSELF (1<<0) #define REDISMODULE_CMDFILTER_NOSELF (1<<0)
/* Declare that the module can handle errors with RedisModule_SetModuleOptions. */
#define REDISMODULE_OPTIONS_HANDLE_IO_ERRORS (1<<0)
/* ------------------------- End of common defines ------------------------ */ /* ------------------------- End of common defines ------------------------ */
#ifndef REDISMODULE_CORE #ifndef REDISMODULE_CORE
...@@ -166,6 +173,8 @@ typedef void (*RedisModuleDisconnectFunc)(RedisModuleCtx *ctx, RedisModuleBlocke ...@@ -166,6 +173,8 @@ typedef void (*RedisModuleDisconnectFunc)(RedisModuleCtx *ctx, RedisModuleBlocke
typedef int (*RedisModuleNotificationFunc)(RedisModuleCtx *ctx, int type, const char *event, RedisModuleString *key); typedef int (*RedisModuleNotificationFunc)(RedisModuleCtx *ctx, int type, const char *event, RedisModuleString *key);
typedef void *(*RedisModuleTypeLoadFunc)(RedisModuleIO *rdb, int encver); typedef void *(*RedisModuleTypeLoadFunc)(RedisModuleIO *rdb, int encver);
typedef void (*RedisModuleTypeSaveFunc)(RedisModuleIO *rdb, void *value); typedef void (*RedisModuleTypeSaveFunc)(RedisModuleIO *rdb, void *value);
typedef int (*RedisModuleTypeAuxLoadFunc)(RedisModuleIO *rdb, int encver, int when);
typedef void (*RedisModuleTypeAuxSaveFunc)(RedisModuleIO *rdb, int when);
typedef void (*RedisModuleTypeRewriteFunc)(RedisModuleIO *aof, RedisModuleString *key, void *value); typedef void (*RedisModuleTypeRewriteFunc)(RedisModuleIO *aof, RedisModuleString *key, void *value);
typedef size_t (*RedisModuleTypeMemUsageFunc)(const void *value); typedef size_t (*RedisModuleTypeMemUsageFunc)(const void *value);
typedef void (*RedisModuleTypeDigestFunc)(RedisModuleDigest *digest, void *value); typedef void (*RedisModuleTypeDigestFunc)(RedisModuleDigest *digest, void *value);
...@@ -175,7 +184,7 @@ typedef void (*RedisModuleTimerProc)(RedisModuleCtx *ctx, void *data); ...@@ -175,7 +184,7 @@ typedef void (*RedisModuleTimerProc)(RedisModuleCtx *ctx, void *data);
typedef void (*RedisModuleCommandFilterFunc) (RedisModuleCommandFilterCtx *filter); typedef void (*RedisModuleCommandFilterFunc) (RedisModuleCommandFilterCtx *filter);
typedef void (*RedisModuleForkDoneHandler) (int exitcode, int bysignal, void *user_data); typedef void (*RedisModuleForkDoneHandler) (int exitcode, int bysignal, void *user_data);
#define REDISMODULE_TYPE_METHOD_VERSION 1 #define REDISMODULE_TYPE_METHOD_VERSION 2
typedef struct RedisModuleTypeMethods { typedef struct RedisModuleTypeMethods {
uint64_t version; uint64_t version;
RedisModuleTypeLoadFunc rdb_load; RedisModuleTypeLoadFunc rdb_load;
...@@ -184,6 +193,9 @@ typedef struct RedisModuleTypeMethods { ...@@ -184,6 +193,9 @@ typedef struct RedisModuleTypeMethods {
RedisModuleTypeMemUsageFunc mem_usage; RedisModuleTypeMemUsageFunc mem_usage;
RedisModuleTypeDigestFunc digest; RedisModuleTypeDigestFunc digest;
RedisModuleTypeFreeFunc free; RedisModuleTypeFreeFunc free;
RedisModuleTypeAuxLoadFunc aux_load;
RedisModuleTypeAuxSaveFunc aux_save;
int aux_save_triggers;
} RedisModuleTypeMethods; } RedisModuleTypeMethods;
#define REDISMODULE_GET_API(name) \ #define REDISMODULE_GET_API(name) \
...@@ -272,6 +284,8 @@ RedisModuleType *REDISMODULE_API_FUNC(RedisModule_CreateDataType)(RedisModuleCtx ...@@ -272,6 +284,8 @@ RedisModuleType *REDISMODULE_API_FUNC(RedisModule_CreateDataType)(RedisModuleCtx
int REDISMODULE_API_FUNC(RedisModule_ModuleTypeSetValue)(RedisModuleKey *key, RedisModuleType *mt, void *value); int REDISMODULE_API_FUNC(RedisModule_ModuleTypeSetValue)(RedisModuleKey *key, RedisModuleType *mt, void *value);
RedisModuleType *REDISMODULE_API_FUNC(RedisModule_ModuleTypeGetType)(RedisModuleKey *key); RedisModuleType *REDISMODULE_API_FUNC(RedisModule_ModuleTypeGetType)(RedisModuleKey *key);
void *REDISMODULE_API_FUNC(RedisModule_ModuleTypeGetValue)(RedisModuleKey *key); void *REDISMODULE_API_FUNC(RedisModule_ModuleTypeGetValue)(RedisModuleKey *key);
int REDISMODULE_API_FUNC(RedisModule_IsIOError)(RedisModuleIO *io);
void REDISMODULE_API_FUNC(RedisModule_SetModuleOptions)(RedisModuleCtx *ctx, int options);
void REDISMODULE_API_FUNC(RedisModule_SaveUnsigned)(RedisModuleIO *io, uint64_t value); void REDISMODULE_API_FUNC(RedisModule_SaveUnsigned)(RedisModuleIO *io, uint64_t value);
uint64_t REDISMODULE_API_FUNC(RedisModule_LoadUnsigned)(RedisModuleIO *io); uint64_t REDISMODULE_API_FUNC(RedisModule_LoadUnsigned)(RedisModuleIO *io);
void REDISMODULE_API_FUNC(RedisModule_SaveSigned)(RedisModuleIO *io, int64_t value); void REDISMODULE_API_FUNC(RedisModule_SaveSigned)(RedisModuleIO *io, int64_t value);
...@@ -287,6 +301,7 @@ void REDISMODULE_API_FUNC(RedisModule_SaveFloat)(RedisModuleIO *io, float value) ...@@ -287,6 +301,7 @@ void REDISMODULE_API_FUNC(RedisModule_SaveFloat)(RedisModuleIO *io, float value)
float REDISMODULE_API_FUNC(RedisModule_LoadFloat)(RedisModuleIO *io); float REDISMODULE_API_FUNC(RedisModule_LoadFloat)(RedisModuleIO *io);
void REDISMODULE_API_FUNC(RedisModule_Log)(RedisModuleCtx *ctx, const char *level, const char *fmt, ...); void REDISMODULE_API_FUNC(RedisModule_Log)(RedisModuleCtx *ctx, const char *level, const char *fmt, ...);
void REDISMODULE_API_FUNC(RedisModule_LogIOError)(RedisModuleIO *io, const char *levelstr, const char *fmt, ...); void REDISMODULE_API_FUNC(RedisModule_LogIOError)(RedisModuleIO *io, const char *levelstr, const char *fmt, ...);
void REDISMODULE_API_FUNC(RedisModule__Assert)(const char *estr, const char *file, int line);
int REDISMODULE_API_FUNC(RedisModule_StringAppendBuffer)(RedisModuleCtx *ctx, RedisModuleString *str, const char *buf, size_t len); int REDISMODULE_API_FUNC(RedisModule_StringAppendBuffer)(RedisModuleCtx *ctx, RedisModuleString *str, const char *buf, size_t len);
void REDISMODULE_API_FUNC(RedisModule_RetainString)(RedisModuleCtx *ctx, RedisModuleString *str); void REDISMODULE_API_FUNC(RedisModule_RetainString)(RedisModuleCtx *ctx, RedisModuleString *str);
int REDISMODULE_API_FUNC(RedisModule_StringCompare)(RedisModuleString *a, RedisModuleString *b); int REDISMODULE_API_FUNC(RedisModule_StringCompare)(RedisModuleString *a, RedisModuleString *b);
...@@ -448,6 +463,8 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -448,6 +463,8 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(ModuleTypeSetValue); REDISMODULE_GET_API(ModuleTypeSetValue);
REDISMODULE_GET_API(ModuleTypeGetType); REDISMODULE_GET_API(ModuleTypeGetType);
REDISMODULE_GET_API(ModuleTypeGetValue); REDISMODULE_GET_API(ModuleTypeGetValue);
REDISMODULE_GET_API(IsIOError);
REDISMODULE_GET_API(SetModuleOptions);
REDISMODULE_GET_API(SaveUnsigned); REDISMODULE_GET_API(SaveUnsigned);
REDISMODULE_GET_API(LoadUnsigned); REDISMODULE_GET_API(LoadUnsigned);
REDISMODULE_GET_API(SaveSigned); REDISMODULE_GET_API(SaveSigned);
...@@ -463,6 +480,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -463,6 +480,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(EmitAOF); REDISMODULE_GET_API(EmitAOF);
REDISMODULE_GET_API(Log); REDISMODULE_GET_API(Log);
REDISMODULE_GET_API(LogIOError); REDISMODULE_GET_API(LogIOError);
REDISMODULE_GET_API(_Assert);
REDISMODULE_GET_API(StringAppendBuffer); REDISMODULE_GET_API(StringAppendBuffer);
REDISMODULE_GET_API(RetainString); REDISMODULE_GET_API(RetainString);
REDISMODULE_GET_API(StringCompare); REDISMODULE_GET_API(StringCompare);
...@@ -542,6 +560,8 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -542,6 +560,8 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
return REDISMODULE_OK; return REDISMODULE_OK;
} }
#define RedisModule_Assert(_e) ((_e)?(void)0 : (RedisModule__Assert(#_e,__FILE__,__LINE__),exit(1)))
#else #else
/* Things only defined for the modules core, not exported to modules /* Things only defined for the modules core, not exported to modules
......
...@@ -823,7 +823,9 @@ void replconfCommand(client *c) { ...@@ -823,7 +823,9 @@ void replconfCommand(client *c) {
c->repl_ack_time = server.unixtime; c->repl_ack_time = server.unixtime;
/* If this was a diskless replication, we need to really put /* If this was a diskless replication, we need to really put
* the slave online when the first ACK is received (which * the slave online when the first ACK is received (which
* confirms slave is online and ready to get more data). */ * confirms slave is online and ready to get more data). This
* allows for simpler and less CPU intensive EOF detection
* when streaming RDB files. */
if (c->repl_put_online_on_ack && c->replstate == SLAVE_STATE_ONLINE) if (c->repl_put_online_on_ack && c->replstate == SLAVE_STATE_ONLINE)
putSlaveOnline(c); putSlaveOnline(c);
/* Note: this command does not reply anything! */ /* Note: this command does not reply anything! */
...@@ -842,18 +844,20 @@ void replconfCommand(client *c) { ...@@ -842,18 +844,20 @@ void replconfCommand(client *c) {
addReply(c,shared.ok); addReply(c,shared.ok);
} }
/* This function puts a slave in the online state, and should be called just /* This function puts a replica in the online state, and should be called just
* after a slave received the RDB file for the initial synchronization, and * after a replica received the RDB file for the initial synchronization, and
* we are finally ready to send the incremental stream of commands. * we are finally ready to send the incremental stream of commands.
* *
* It does a few things: * It does a few things:
* *
* 1) Put the slave in ONLINE state (useless when the function is called * 1) Put the slave in ONLINE state. Note that the function may also be called
* because state is already ONLINE but repl_put_online_on_ack is true). * for a replicas that are already in ONLINE state, but having the flag
* repl_put_online_on_ack set to true: we still have to install the write
* handler in that case. This function will take care of that.
* 2) Make sure the writable event is re-installed, since calling the SYNC * 2) Make sure the writable event is re-installed, since calling the SYNC
* command disables it, so that we can accumulate output buffer without * command disables it, so that we can accumulate output buffer without
* sending it to the slave. * sending it to the replica.
* 3) Update the count of good slaves. */ * 3) Update the count of "good replicas". */
void putSlaveOnline(client *slave) { void putSlaveOnline(client *slave) {
slave->replstate = SLAVE_STATE_ONLINE; slave->replstate = SLAVE_STATE_ONLINE;
slave->repl_put_online_on_ack = 0; slave->repl_put_online_on_ack = 0;
...@@ -965,11 +969,31 @@ void updateSlavesWaitingBgsave(int bgsaveerr, int type) { ...@@ -965,11 +969,31 @@ void updateSlavesWaitingBgsave(int bgsaveerr, int type) {
serverLog(LL_NOTICE, serverLog(LL_NOTICE,
"Streamed RDB transfer with replica %s succeeded (socket). Waiting for REPLCONF ACK from slave to enable streaming", "Streamed RDB transfer with replica %s succeeded (socket). Waiting for REPLCONF ACK from slave to enable streaming",
replicationGetSlaveName(slave)); replicationGetSlaveName(slave));
/* Note: we wait for a REPLCONF ACK message from slave in /* Note: we wait for a REPLCONF ACK message from the replica in
* order to really put it online (install the write handler * order to really put it online (install the write handler
* so that the accumulated data can be transferred). However * so that the accumulated data can be transferred). However
* we change the replication state ASAP, since our slave * we change the replication state ASAP, since our slave
* is technically online now. */ * is technically online now.
*
* So things work like that:
*
* 1. We end trasnferring the RDB file via socket.
* 2. The replica is put ONLINE but the write handler
* is not installed.
* 3. The replica however goes really online, and pings us
* back via REPLCONF ACK commands.
* 4. Now we finally install the write handler, and send
* the buffers accumulated so far to the replica.
*
* But why we do that? Because the replica, when we stream
* the RDB directly via the socket, must detect the RDB
* EOF (end of file), that is a special random string at the
* end of the RDB (for streamed RDBs we don't know the length
* in advance). Detecting such final EOF string is much
* simpler and less CPU intensive if no more data is sent
* after such final EOF. So we don't want to glue the end of
* the RDB trasfer with the start of the other replication
* data. */
slave->replstate = SLAVE_STATE_ONLINE; slave->replstate = SLAVE_STATE_ONLINE;
slave->repl_put_online_on_ack = 1; slave->repl_put_online_on_ack = 1;
slave->repl_ack_time = server.unixtime; /* Timeout otherwise. */ slave->repl_ack_time = server.unixtime; /* Timeout otherwise. */
...@@ -1115,8 +1139,15 @@ void restartAOFAfterSYNC() { ...@@ -1115,8 +1139,15 @@ void restartAOFAfterSYNC() {
static int useDisklessLoad() { static int useDisklessLoad() {
/* compute boolean decision to use diskless load */ /* compute boolean decision to use diskless load */
return server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB || int enabled = server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB ||
(server.repl_diskless_load == REPL_DISKLESS_LOAD_WHEN_DB_EMPTY && dbTotalServerKeyCount()==0); (server.repl_diskless_load == REPL_DISKLESS_LOAD_WHEN_DB_EMPTY && dbTotalServerKeyCount()==0);
/* Check all modules handle read errors, otherwise it's not safe to use diskless load. */
if (enabled && !moduleAllDatatypesHandleErrors()) {
serverLog(LL_WARNING,
"Skipping diskless-load because there are modules that don't handle read errors.");
enabled = 0;
}
return enabled;
} }
/* Helper function for readSyncBulkPayload() to make backups of the current /* Helper function for readSyncBulkPayload() to make backups of the current
......
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