Commit d5c14c70 authored by Oran Agra's avatar Oran Agra
Browse files

Merge remote-tracking branch 'antirez/unstable' into modules_info

parents 61853ad8 a9628142
......@@ -138,6 +138,7 @@ int clusterLoadConfig(char *filename) {
/* Handle the special "vars" line. Don't pretend it is the last
* line even if it actually is when generated by Redis. */
if (strcasecmp(argv[0],"vars") == 0) {
if (!(argc % 2)) goto fmterr;
for (j = 1; j < argc; j += 2) {
if (strcasecmp(argv[j],"currentEpoch") == 0) {
server.cluster->currentEpoch =
......@@ -4251,12 +4252,9 @@ NULL
}
} else if (!strcasecmp(c->argv[1]->ptr,"nodes") && c->argc == 2) {
/* CLUSTER NODES */
robj *o;
sds ci = clusterGenNodesDescription(0);
o = createObject(OBJ_STRING,ci);
addReplyBulk(c,o);
decrRefCount(o);
sds nodes = clusterGenNodesDescription(0);
addReplyVerbatim(c,nodes,sdslen(nodes),"txt");
sdsfree(nodes);
} else if (!strcasecmp(c->argv[1]->ptr,"myid") && c->argc == 2) {
/* CLUSTER MYID */
addReplyBulkCBuffer(c,myself->name, CLUSTER_NAMELEN);
......@@ -4498,10 +4496,8 @@ NULL
"cluster_stats_messages_received:%lld\r\n", tot_msg_received);
/* Produce the reply protocol. */
addReplySds(c,sdscatprintf(sdsempty(),"$%lu\r\n",
(unsigned long)sdslen(info)));
addReplySds(c,info);
addReply(c,shared.crlf);
addReplyVerbatim(c,info,sdslen(info),"txt");
sdsfree(info);
} else if (!strcasecmp(c->argv[1]->ptr,"saveconfig") && c->argc == 2) {
int retval = clusterSaveConfig(1);
......@@ -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
* complement of RESTORE and can be useful for different applications. */
void dumpCommand(client *c) {
robj *o, *dumpobj;
robj *o;
rio payload;
/* Check if the key is here. */
......@@ -4845,9 +4841,7 @@ void dumpCommand(client *c) {
createDumpPayload(&payload,o,c->argv[1]);
/* Transfer to the client */
dumpobj = createObject(OBJ_STRING,payload.io.buffer.ptr);
addReplyBulk(c,dumpobj);
decrRefCount(dumpobj);
addReplyBulkSds(c,payload.io.buffer.ptr);
return;
}
......
......@@ -672,6 +672,10 @@ void loadServerConfigFromString(char *config) {
server.lua_time_limit = strtoll(argv[1],NULL,10);
} else if (!strcasecmp(argv[0],"lua-replicate-commands") && argc == 2) {
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") &&
argc == 2)
{
......@@ -686,6 +690,17 @@ void loadServerConfigFromString(char *config) {
}
} else if (!strcasecmp(argv[0],"slowlog-max-len") && argc == 2) {
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") &&
argc == 5)
{
......@@ -1133,6 +1148,8 @@ void configSetCommand(client *c) {
"slowlog-max-len",ll,0,LONG_MAX) {
/* Cast to unsigned. */
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(
"latency-monitor-threshold",server.latency_monitor_threshold,0,LLONG_MAX){
} config_set_numerical_field(
......@@ -1338,8 +1355,8 @@ void configGetCommand(client *c) {
server.slowlog_log_slower_than);
config_get_numerical_field("latency-monitor-threshold",
server.latency_monitor_threshold);
config_get_numerical_field("slowlog-max-len",
server.slowlog_max_len);
config_get_numerical_field("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("cluster-announce-port",server.cluster_announce_port);
config_get_numerical_field("cluster-announce-bus-port",server.cluster_announce_bus_port);
......@@ -1470,12 +1487,10 @@ void configGetCommand(client *c) {
matches++;
}
if (stringmatch(pattern,"notify-keyspace-events",1)) {
robj *flagsobj = createObject(OBJ_STRING,
keyspaceEventsFlagsToString(server.notify_keyspace_events));
sds flags = keyspaceEventsFlagsToString(server.notify_keyspace_events);
addReplyBulkCString(c,"notify-keyspace-events");
addReplyBulk(c,flagsobj);
decrRefCount(flagsobj);
addReplyBulkSds(c,flags);
matches++;
}
if (stringmatch(pattern,"bind",1)) {
......@@ -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,"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,"tracking-table-max-fill",server.tracking_table_max_fill,CONFIG_DEFAULT_TRACKING_TABLE_MAX_FILL);
rewriteConfigNotifykeyspaceeventsOption(state);
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);
......
......@@ -60,10 +60,7 @@ robj *lookupKey(redisDb *db, robj *key, int flags) {
/* Update the access time for the ageing algorithm.
* Don't do it if we have a saving child, as this will trigger
* a copy on write madness. */
if (server.rdb_child_pid == -1 &&
server.aof_child_pid == -1 &&
!(flags & LOOKUP_NOTOUCH))
{
if (!hasActiveChildProcess() && !(flags & LOOKUP_NOTOUCH)){
if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
updateLFU(val);
} else {
......@@ -353,6 +350,11 @@ long long emptyDbGeneric(redisDb *dbarray, int dbnum, int flags, void(callback)(
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;
if (dbnum == -1) {
startdb = 0;
......@@ -412,12 +414,12 @@ long long dbTotalServerKeyCount() {
void signalModifiedKey(redisDb *db, robj *key) {
touchWatchedKey(db,key);
if (server.tracking_clients) trackingInvalidateKey(key);
trackingInvalidateKey(key);
}
void signalFlushedDb(int dbid) {
touchWatchedKeysOnFlush(dbid);
if (server.tracking_clients) trackingInvalidateKeysOnFlush(dbid);
trackingInvalidateKeysOnFlush(dbid);
}
/*-----------------------------------------------------------------------------
......@@ -453,7 +455,6 @@ void flushdbCommand(client *c) {
int flags;
if (getFlushCommandFlags(c,&flags) == C_ERR) return;
signalFlushedDb(c->db->id);
server.dirty += emptyDb(c->db->id,flags,NULL);
addReply(c,shared.ok);
}
......@@ -465,7 +466,6 @@ void flushallCommand(client *c) {
int flags;
if (getFlushCommandFlags(c,&flags) == C_ERR) return;
signalFlushedDb(-1);
server.dirty += emptyDb(-1,flags,NULL);
addReply(c,shared.ok);
if (server.rdb_child_pid != -1) killRDBChild();
......
......@@ -638,7 +638,8 @@ NULL
dictGetStats(buf,sizeof(buf),server.db[dbid].expires);
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) {
robj *o;
dict *ht = NULL;
......@@ -665,7 +666,7 @@ NULL
} else {
char buf[4096];
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) {
serverLog(LL_WARNING,"Changing replication IDs after receiving DEBUG change-repl-id");
......@@ -1110,6 +1111,33 @@ void logRegisters(ucontext_t *uc) {
(unsigned long) uc->uc_mcontext.mc_cs
);
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
serverLog(LL_WARNING,
" Dumping of registers not supported for this OS/arch");
......
......@@ -1039,7 +1039,7 @@ void activeDefragCycle(void) {
mstime_t latency;
int quit = 0;
if (server.aof_child_pid!=-1 || server.rdb_child_pid!=-1)
if (hasActiveChildProcess())
return; /* Defragging memory while there's a fork will just do damage. */
/* Once a second, check if we the fragmentation justfies starting a scan
......
......@@ -64,7 +64,7 @@ int activeExpireCycleTryExpire(redisDb *db, dictEntry *de, long long now) {
dbSyncDelete(db,keyobj);
notifyKeyspaceEvent(NOTIFY_EXPIRED,
"expired",keyobj,db->id);
if (server.tracking_clients) trackingInvalidateKey(keyobj);
trackingInvalidateKey(keyobj);
decrRefCount(keyobj);
server.stat_expiredkeys++;
return 1;
......
......@@ -700,7 +700,7 @@ int hllSparseSet(robj *o, long index, uint8_t count) {
p += oplen;
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;
if (next >= end) next = NULL;
......@@ -1242,7 +1242,7 @@ void pfcountCommand(client *c) {
if (o == NULL) continue; /* Assume empty HLL for non existing var.*/
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]). */
if (hllMerge(registers,o) == C_ERR) {
addReplySds(c,sdsnew(invalid_hll_err));
......@@ -1329,7 +1329,7 @@ void pfmergeCommand(client *c) {
hdr = o->ptr;
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]). */
if (hllMerge(max,o) == C_ERR) {
addReplySds(c,sdsnew(invalid_hll_err));
......
......@@ -599,7 +599,7 @@ NULL
event = dictGetKey(de);
graph = latencyCommandGenSparkeline(event,ts);
addReplyBulkCString(c,graph);
addReplyVerbatim(c,graph,sdslen(graph),"txt");
sdsfree(graph);
} else if (!strcasecmp(c->argv[1]->ptr,"latest") && c->argc == 2) {
/* LATENCY LATEST */
......@@ -608,7 +608,7 @@ NULL
/* LATENCY DOCTOR */
sds report = createLatencyReport();
addReplyBulkCBuffer(c,report,sdslen(report));
addReplyVerbatim(c,report,sdslen(report),"txt");
sdsfree(report);
} else if (!strcasecmp(c->argv[1]->ptr,"reset") && c->argc >= 2) {
/* LATENCY RESET */
......
......@@ -43,7 +43,8 @@ void lolwutUnstableCommand(client *c) {
sds rendered = sdsnew("Redis ver. ");
rendered = sdscat(rendered,REDIS_VERSION);
rendered = sdscatlen(rendered,"\n",1);
addReplyBulkSds(c,rendered);
addReplyVerbatim(c,rendered,sdslen(rendered),"txt");
sdsfree(rendered);
}
void lolwutCommand(client *c) {
......
......@@ -277,6 +277,7 @@ void lolwut5Command(client *c) {
"\nGeorg Nees - schotter, plotter on paper, 1968. Redis ver. ");
rendered = sdscat(rendered,REDIS_VERSION);
rendered = sdscatlen(rendered,"\n",1);
addReplyBulkSds(c,rendered);
addReplyVerbatim(c,rendered,sdslen(rendered),"txt");
sdsfree(rendered);
lwFreeCanvas(canvas);
}
This diff is collapsed.
......@@ -175,7 +175,19 @@ void execCommand(client *c) {
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. */
c->mstate.commands[j].argc = c->argc;
......
......@@ -1990,7 +1990,7 @@ NULL
return;
}
sds o = getAllClientsInfoString(type);
addReplyBulkCBuffer(c,o,sdslen(o));
addReplyVerbatim(c,o,sdslen(o),"txt");
sdsfree(o);
} else if (!strcasecmp(c->argv[1]->ptr,"reply") && c->argc == 3) {
/* CLIENT REPLY ON|OFF|SKIP */
......@@ -2468,17 +2468,27 @@ void flushSlavesOutputBuffers(void) {
listRewind(server.slaves,&li);
while((ln = listNext(&li))) {
client *slave = listNodeValue(ln);
int events;
/* Note that the following will not flush output buffers of slaves
* in STATE_ONLINE but having put_online_on_ack set to true: in this
* case the writable event is never installed, since the purpose
* of put_online_on_ack is to postpone the moment it is installed.
* This is what we want since slaves in this state should not receive
* writes before the first ACK. */
events = aeGetFileEvents(server.el,slave->fd);
if (events & AE_WRITABLE &&
slave->replstate == SLAVE_STATE_ONLINE &&
int events = aeGetFileEvents(server.el,slave->fd);
int can_receive_writes = (events & AE_WRITABLE) ||
(slave->flags & CLIENT_PENDING_WRITE);
/* We don't want to send the pending data to the replica in a few
* cases:
*
* 1. For some reason there is neither the write handler installed
* nor the client is flagged as to have pending writes: for some
* reason this replica may not be set to receive data. This is
* 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))
{
writeToClient(slave->fd,slave,0);
......
......@@ -467,10 +467,15 @@ robj *tryObjectEncoding(robj *o) {
incrRefCount(shared.integers[value]);
return shared.integers[value];
} else {
if (o->encoding == OBJ_ENCODING_RAW) sdsfree(o->ptr);
o->encoding = OBJ_ENCODING_INT;
o->ptr = (void*) value;
return o;
if (o->encoding == OBJ_ENCODING_RAW) {
sdsfree(o->ptr);
o->encoding = OBJ_ENCODING_INT;
o->ptr = (void*) value;
return o;
} else if (o->encoding == OBJ_ENCODING_EMBSTR) {
decrRefCount(o);
return createStringObjectFromLongLongForValue(value);
}
}
}
......@@ -1435,13 +1440,15 @@ NULL
#if defined(USE_JEMALLOC)
sds info = sdsempty();
je_malloc_stats_print(inputCatSds, &info, NULL);
addReplyBulkSds(c, info);
addReplyVerbatim(c,info,sdslen(info),"txt");
sdsfree(info);
#else
addReplyBulkCString(c,"Stats not supported for the current allocator");
#endif
} else if (!strcasecmp(c->argv[1]->ptr,"doctor") && c->argc == 2) {
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) {
#if defined(USE_JEMALLOC)
char tmp[32];
......
......@@ -260,7 +260,7 @@ int rdbEncodeInteger(long long value, unsigned char *enc) {
/* Loads an integer-encoded object with the specified encoding type "enctype".
* 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) {
int plain = flags & RDB_LOAD_PLAIN;
int sds = flags & RDB_LOAD_SDS;
......@@ -971,7 +971,6 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key) {
RedisModuleIO io;
moduleValue *mv = o->ptr;
moduleType *mt = mv->type;
moduleInitIOContext(io,mt,rdb,key);
/* Write the "module" identifier as prefix, so that we'll be able
* to call the right module during loading. */
......@@ -980,10 +979,13 @@ ssize_t rdbSaveObject(rio *rdb, robj *o, robj *key) {
io.bytes += retval;
/* Then write the module-specific representation + EOF marker. */
moduleInitIOContext(io,mt,rdb,key);
mt->rdb_save(&io,mv->value);
retval = rdbSaveLen(rdb,RDB_MODULE_OPCODE_EOF);
if (retval == -1) return -1;
io.bytes += retval;
if (retval == -1)
io.error = 1;
else
io.bytes += retval;
if (io.ctx) {
moduleFreeContext(io.ctx);
......@@ -1101,6 +1103,45 @@ int rdbSaveInfoAuxFields(rio *rdb, int flags, rdbSaveInfo *rsi) {
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
* 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
......@@ -1122,6 +1163,7 @@ int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi) {
snprintf(magic,sizeof(magic),"REDIS%04d",RDB_VERSION);
if (rdbWriteRaw(rdb,magic,9) == -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++) {
redisDb *db = server.db+j;
......@@ -1183,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. */
}
if (rdbSaveModulesAux(rdb, REDISMODULE_AUX_AFTER_RDB) == -1) goto werr;
/* EOF opcode */
if (rdbSaveType(rdb,RDB_OPCODE_EOF) == -1) goto werr;
......@@ -1291,40 +1335,25 @@ werr:
int rdbSaveBackground(char *filename, rdbSaveInfo *rsi) {
pid_t childpid;
long long start;
if (server.aof_child_pid != -1 || server.rdb_child_pid != -1) return C_ERR;
if (hasActiveChildProcess()) return C_ERR;
server.dirty_before_bgsave = server.dirty;
server.lastbgsave_try = time(NULL);
openChildInfoPipe();
start = ustime();
if ((childpid = fork()) == 0) {
if ((childpid = redisFork()) == 0) {
int retval;
/* Child */
closeListeningSockets(0);
redisSetProcTitle("redis-rdb-bgsave");
retval = rdbSave(filename,rsi);
if (retval == C_OK) {
size_t private_dirty = zmalloc_get_private_dirty(-1);
if (private_dirty) {
serverLog(LL_NOTICE,
"RDB: %zu MB of memory used by copy-on-write",
private_dirty/(1024*1024));
}
server.child_info_data.cow_size = private_dirty;
sendChildInfo(CHILD_INFO_TYPE_RDB);
sendChildCOWInfo(CHILD_INFO_TYPE_RDB, "RDB");
}
exitFromChild((retval == C_OK) ? 0 : 1);
} else {
/* Parent */
server.stat_fork_time = ustime()-start;
server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */
latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000);
if (childpid == -1) {
closeChildInfoPipe();
server.lastbgsave_status = C_ERR;
......@@ -1336,7 +1365,6 @@ int rdbSaveBackground(char *filename, rdbSaveInfo *rsi) {
server.rdb_save_time_start = time(NULL);
server.rdb_child_pid = childpid;
server.rdb_child_type = RDB_CHILD_TYPE_DISK;
updateDictResizePolicy();
return C_OK;
}
return C_OK; /* unreached */
......@@ -2089,16 +2117,15 @@ int rdbLoadRio(rio *rdb, rdbSaveInfo *rsi, int loading_aof) {
decrRefCount(auxval);
continue; /* Read type again. */
} else if (type == RDB_OPCODE_MODULE_AUX) {
/* This is just for compatibility with the future: we have plans
* to add the ability for modules to store anything in the RDB
* file, like data that is not related to the Redis key space.
* 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. */
/* Load module data that is not related to the Redis key space.
* Such data can be potentially be stored both before and after the
* RDB keys-values section. */
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);
char name[10];
moduleTypeNameByID(name,moduleid);
......@@ -2108,10 +2135,32 @@ 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);
exit(1);
} else if (!rdbCheckMode && mt != NULL) {
/* This version of Redis actually does not know what to do
* with modules AUX data... */
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);
exit(1);
if (!mt->aux_load) {
/* Module doesn't support AUX. */
serverLog(LL_WARNING,"The RDB file contains module AUX data, but the module '%s' doesn't seem to support it.", name);
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 {
/* RDB check mode. */
robj *aux = rdbLoadCheckModuleValue(rdb,name);
......@@ -2366,10 +2415,9 @@ int rdbSaveToSlavesSockets(rdbSaveInfo *rsi) {
listNode *ln;
listIter li;
pid_t childpid;
long long start;
int pipefds[2];
if (server.aof_child_pid != -1 || server.rdb_child_pid != -1) return C_ERR;
if (hasActiveChildProcess()) return C_ERR;
/* Before to fork, create a pipe that will be used in order to
* send back to the parent the IDs of the slaves that successfully
......@@ -2405,8 +2453,7 @@ int rdbSaveToSlavesSockets(rdbSaveInfo *rsi) {
/* Create the child process. */
openChildInfoPipe();
start = ustime();
if ((childpid = fork()) == 0) {
if ((childpid = redisFork()) == 0) {
/* Child */
int retval;
rio slave_sockets;
......@@ -2414,7 +2461,6 @@ int rdbSaveToSlavesSockets(rdbSaveInfo *rsi) {
rioInitWithFdset(&slave_sockets,fds,numfds);
zfree(fds);
closeListeningSockets(0);
redisSetProcTitle("redis-rdb-to-slaves");
retval = rdbSaveRioWithEOFMark(&slave_sockets,NULL,rsi);
......@@ -2422,16 +2468,7 @@ int rdbSaveToSlavesSockets(rdbSaveInfo *rsi) {
retval = C_ERR;
if (retval == C_OK) {
size_t private_dirty = zmalloc_get_private_dirty(-1);
if (private_dirty) {
serverLog(LL_NOTICE,
"RDB: %zu MB of memory used by copy-on-write",
private_dirty/(1024*1024));
}
server.child_info_data.cow_size = private_dirty;
sendChildInfo(CHILD_INFO_TYPE_RDB);
sendChildCOWInfo(CHILD_INFO_TYPE_RDB, "RDB");
/* If we are returning OK, at least one slave was served
* with the RDB file as expected, so we need to send a report
......@@ -2500,16 +2537,11 @@ int rdbSaveToSlavesSockets(rdbSaveInfo *rsi) {
close(pipefds[1]);
closeChildInfoPipe();
} else {
server.stat_fork_time = ustime()-start;
server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */
latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000);
serverLog(LL_NOTICE,"Background RDB transfer started by pid %d",
childpid);
server.rdb_save_time_start = time(NULL);
server.rdb_child_pid = childpid;
server.rdb_child_type = RDB_CHILD_TYPE_SOCKET;
updateDictResizePolicy();
}
zfree(clientids);
zfree(fds);
......@@ -2552,15 +2584,15 @@ void bgsaveCommand(client *c) {
if (server.rdb_child_pid != -1) {
addReplyError(c,"Background save already in progress");
} else if (server.aof_child_pid != -1) {
} else if (hasActiveChildProcess()) {
if (schedule) {
server.rdb_bgsave_scheduled = 1;
addReplyStatus(c,"Background saving scheduled");
} else {
addReplyError(c,
"An AOF log rewriting in progress: can't BGSAVE right now. "
"Use BGSAVE SCHEDULE in order to schedule a BGSAVE whenever "
"possible.");
"Another child process is active (AOF?): can't BGSAVE right now. "
"Use BGSAVE SCHEDULE in order to schedule a BGSAVE whenever "
"possible.");
}
} else if (rdbSaveBackground(server.rdb_filename,rsiptr) == C_OK) {
addReplyStatus(c,"Background saving started");
......
......@@ -145,6 +145,7 @@ size_t rdbSavedObjectLen(robj *o);
robj *rdbLoadObject(int type, rio *rdb, robj *key);
void backgroundSaveDoneHandler(int exitcode, int bysignal);
int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime);
ssize_t rdbSaveSingleModuleAux(rio *rdb, int when, moduleType *mt);
robj *rdbLoadStringObject(rio *rdb);
ssize_t rdbSaveStringObject(rio *rdb, robj *obj);
ssize_t rdbSaveRawString(rio *rdb, unsigned char *s, size_t len);
......
......@@ -104,6 +104,7 @@ static struct config {
int is_fetching_slots;
int is_updating_slots;
int slots_last_update;
int enable_tracking;
/* Thread mutexes to be used as fallbacks by atomicvar.h */
pthread_mutex_t requests_issued_mutex;
pthread_mutex_t requests_finished_mutex;
......@@ -255,7 +256,7 @@ static redisConfig *getRedisConfig(const char *ip, int port,
goto fail;
}
if(config.auth){
if(config.auth) {
void *authReply = NULL;
redisAppendCommand(c, "AUTH %s", config.auth);
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) {
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
* buffer with the SELECT command, that will be discarded the first
* time the replies are received, so if the client is reused the
......@@ -1350,6 +1359,8 @@ int parseOptions(int argc, const char **argv) {
} else if (config.num_threads < 0) config.num_threads = 0;
} else if (!strcmp(argv[i],"--cluster")) {
config.cluster_mode = 1;
} else if (!strcmp(argv[i],"--enable-tracking")) {
config.enable_tracking = 1;
} else if (!strcmp(argv[i],"--help")) {
exit_status = 0;
goto usage;
......@@ -1380,6 +1391,7 @@ usage:
" --dbnum <db> SELECT the specified db number (default 0)\n"
" --threads <num> Enable multi-thread 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"
" -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"
......@@ -1504,6 +1516,7 @@ int main(int argc, const char **argv) {
config.is_fetching_slots = 0;
config.is_updating_slots = 0;
config.slots_last_update = 0;
config.enable_tracking = 0;
i = parseOptions(argc,argv);
argc -= i;
......
This diff is collapsed.
This diff is collapsed.
This diff is collapsed.
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment