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

Merge remote-tracking branch 'origin/unstable' into 7.0

parents d375595d f22bfe86
/* Automatically generated by ./generate-command-help.rb, do not edit. */ /* Automatically generated by ./utils/generate-command-help.rb, do not edit. */
#ifndef __REDIS_HELP_H #ifndef __REDIS_HELP_H
#define __REDIS_HELP_H #define __REDIS_HELP_H
...@@ -135,7 +135,7 @@ struct commandHelp { ...@@ -135,7 +135,7 @@ struct commandHelp {
15, 15,
"3.2.0" }, "3.2.0" },
{ "BITFIELD_RO", { "BITFIELD_RO",
"key GET encoding offset [encoding offset ...]", "key GET encoding offset [GET encoding offset ...]",
"Perform arbitrary bitfield integer operations on strings. Read-only variant of BITFIELD", "Perform arbitrary bitfield integer operations on strings. Read-only variant of BITFIELD",
15, 15,
"6.2.0" }, "6.2.0" },
...@@ -255,7 +255,7 @@ struct commandHelp { ...@@ -255,7 +255,7 @@ struct commandHelp {
8, 8,
"2.6.9" }, "2.6.9" },
{ "CLIENT TRACKING", { "CLIENT TRACKING",
"ON|OFF [REDIRECT client-id] [PREFIX prefix [prefix ...]] [BCAST] [OPTIN] [OPTOUT] [NOLOOP]", "ON|OFF [REDIRECT client-id] [PREFIX prefix [PREFIX prefix ...]] [BCAST] [OPTIN] [OPTOUT] [NOLOOP]",
"Enable or disable server assisted client side caching support", "Enable or disable server assisted client side caching support",
8, 8,
"6.0.0" }, "6.0.0" },
...@@ -1025,7 +1025,7 @@ struct commandHelp { ...@@ -1025,7 +1025,7 @@ struct commandHelp {
9, 9,
"4.0.0" }, "4.0.0" },
{ "MODULE LOADEX", { "MODULE LOADEX",
"path [CONFIG name value [name value ...]] [ARGS arg [arg ...]]", "path [CONFIG name value [CONFIG name value ...]] [ARGS arg [arg ...]]",
"Load a module with extended parameters", "Load a module with extended parameters",
9, 9,
"7.0.0" }, "7.0.0" },
...@@ -1120,7 +1120,7 @@ struct commandHelp { ...@@ -1120,7 +1120,7 @@ struct commandHelp {
11, 11,
"2.8.9" }, "2.8.9" },
{ "PFDEBUG", { "PFDEBUG",
"", "subcommand key",
"Internal commands for debugging HyperLogLog values", "Internal commands for debugging HyperLogLog values",
11, 11,
"2.8.9" }, "2.8.9" },
...@@ -1195,7 +1195,7 @@ struct commandHelp { ...@@ -1195,7 +1195,7 @@ struct commandHelp {
6, 6,
"7.0.0" }, "7.0.0" },
{ "PUBSUB SHARDNUMSUB", { "PUBSUB SHARDNUMSUB",
"[channel [channel ...]]", "[shardchannel [shardchannel ...]]",
"Get the count of subscribers for shard channels", "Get the count of subscribers for shard channels",
6, 6,
"7.0.0" }, "7.0.0" },
...@@ -1255,7 +1255,7 @@ struct commandHelp { ...@@ -1255,7 +1255,7 @@ struct commandHelp {
0, 0,
"2.6.0" }, "2.6.0" },
{ "RESTORE-ASKING", { "RESTORE-ASKING",
"", "key ttl serialized-value [REPLACE] [ABSTTL] [IDLETIME seconds] [FREQ frequency]",
"An internal command for migrating keys in a cluster", "An internal command for migrating keys in a cluster",
9, 9,
"3.0.0" }, "3.0.0" },
...@@ -1450,12 +1450,12 @@ struct commandHelp { ...@@ -1450,12 +1450,12 @@ struct commandHelp {
3, 3,
"1.0.0" }, "1.0.0" },
{ "SORT", { "SORT",
"key [BY pattern] [LIMIT offset count] [GET pattern [pattern ...]] [ASC|DESC] [ALPHA] [STORE destination]", "key [BY pattern] [LIMIT offset count] [GET pattern [GET pattern ...]] [ASC|DESC] [ALPHA] [STORE destination]",
"Sort the elements in a list, set or sorted set", "Sort the elements in a list, set or sorted set",
0, 0,
"1.0.0" }, "1.0.0" },
{ "SORT_RO", { "SORT_RO",
"key [BY pattern] [LIMIT offset count] [GET pattern [pattern ...]] [ASC|DESC] [ALPHA]", "key [BY pattern] [LIMIT offset count] [GET pattern [GET pattern ...]] [ASC|DESC] [ALPHA]",
"Sort the elements in a list, set or sorted set. Read-only variant of SORT.", "Sort the elements in a list, set or sorted set. Read-only variant of SORT.",
0, 0,
"7.0.0" }, "7.0.0" },
...@@ -1465,7 +1465,7 @@ struct commandHelp { ...@@ -1465,7 +1465,7 @@ struct commandHelp {
3, 3,
"1.0.0" }, "1.0.0" },
{ "SPUBLISH", { "SPUBLISH",
"channel message", "shardchannel message",
"Post a message to a shard channel", "Post a message to a shard channel",
6, 6,
"7.0.0" }, "7.0.0" },
...@@ -1485,7 +1485,7 @@ struct commandHelp { ...@@ -1485,7 +1485,7 @@ struct commandHelp {
3, 3,
"2.8.0" }, "2.8.0" },
{ "SSUBSCRIBE", { "SSUBSCRIBE",
"channel [channel ...]", "shardchannel [shardchannel ...]",
"Listen for messages published to the given shard channels", "Listen for messages published to the given shard channels",
6, 6,
"7.0.0" }, "7.0.0" },
...@@ -1515,7 +1515,7 @@ struct commandHelp { ...@@ -1515,7 +1515,7 @@ struct commandHelp {
3, 3,
"1.0.0" }, "1.0.0" },
{ "SUNSUBSCRIBE", { "SUNSUBSCRIBE",
"[channel [channel ...]]", "[shardchannel [shardchannel ...]]",
"Stop listening for messages posted to the given shard channels", "Stop listening for messages posted to the given shard channels",
6, 6,
"7.0.0" }, "7.0.0" },
......
...@@ -1495,8 +1495,13 @@ cleanup: ...@@ -1495,8 +1495,13 @@ cleanup:
if (o) decrRefCount(o); if (o) decrRefCount(o);
} }
/* PFDEBUG <subcommand> <key> ... args ... /* Different debugging related operations about the HLL implementation.
* Different debugging related operations about the HLL implementation. */ *
* PFDEBUG GETREG <key>
* PFDEBUG DECODE <key>
* PFDEBUG ENCODING <key>
* PFDEBUG TODENSE <key>
*/
void pfdebugCommand(client *c) { void pfdebugCommand(client *c) {
char *cmd = c->argv[1]->ptr; char *cmd = c->argv[1]->ptr;
struct hllhdr *hdr; struct hllhdr *hdr;
......
...@@ -59,39 +59,6 @@ dictType latencyTimeSeriesDictType = { ...@@ -59,39 +59,6 @@ dictType latencyTimeSeriesDictType = {
/* ------------------------- Utility functions ------------------------------ */ /* ------------------------- Utility functions ------------------------------ */
#ifdef __linux__
#include <sys/prctl.h>
/* Returns 1 if Transparent Huge Pages support is enabled in the kernel.
* Otherwise (or if we are unable to check) 0 is returned. */
int THPIsEnabled(void) {
char buf[1024];
FILE *fp = fopen("/sys/kernel/mm/transparent_hugepage/enabled","r");
if (!fp) return 0;
if (fgets(buf,sizeof(buf),fp) == NULL) {
fclose(fp);
return 0;
}
fclose(fp);
return (strstr(buf,"[always]") != NULL) ? 1 : 0;
}
/* since linux-3.5, kernel supports to set the state of the "THP disable" flag
* for the calling thread. PR_SET_THP_DISABLE is defined in linux/prctl.h */
int THPDisable(void) {
int ret = -EINVAL;
if (!server.disable_thp)
return ret;
#ifdef PR_SET_THP_DISABLE
ret = prctl(PR_SET_THP_DISABLE, 1, 0, 0, 0);
#endif
return ret;
}
#endif
/* Report the amount of AnonHugePages in smap, in bytes. If the return /* Report the amount of AnonHugePages in smap, in bytes. If the return
* value of the function is non-zero, the process is being targeted by * value of the function is non-zero, the process is being targeted by
* THP support, and is likely to have memory usage / latency issues. */ * THP support, and is likely to have memory usage / latency issues. */
......
...@@ -63,8 +63,6 @@ struct latencyStats { ...@@ -63,8 +63,6 @@ struct latencyStats {
void latencyMonitorInit(void); void latencyMonitorInit(void);
void latencyAddSample(const char *event, mstime_t latency); void latencyAddSample(const char *event, mstime_t latency);
int THPIsEnabled(void);
int THPDisable(void);
/* Latency monitoring macros. */ /* Latency monitoring macros. */
......
This diff is collapsed.
...@@ -56,7 +56,7 @@ void freeClientMultiState(client *c) { ...@@ -56,7 +56,7 @@ void freeClientMultiState(client *c) {
} }
/* Add a new command into the MULTI commands queue */ /* Add a new command into the MULTI commands queue */
void queueMultiCommand(client *c) { void queueMultiCommand(client *c, uint64_t cmd_flags) {
multiCmd *mc; multiCmd *mc;
/* No sense to waste memory if the transaction is already aborted. /* No sense to waste memory if the transaction is already aborted.
...@@ -75,8 +75,8 @@ void queueMultiCommand(client *c) { ...@@ -75,8 +75,8 @@ void queueMultiCommand(client *c) {
mc->argv_len = c->argv_len; mc->argv_len = c->argv_len;
c->mstate.count++; c->mstate.count++;
c->mstate.cmd_flags |= c->cmd->flags; c->mstate.cmd_flags |= cmd_flags;
c->mstate.cmd_inv_flags |= ~c->cmd->flags; c->mstate.cmd_inv_flags |= ~cmd_flags;
c->mstate.argv_len_sums += c->argv_len_sum + sizeof(robj*)*c->argc; c->mstate.argv_len_sums += c->argv_len_sum + sizeof(robj*)*c->argc;
/* Reset the client's args since we copied them into the mstate and shouldn't /* Reset the client's args since we copied them into the mstate and shouldn't
......
...@@ -1953,7 +1953,13 @@ int writeToClient(client *c, int handler_installed) { ...@@ -1953,7 +1953,13 @@ int writeToClient(client *c, int handler_installed) {
zmalloc_used_memory() < server.maxmemory) && zmalloc_used_memory() < server.maxmemory) &&
!(c->flags & CLIENT_SLAVE)) break; !(c->flags & CLIENT_SLAVE)) break;
} }
atomicIncr(server.stat_net_output_bytes, totwritten);
if (getClientType(c) == CLIENT_TYPE_SLAVE) {
atomicIncr(server.stat_net_repl_output_bytes, totwritten);
} else {
atomicIncr(server.stat_net_output_bytes, totwritten);
}
if (nwritten == -1) { if (nwritten == -1) {
if (connGetState(c->conn) != CONN_STATE_CONNECTED) { if (connGetState(c->conn) != CONN_STATE_CONNECTED) {
serverLog(LL_VERBOSE, serverLog(LL_VERBOSE,
...@@ -2499,7 +2505,7 @@ int processInputBuffer(client *c) { ...@@ -2499,7 +2505,7 @@ int processInputBuffer(client *c) {
* condition on the slave. We want just to accumulate the replication * condition on the slave. We want just to accumulate the replication
* stream (instead of replying -BUSY like we do with other clients) and * stream (instead of replying -BUSY like we do with other clients) and
* later resume the processing. */ * later resume the processing. */
if (scriptIsTimedout() && c->flags & CLIENT_MASTER) break; if (isInsideYieldingLongCommand() && c->flags & CLIENT_MASTER) break;
/* CLIENT_CLOSE_AFTER_REPLY closes the connection once the reply is /* CLIENT_CLOSE_AFTER_REPLY closes the connection once the reply is
* written to the client. Make sure to not let the reply grow after * written to the client. Make sure to not let the reply grow after
...@@ -2655,8 +2661,13 @@ void readQueryFromClient(connection *conn) { ...@@ -2655,8 +2661,13 @@ void readQueryFromClient(connection *conn) {
if (c->querybuf_peak < qblen) c->querybuf_peak = qblen; if (c->querybuf_peak < qblen) c->querybuf_peak = qblen;
c->lastinteraction = server.unixtime; c->lastinteraction = server.unixtime;
if (c->flags & CLIENT_MASTER) c->read_reploff += nread; if (c->flags & CLIENT_MASTER) {
atomicIncr(server.stat_net_input_bytes, nread); c->read_reploff += nread;
atomicIncr(server.stat_net_repl_input_bytes, nread);
} else {
atomicIncr(server.stat_net_input_bytes, nread);
}
if (!(c->flags & CLIENT_MASTER) && sdslen(c->querybuf) > server.client_max_querybuf_len) { if (!(c->flags & CLIENT_MASTER) && sdslen(c->querybuf) > server.client_max_querybuf_len) {
sds ci = catClientInfoString(sdsempty(),c), bytes = sdsempty(); sds ci = catClientInfoString(sdsempty(),c), bytes = sdsempty();
...@@ -3557,7 +3568,6 @@ void replaceClientCommandVector(client *c, int argc, robj **argv) { ...@@ -3557,7 +3568,6 @@ void replaceClientCommandVector(client *c, int argc, robj **argv) {
int j; int j;
retainOriginalCommandVector(c); retainOriginalCommandVector(c);
freeClientArgv(c); freeClientArgv(c);
zfree(c->argv);
c->argv = argv; c->argv = argv;
c->argc = argc; c->argc = argc;
c->argv_len_sum = 0; c->argv_len_sum = 0;
......
...@@ -930,7 +930,6 @@ char *strEncoding(int encoding) { ...@@ -930,7 +930,6 @@ char *strEncoding(int encoding) {
case OBJ_ENCODING_INT: return "int"; case OBJ_ENCODING_INT: return "int";
case OBJ_ENCODING_HT: return "hashtable"; case OBJ_ENCODING_HT: return "hashtable";
case OBJ_ENCODING_QUICKLIST: return "quicklist"; case OBJ_ENCODING_QUICKLIST: return "quicklist";
case OBJ_ENCODING_ZIPLIST: return "ziplist";
case OBJ_ENCODING_LISTPACK: return "listpack"; case OBJ_ENCODING_LISTPACK: return "listpack";
case OBJ_ENCODING_INTSET: return "intset"; case OBJ_ENCODING_INTSET: return "intset";
case OBJ_ENCODING_SKIPLIST: return "skiplist"; case OBJ_ENCODING_SKIPLIST: return "skiplist";
...@@ -998,8 +997,6 @@ size_t objectComputeSize(robj *key, robj *o, size_t sample_size, int dbid) { ...@@ -998,8 +997,6 @@ size_t objectComputeSize(robj *key, robj *o, size_t sample_size, int dbid) {
samples++; samples++;
} while ((node = node->next) && samples < sample_size); } while ((node = node->next) && samples < sample_size);
asize += (double)elesize/samples*ql->len; asize += (double)elesize/samples*ql->len;
} else if (o->encoding == OBJ_ENCODING_ZIPLIST) {
asize = sizeof(*o)+zmalloc_size(o->ptr);
} else { } else {
serverPanic("Unknown list encoding"); serverPanic("Unknown list encoding");
} }
......
...@@ -39,6 +39,7 @@ typedef struct pubsubtype { ...@@ -39,6 +39,7 @@ typedef struct pubsubtype {
dict **serverPubSubChannels; dict **serverPubSubChannels;
robj **subscribeMsg; robj **subscribeMsg;
robj **unsubscribeMsg; robj **unsubscribeMsg;
robj **messageBulk;
}pubsubtype; }pubsubtype;
/* /*
...@@ -78,6 +79,7 @@ pubsubtype pubSubType = { ...@@ -78,6 +79,7 @@ pubsubtype pubSubType = {
.serverPubSubChannels = &server.pubsub_channels, .serverPubSubChannels = &server.pubsub_channels,
.subscribeMsg = &shared.subscribebulk, .subscribeMsg = &shared.subscribebulk,
.unsubscribeMsg = &shared.unsubscribebulk, .unsubscribeMsg = &shared.unsubscribebulk,
.messageBulk = &shared.messagebulk,
}; };
/* /*
...@@ -89,7 +91,8 @@ pubsubtype pubSubShardType = { ...@@ -89,7 +91,8 @@ pubsubtype pubSubShardType = {
.subscriptionCount = clientShardSubscriptionsCount, .subscriptionCount = clientShardSubscriptionsCount,
.serverPubSubChannels = &server.pubsubshard_channels, .serverPubSubChannels = &server.pubsubshard_channels,
.subscribeMsg = &shared.ssubscribebulk, .subscribeMsg = &shared.ssubscribebulk,
.unsubscribeMsg = &shared.sunsubscribebulk .unsubscribeMsg = &shared.sunsubscribebulk,
.messageBulk = &shared.smessagebulk,
}; };
/*----------------------------------------------------------------------------- /*-----------------------------------------------------------------------------
...@@ -101,12 +104,12 @@ pubsubtype pubSubShardType = { ...@@ -101,12 +104,12 @@ pubsubtype pubSubShardType = {
* message. However if the caller sets 'msg' as NULL, it will be able * message. However if the caller sets 'msg' as NULL, it will be able
* to send a special message (for instance an Array type) by using the * to send a special message (for instance an Array type) by using the
* addReply*() API family. */ * addReply*() API family. */
void addReplyPubsubMessage(client *c, robj *channel, robj *msg) { void addReplyPubsubMessage(client *c, robj *channel, robj *msg, robj *message_bulk) {
if (c->resp == 2) if (c->resp == 2)
addReply(c,shared.mbulkhdr[3]); addReply(c,shared.mbulkhdr[3]);
else else
addReplyPushLen(c,3); addReplyPushLen(c,3);
addReply(c,shared.messagebulk); addReply(c,message_bulk);
addReplyBulk(c,channel); addReplyBulk(c,channel);
if (msg) addReplyBulk(c,msg); if (msg) addReplyBulk(c,msg);
} }
...@@ -461,7 +464,7 @@ int pubsubPublishMessageInternal(robj *channel, robj *message, pubsubtype type) ...@@ -461,7 +464,7 @@ int pubsubPublishMessageInternal(robj *channel, robj *message, pubsubtype type)
listRewind(list,&li); listRewind(list,&li);
while ((ln = listNext(&li)) != NULL) { while ((ln = listNext(&li)) != NULL) {
client *c = ln->value; client *c = ln->value;
addReplyPubsubMessage(c,channel,message); addReplyPubsubMessage(c,channel,message,*type.messageBulk);
updateClientMemUsage(c); updateClientMemUsage(c);
receivers++; receivers++;
} }
...@@ -604,10 +607,10 @@ void pubsubCommand(client *c) { ...@@ -604,10 +607,10 @@ void pubsubCommand(client *c) {
" Return number of subscriptions to patterns.", " Return number of subscriptions to patterns.",
"NUMSUB [<channel> ...]", "NUMSUB [<channel> ...]",
" Return the number of subscribers for the specified channels, excluding", " Return the number of subscribers for the specified channels, excluding",
" pattern subscriptions(default: no channels)." " pattern subscriptions(default: no channels).",
"SHARDCHANNELS [<pattern>]", "SHARDCHANNELS [<pattern>]",
" Return the currently active shard level channels matching a <pattern> (default: '*').", " Return the currently active shard level channels matching a <pattern> (default: '*').",
"SHARDNUMSUB [<channel> ...]", "SHARDNUMSUB [<shardchannel> ...]",
" Return the number of subscribers for the specified shard level channel(s)", " Return the number of subscribers for the specified shard level channel(s)",
NULL NULL
}; };
...@@ -639,7 +642,7 @@ NULL ...@@ -639,7 +642,7 @@ NULL
sds pat = (c->argc == 2) ? NULL : c->argv[2]->ptr; sds pat = (c->argc == 2) ? NULL : c->argv[2]->ptr;
channelList(c,pat,server.pubsubshard_channels); channelList(c,pat,server.pubsubshard_channels);
} else if (!strcasecmp(c->argv[1]->ptr,"shardnumsub") && c->argc >= 2) { } else if (!strcasecmp(c->argv[1]->ptr,"shardnumsub") && c->argc >= 2) {
/* PUBSUB SHARDNUMSUB [Channel_1 ... Channel_N] */ /* PUBSUB SHARDNUMSUB [ShardChannel_1 ... ShardChannel_N] */
int j; int j;
addReplyArrayLen(c, (c->argc-2)*2); addReplyArrayLen(c, (c->argc-2)*2);
...@@ -676,7 +679,7 @@ void channelList(client *c, sds pat, dict *pubsub_channels) { ...@@ -676,7 +679,7 @@ void channelList(client *c, sds pat, dict *pubsub_channels) {
setDeferredArrayLen(c,replylen,mblen); setDeferredArrayLen(c,replylen,mblen);
} }
/* SPUBLISH <channel> <message> */ /* SPUBLISH <shardchannel> <message> */
void spublishCommand(client *c) { void spublishCommand(client *c) {
int receivers = pubsubPublishMessageAndPropagateToCluster(c->argv[1],c->argv[2],1); int receivers = pubsubPublishMessageAndPropagateToCluster(c->argv[1],c->argv[2],1);
if (!server.cluster_enabled) if (!server.cluster_enabled)
...@@ -684,7 +687,7 @@ void spublishCommand(client *c) { ...@@ -684,7 +687,7 @@ void spublishCommand(client *c) {
addReplyLongLong(c,receivers); addReplyLongLong(c,receivers);
} }
/* SSUBSCRIBE channel [channel ...] */ /* SSUBSCRIBE shardchannel [shardchannel ...] */
void ssubscribeCommand(client *c) { void ssubscribeCommand(client *c) {
if (c->flags & CLIENT_DENY_BLOCKING) { if (c->flags & CLIENT_DENY_BLOCKING) {
/* A client that has CLIENT_DENY_BLOCKING flag on /* A client that has CLIENT_DENY_BLOCKING flag on
...@@ -708,7 +711,7 @@ void ssubscribeCommand(client *c) { ...@@ -708,7 +711,7 @@ void ssubscribeCommand(client *c) {
} }
/* SUNSUBSCRIBE [channel ...] */ /* SUNSUBSCRIBE [shardchannel [shardchannel ...]] */
void sunsubscribeCommand(client *c) { void sunsubscribeCommand(client *c) {
if (c->argc == 1) { if (c->argc == 1) {
pubsubUnsubscribeShardAllChannels(c, 1); pubsubUnsubscribeShardAllChannels(c, 1);
......
...@@ -43,7 +43,7 @@ ...@@ -43,7 +43,7 @@
#endif #endif
/* Optimization levels for size-based filling. /* Optimization levels for size-based filling.
* Note that the largest possible limit is 16k, so even if each record takes * Note that the largest possible limit is 64k, so even if each record takes
* just one byte, it still won't overflow the 16 bit count field. */ * just one byte, it still won't overflow the 16 bit count field. */
static const size_t optimization_level[] = {4096, 8192, 16384, 32768, 65536}; static const size_t optimization_level[] = {4096, 8192, 16384, 32768, 65536};
......
...@@ -39,7 +39,7 @@ ...@@ -39,7 +39,7 @@
* We use bit fields keep the quicklistNode at 32 bytes. * We use bit fields keep the quicklistNode at 32 bytes.
* count: 16 bits, max 65536 (max lp bytes is 65k, so max count actually < 32k). * count: 16 bits, max 65536 (max lp bytes is 65k, so max count actually < 32k).
* encoding: 2 bits, RAW=1, LZF=2. * encoding: 2 bits, RAW=1, LZF=2.
* container: 2 bits, PLAIN=1, PACKED=2. * container: 2 bits, PLAIN=1 (a single item as char array), PACKED=2 (listpack with multiple items).
* recompress: 1 bit, bool, true if node is temporary decompressed for usage. * recompress: 1 bit, bool, true if node is temporary decompressed for usage.
* attempted_compress: 1 bit, boolean, used for verifying during testing. * attempted_compress: 1 bit, boolean, used for verifying during testing.
* extra: 10 bits, free for future use; pads out the remainder of 32 bits */ * extra: 10 bits, free for future use; pads out the remainder of 32 bits */
......
...@@ -2746,7 +2746,7 @@ void stopLoading(int success) { ...@@ -2746,7 +2746,7 @@ void stopLoading(int success) {
} }
void startSaving(int rdbflags) { void startSaving(int rdbflags) {
/* Fire the persistence modules end event. */ /* Fire the persistence modules start event. */
int subevent; int subevent;
if (rdbflags & RDBFLAGS_AOF_PREAMBLE && getpid() != server.pid) if (rdbflags & RDBFLAGS_AOF_PREAMBLE && getpid() != server.pid)
subevent = REDISMODULE_SUBEVENT_PERSISTENCE_AOF_START; subevent = REDISMODULE_SUBEVENT_PERSISTENCE_AOF_START;
...@@ -2782,6 +2782,9 @@ void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) { ...@@ -2782,6 +2782,9 @@ void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) {
processEventsWhileBlocked(); processEventsWhileBlocked();
processModuleLoadingProgressEvent(0); processModuleLoadingProgressEvent(0);
} }
if (server.repl_state == REPL_STATE_TRANSFER && rioCheckType(r) == RIO_TYPE_CONN) {
atomicIncr(server.stat_net_repl_input_bytes, len);
}
} }
/* Save the given functions_ctx to the rdb. /* Save the given functions_ctx to the rdb.
...@@ -3476,6 +3479,9 @@ void saveCommand(client *c) { ...@@ -3476,6 +3479,9 @@ void saveCommand(client *c) {
addReplyError(c,"Background save already in progress"); addReplyError(c,"Background save already in progress");
return; return;
} }
server.stat_rdb_saves++;
rdbSaveInfo rsi, *rsiptr; rdbSaveInfo rsi, *rsiptr;
rsiptr = rdbPopulateSaveInfo(&rsi); rsiptr = rdbPopulateSaveInfo(&rsi);
if (rdbSave(SLAVE_REQ_NONE,server.rdb_filename,rsiptr) == C_OK) { if (rdbSave(SLAVE_REQ_NONE,server.rdb_filename,rsiptr) == C_OK) {
......
...@@ -8220,6 +8220,11 @@ static void getKeySizes(redisReply *keys, typeinfo **types, ...@@ -8220,6 +8220,11 @@ static void getKeySizes(redisReply *keys, typeinfo **types,
} }
} }
static void longStatLoopModeStop(int s) {
UNUSED(s);
force_cancel_loop = 1;
}
static void findBigKeys(int memkeys, unsigned memkeys_samples) { static void findBigKeys(int memkeys, unsigned memkeys_samples) {
unsigned long long sampled = 0, total_keys, totlen=0, *sizes=NULL, it=0, scan_loops = 0; unsigned long long sampled = 0, total_keys, totlen=0, *sizes=NULL, it=0, scan_loops = 0;
redisReply *reply, *keys; redisReply *reply, *keys;
...@@ -8237,6 +8242,7 @@ static void findBigKeys(int memkeys, unsigned memkeys_samples) { ...@@ -8237,6 +8242,7 @@ static void findBigKeys(int memkeys, unsigned memkeys_samples) {
typeinfo_add(types_dict, "zset", &type_zset); typeinfo_add(types_dict, "zset", &type_zset);
typeinfo_add(types_dict, "stream", &type_stream); typeinfo_add(types_dict, "stream", &type_stream);
signal(SIGINT, longStatLoopModeStop);
/* Total keys pre scanning */ /* Total keys pre scanning */
total_keys = getDbSize(); total_keys = getDbSize();
...@@ -8315,14 +8321,14 @@ static void findBigKeys(int memkeys, unsigned memkeys_samples) { ...@@ -8315,14 +8321,14 @@ static void findBigKeys(int memkeys, unsigned memkeys_samples) {
} }
freeReplyObject(reply); freeReplyObject(reply);
} while(it != 0); } while(force_cancel_loop == 0 && it != 0);
if(types) zfree(types); if(types) zfree(types);
if(sizes) zfree(sizes); if(sizes) zfree(sizes);
/* We're done */ /* We're done */
printf("\n-------- summary -------\n\n"); printf("\n-------- summary -------\n\n");
if (force_cancel_loop) printf("[%05.2f%%] ", pct);
printf("Sampled %llu keys in the keyspace!\n", sampled); printf("Sampled %llu keys in the keyspace!\n", sampled);
printf("Total key length in bytes is %llu (avg len %.2f)\n\n", printf("Total key length in bytes is %llu (avg len %.2f)\n\n",
totlen, totlen ? (double)totlen/sampled : 0); totlen, totlen ? (double)totlen/sampled : 0);
...@@ -8401,6 +8407,7 @@ static void findHotKeys(void) { ...@@ -8401,6 +8407,7 @@ static void findHotKeys(void) {
unsigned int arrsize = 0, i, k; unsigned int arrsize = 0, i, k;
double pct; double pct;
signal(SIGINT, longStatLoopModeStop);
/* Total keys pre scanning */ /* Total keys pre scanning */
total_keys = getDbSize(); total_keys = getDbSize();
...@@ -8466,13 +8473,13 @@ static void findHotKeys(void) { ...@@ -8466,13 +8473,13 @@ static void findHotKeys(void) {
} }
freeReplyObject(reply); freeReplyObject(reply);
} while(it != 0); } while(force_cancel_loop ==0 && it != 0);
if (freqs) zfree(freqs); if (freqs) zfree(freqs);
/* We're done */ /* We're done */
printf("\n-------- summary -------\n\n"); printf("\n-------- summary -------\n\n");
if(force_cancel_loop)printf("[%05.2f%%] ",pct);
printf("Sampled %llu keys in the keyspace!\n", sampled); printf("Sampled %llu keys in the keyspace!\n", sampled);
for (i=1; i<= HOTKEYS_SAMPLE; i++) { for (i=1; i<= HOTKEYS_SAMPLE; i++) {
...@@ -8643,7 +8650,7 @@ static void statMode(void) { ...@@ -8643,7 +8650,7 @@ static void statMode(void) {
static void scanMode(void) { static void scanMode(void) {
redisReply *reply; redisReply *reply;
unsigned long long cur = 0; unsigned long long cur = 0;
signal(SIGINT, longStatLoopModeStop);
do { do {
reply = sendScan(&cur); reply = sendScan(&cur);
for (unsigned int j = 0; j < reply->element[1]->elements; j++) { for (unsigned int j = 0; j < reply->element[1]->elements; j++) {
...@@ -8658,7 +8665,7 @@ static void scanMode(void) { ...@@ -8658,7 +8665,7 @@ static void scanMode(void) {
} }
freeReplyObject(reply); freeReplyObject(reply);
if (config.interval) usleep(config.interval); if (config.interval) usleep(config.interval);
} while(cur != 0); } while(force_cancel_loop == 0 && cur != 0);
exit(0); exit(0);
} }
...@@ -8785,11 +8792,6 @@ unsigned long compute_something_fast(void) { ...@@ -8785,11 +8792,6 @@ unsigned long compute_something_fast(void) {
return output; return output;
} }
static void intrinsicLatencyModeStop(int s) {
UNUSED(s);
force_cancel_loop = 1;
}
static void sigIntHandler(int s) { static void sigIntHandler(int s) {
UNUSED(s); UNUSED(s);
...@@ -8807,7 +8809,7 @@ static void intrinsicLatencyMode(void) { ...@@ -8807,7 +8809,7 @@ static void intrinsicLatencyMode(void) {
run_time = (long long)config.intrinsic_latency_duration * 1000000; run_time = (long long)config.intrinsic_latency_duration * 1000000;
test_end = ustime() + run_time; test_end = ustime() + run_time;
signal(SIGINT, intrinsicLatencyModeStop); signal(SIGINT, longStatLoopModeStop);
while(1) { while(1) {
long long start, end, latency; long long start, end, latency;
......
...@@ -185,11 +185,12 @@ This flag should not be used directly by the module. ...@@ -185,11 +185,12 @@ This flag should not be used directly by the module.
#define REDISMODULE_NOTIFY_KEY_MISS (1<<11) /* m (Note: This one is excluded from REDISMODULE_NOTIFY_ALL on purpose) */ #define REDISMODULE_NOTIFY_KEY_MISS (1<<11) /* m (Note: This one is excluded from REDISMODULE_NOTIFY_ALL on purpose) */
#define REDISMODULE_NOTIFY_LOADED (1<<12) /* module only key space notification, indicate a key loaded from rdb */ #define REDISMODULE_NOTIFY_LOADED (1<<12) /* module only key space notification, indicate a key loaded from rdb */
#define REDISMODULE_NOTIFY_MODULE (1<<13) /* d, module key space notification */ #define REDISMODULE_NOTIFY_MODULE (1<<13) /* d, module key space notification */
#define REDISMODULE_NOTIFY_NEW (1<<14) /* n, new key notification */
/* Next notification flag, must be updated when adding new flags above! /* Next notification flag, must be updated when adding new flags above!
This flag should not be used directly by the module. This flag should not be used directly by the module.
* Use RedisModule_GetKeyspaceNotificationFlagsAll instead. */ * Use RedisModule_GetKeyspaceNotificationFlagsAll instead. */
#define _REDISMODULE_NOTIFY_NEXT (1<<14) #define _REDISMODULE_NOTIFY_NEXT (1<<15)
#define REDISMODULE_NOTIFY_ALL (REDISMODULE_NOTIFY_GENERIC | REDISMODULE_NOTIFY_STRING | REDISMODULE_NOTIFY_LIST | REDISMODULE_NOTIFY_SET | REDISMODULE_NOTIFY_HASH | REDISMODULE_NOTIFY_ZSET | REDISMODULE_NOTIFY_EXPIRED | REDISMODULE_NOTIFY_EVICTED | REDISMODULE_NOTIFY_STREAM | REDISMODULE_NOTIFY_MODULE) /* A */ #define REDISMODULE_NOTIFY_ALL (REDISMODULE_NOTIFY_GENERIC | REDISMODULE_NOTIFY_STRING | REDISMODULE_NOTIFY_LIST | REDISMODULE_NOTIFY_SET | REDISMODULE_NOTIFY_HASH | REDISMODULE_NOTIFY_ZSET | REDISMODULE_NOTIFY_EXPIRED | REDISMODULE_NOTIFY_EVICTED | REDISMODULE_NOTIFY_STREAM | REDISMODULE_NOTIFY_MODULE) /* A */
...@@ -248,7 +249,7 @@ typedef uint64_t RedisModuleTimerID; ...@@ -248,7 +249,7 @@ typedef uint64_t RedisModuleTimerID;
/* When set, Redis will not call RedisModule_SignalModifiedKey(), implicitly in /* When set, Redis will not call RedisModule_SignalModifiedKey(), implicitly in
* RedisModule_CloseKey, and the module needs to do that when manually when keys * RedisModule_CloseKey, and the module needs to do that when manually when keys
* are modified from the user's sperspective, to invalidate WATCH. */ * are modified from the user's perspective, to invalidate WATCH. */
#define REDISMODULE_OPTION_NO_IMPLICIT_SIGNAL_MODIFIED (1<<1) #define REDISMODULE_OPTION_NO_IMPLICIT_SIGNAL_MODIFIED (1<<1)
/* Declare that the module can handle diskless async replication with RedisModule_SetModuleOptions. */ /* Declare that the module can handle diskless async replication with RedisModule_SetModuleOptions. */
...@@ -346,7 +347,7 @@ typedef struct { ...@@ -346,7 +347,7 @@ typedef struct {
const char *keyword; const char *keyword;
/* An index in argv from which to start searching. /* An index in argv from which to start searching.
* Can be negative, which means start search from the end, in reverse * Can be negative, which means start search from the end, in reverse
* (Example: -2 means to start in reverse from the panultimate arg) */ * (Example: -2 means to start in reverse from the penultimate arg) */
int startfrom; int startfrom;
} keyword; } keyword;
} bs; } bs;
...@@ -1160,6 +1161,7 @@ REDISMODULE_API int (*RedisModule_ExitFromChild)(int retcode) REDISMODULE_ATTR; ...@@ -1160,6 +1161,7 @@ REDISMODULE_API int (*RedisModule_ExitFromChild)(int retcode) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_KillForkChild)(int child_pid) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_KillForkChild)(int child_pid) REDISMODULE_ATTR;
REDISMODULE_API float (*RedisModule_GetUsedMemoryRatio)() REDISMODULE_ATTR; REDISMODULE_API float (*RedisModule_GetUsedMemoryRatio)() REDISMODULE_ATTR;
REDISMODULE_API size_t (*RedisModule_MallocSize)(void* ptr) REDISMODULE_ATTR; REDISMODULE_API size_t (*RedisModule_MallocSize)(void* ptr) REDISMODULE_ATTR;
REDISMODULE_API size_t (*RedisModule_MallocUsableSize)(void *ptr) REDISMODULE_ATTR;
REDISMODULE_API size_t (*RedisModule_MallocSizeString)(RedisModuleString* str) REDISMODULE_ATTR; REDISMODULE_API size_t (*RedisModule_MallocSizeString)(RedisModuleString* str) REDISMODULE_ATTR;
REDISMODULE_API size_t (*RedisModule_MallocSizeDict)(RedisModuleDict* dict) REDISMODULE_ATTR; REDISMODULE_API size_t (*RedisModule_MallocSizeDict)(RedisModuleDict* dict) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleUser * (*RedisModule_CreateModuleUser)(const char *name) REDISMODULE_ATTR; REDISMODULE_API RedisModuleUser * (*RedisModule_CreateModuleUser)(const char *name) REDISMODULE_ATTR;
...@@ -1190,7 +1192,7 @@ REDISMODULE_API const RedisModuleString * (*RedisModule_GetKeyNameFromDefragCtx) ...@@ -1190,7 +1192,7 @@ REDISMODULE_API const RedisModuleString * (*RedisModule_GetKeyNameFromDefragCtx)
REDISMODULE_API int (*RedisModule_EventLoopAdd)(int fd, int mask, RedisModuleEventLoopFunc func, void *user_data) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_EventLoopAdd)(int fd, int mask, RedisModuleEventLoopFunc func, void *user_data) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_EventLoopDel)(int fd, int mask) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_EventLoopDel)(int fd, int mask) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_EventLoopAddOneShot)(RedisModuleEventLoopOneShotFunc func, void *user_data) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_EventLoopAddOneShot)(RedisModuleEventLoopOneShotFunc func, void *user_data) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_RegisterBoolConfig)(RedisModuleCtx *ctx, char *name, int default_val, unsigned int flags, RedisModuleConfigGetBoolFunc getfn, RedisModuleConfigSetBoolFunc setfn, RedisModuleConfigApplyFunc applyfn, void *privdata) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_RegisterBoolConfig)(RedisModuleCtx *ctx, const char *name, int default_val, unsigned int flags, RedisModuleConfigGetBoolFunc getfn, RedisModuleConfigSetBoolFunc setfn, RedisModuleConfigApplyFunc applyfn, void *privdata) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_RegisterNumericConfig)(RedisModuleCtx *ctx, const char *name, long long default_val, unsigned int flags, long long min, long long max, RedisModuleConfigGetNumericFunc getfn, RedisModuleConfigSetNumericFunc setfn, RedisModuleConfigApplyFunc applyfn, void *privdata) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_RegisterNumericConfig)(RedisModuleCtx *ctx, const char *name, long long default_val, unsigned int flags, long long min, long long max, RedisModuleConfigGetNumericFunc getfn, RedisModuleConfigSetNumericFunc setfn, RedisModuleConfigApplyFunc applyfn, void *privdata) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_RegisterStringConfig)(RedisModuleCtx *ctx, const char *name, const char *default_val, unsigned int flags, RedisModuleConfigGetStringFunc getfn, RedisModuleConfigSetStringFunc setfn, RedisModuleConfigApplyFunc applyfn, void *privdata) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_RegisterStringConfig)(RedisModuleCtx *ctx, const char *name, const char *default_val, unsigned int flags, RedisModuleConfigGetStringFunc getfn, RedisModuleConfigSetStringFunc setfn, RedisModuleConfigApplyFunc applyfn, void *privdata) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_RegisterEnumConfig)(RedisModuleCtx *ctx, const char *name, int default_val, unsigned int flags, const char **enum_values, const int *int_values, int num_enum_vals, RedisModuleConfigGetEnumFunc getfn, RedisModuleConfigSetEnumFunc setfn, RedisModuleConfigApplyFunc applyfn, void *privdata) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_RegisterEnumConfig)(RedisModuleCtx *ctx, const char *name, int default_val, unsigned int flags, const char **enum_values, const int *int_values, int num_enum_vals, RedisModuleConfigGetEnumFunc getfn, RedisModuleConfigSetEnumFunc setfn, RedisModuleConfigApplyFunc applyfn, void *privdata) REDISMODULE_ATTR;
...@@ -1493,6 +1495,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -1493,6 +1495,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(KillForkChild); REDISMODULE_GET_API(KillForkChild);
REDISMODULE_GET_API(GetUsedMemoryRatio); REDISMODULE_GET_API(GetUsedMemoryRatio);
REDISMODULE_GET_API(MallocSize); REDISMODULE_GET_API(MallocSize);
REDISMODULE_GET_API(MallocUsableSize);
REDISMODULE_GET_API(MallocSizeString); REDISMODULE_GET_API(MallocSizeString);
REDISMODULE_GET_API(MallocSizeDict); REDISMODULE_GET_API(MallocSizeDict);
REDISMODULE_GET_API(CreateModuleUser); REDISMODULE_GET_API(CreateModuleUser);
......
...@@ -1358,7 +1358,7 @@ void sendBulkToSlave(connection *conn) { ...@@ -1358,7 +1358,7 @@ void sendBulkToSlave(connection *conn) {
freeClient(slave); freeClient(slave);
return; return;
} }
atomicIncr(server.stat_net_output_bytes, nwritten); atomicIncr(server.stat_net_repl_output_bytes, nwritten);
sdsrange(slave->replpreamble,nwritten,-1); sdsrange(slave->replpreamble,nwritten,-1);
if (sdslen(slave->replpreamble) == 0) { if (sdslen(slave->replpreamble) == 0) {
sdsfree(slave->replpreamble); sdsfree(slave->replpreamble);
...@@ -1387,7 +1387,7 @@ void sendBulkToSlave(connection *conn) { ...@@ -1387,7 +1387,7 @@ void sendBulkToSlave(connection *conn) {
return; return;
} }
slave->repldboff += nwritten; slave->repldboff += nwritten;
atomicIncr(server.stat_net_output_bytes, nwritten); atomicIncr(server.stat_net_repl_output_bytes, nwritten);
if (slave->repldboff == slave->repldbsize) { if (slave->repldboff == slave->repldbsize) {
close(slave->repldbfd); close(slave->repldbfd);
slave->repldbfd = -1; slave->repldbfd = -1;
...@@ -1419,7 +1419,7 @@ void rdbPipeWriteHandlerConnRemoved(struct connection *conn) { ...@@ -1419,7 +1419,7 @@ void rdbPipeWriteHandlerConnRemoved(struct connection *conn) {
void rdbPipeWriteHandler(struct connection *conn) { void rdbPipeWriteHandler(struct connection *conn) {
serverAssert(server.rdb_pipe_bufflen>0); serverAssert(server.rdb_pipe_bufflen>0);
client *slave = connGetPrivateData(conn); client *slave = connGetPrivateData(conn);
int nwritten; ssize_t nwritten;
if ((nwritten = connWrite(conn, server.rdb_pipe_buff + slave->repldboff, if ((nwritten = connWrite(conn, server.rdb_pipe_buff + slave->repldboff,
server.rdb_pipe_bufflen - slave->repldboff)) == -1) server.rdb_pipe_bufflen - slave->repldboff)) == -1)
{ {
...@@ -1431,7 +1431,7 @@ void rdbPipeWriteHandler(struct connection *conn) { ...@@ -1431,7 +1431,7 @@ void rdbPipeWriteHandler(struct connection *conn) {
return; return;
} else { } else {
slave->repldboff += nwritten; slave->repldboff += nwritten;
atomicIncr(server.stat_net_output_bytes, nwritten); atomicIncr(server.stat_net_repl_output_bytes, nwritten);
if (slave->repldboff < server.rdb_pipe_bufflen) { if (slave->repldboff < server.rdb_pipe_bufflen) {
slave->repl_last_partial_write = server.unixtime; slave->repl_last_partial_write = server.unixtime;
return; /* more data to write.. */ return; /* more data to write.. */
...@@ -1491,7 +1491,7 @@ void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData, ...@@ -1491,7 +1491,7 @@ void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData,
int stillAlive = 0; int stillAlive = 0;
for (i=0; i < server.rdb_pipe_numconns; i++) for (i=0; i < server.rdb_pipe_numconns; i++)
{ {
int nwritten; ssize_t nwritten;
connection *conn = server.rdb_pipe_conns[i]; connection *conn = server.rdb_pipe_conns[i];
if (!conn) if (!conn)
continue; continue;
...@@ -1511,7 +1511,7 @@ void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData, ...@@ -1511,7 +1511,7 @@ void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData,
/* Note: when use diskless replication, 'repldboff' is the offset /* Note: when use diskless replication, 'repldboff' is the offset
* of 'rdb_pipe_buff' sent rather than the offset of entire RDB. */ * of 'rdb_pipe_buff' sent rather than the offset of entire RDB. */
slave->repldboff = nwritten; slave->repldboff = nwritten;
atomicIncr(server.stat_net_output_bytes, nwritten); atomicIncr(server.stat_net_repl_output_bytes, nwritten);
} }
/* If we were unable to write all the data to one of the replicas, /* If we were unable to write all the data to one of the replicas,
* setup write handler (and disable pipe read handler, below) */ * setup write handler (and disable pipe read handler, below) */
...@@ -1817,11 +1817,16 @@ void readSyncBulkPayload(connection *conn) { ...@@ -1817,11 +1817,16 @@ void readSyncBulkPayload(connection *conn) {
/* If repl_transfer_size == -1 we still have to read the bulk length /* If repl_transfer_size == -1 we still have to read the bulk length
* from the master reply. */ * from the master reply. */
if (server.repl_transfer_size == -1) { if (server.repl_transfer_size == -1) {
if (connSyncReadLine(conn,buf,1024,server.repl_syncio_timeout*1000) == -1) { nread = connSyncReadLine(conn,buf,1024,server.repl_syncio_timeout*1000);
if (nread == -1) {
serverLog(LL_WARNING, serverLog(LL_WARNING,
"I/O error reading bulk count from MASTER: %s", "I/O error reading bulk count from MASTER: %s",
strerror(errno)); strerror(errno));
goto error; goto error;
} else {
/* nread here is returned by connSyncReadLine(), which calls syncReadLine() and
* convert "\r\n" to '\0' so 1 byte is lost. */
atomicIncr(server.stat_net_repl_input_bytes, nread+1);
} }
if (buf[0] == '-') { if (buf[0] == '-') {
...@@ -1892,7 +1897,7 @@ void readSyncBulkPayload(connection *conn) { ...@@ -1892,7 +1897,7 @@ void readSyncBulkPayload(connection *conn) {
cancelReplicationHandshake(1); cancelReplicationHandshake(1);
return; return;
} }
atomicIncr(server.stat_net_input_bytes, nread); atomicIncr(server.stat_net_repl_input_bytes, nread);
/* When a mark is used, we want to detect EOF asap in order to avoid /* When a mark is used, we want to detect EOF asap in order to avoid
* writing the EOF mark into the file... */ * writing the EOF mark into the file... */
...@@ -2223,8 +2228,8 @@ char *receiveSynchronousResponse(connection *conn) { ...@@ -2223,8 +2228,8 @@ char *receiveSynchronousResponse(connection *conn) {
/* Read the reply from the server. */ /* Read the reply from the server. */
if (connSyncReadLine(conn,buf,sizeof(buf),server.repl_syncio_timeout*1000) == -1) if (connSyncReadLine(conn,buf,sizeof(buf),server.repl_syncio_timeout*1000) == -1)
{ {
return sdscatprintf(sdsempty(),"-Reading from master: %s", serverLog(LL_WARNING, "Failed to read response from the server: %s", strerror(errno));
strerror(errno)); return NULL;
} }
server.repl_transfer_lastio = server.unixtime; server.repl_transfer_lastio = server.unixtime;
return sdsnew(buf); return sdsnew(buf);
...@@ -2407,6 +2412,13 @@ int slaveTryPartialResynchronization(connection *conn, int read_reply) { ...@@ -2407,6 +2412,13 @@ int slaveTryPartialResynchronization(connection *conn, int read_reply) {
/* Reading half */ /* Reading half */
reply = receiveSynchronousResponse(conn); reply = receiveSynchronousResponse(conn);
/* Master did not reply to PSYNC */
if (reply == NULL) {
connSetReadHandler(conn, NULL);
serverLog(LL_WARNING, "Master did not reply to PSYNC, will try later");
return PSYNC_TRY_LATER;
}
if (sdslen(reply) == 0) { if (sdslen(reply) == 0) {
/* The master may send empty newlines after it receives PSYNC /* The master may send empty newlines after it receives PSYNC
* and before to reply, just to keep the connection alive. */ * and before to reply, just to keep the connection alive. */
...@@ -2566,6 +2578,9 @@ void syncWithMaster(connection *conn) { ...@@ -2566,6 +2578,9 @@ void syncWithMaster(connection *conn) {
if (server.repl_state == REPL_STATE_RECEIVE_PING_REPLY) { if (server.repl_state == REPL_STATE_RECEIVE_PING_REPLY) {
err = receiveSynchronousResponse(conn); err = receiveSynchronousResponse(conn);
/* The master did not reply */
if (err == NULL) goto no_response_error;
/* We accept only two replies as valid, a positive +PONG reply /* We accept only two replies as valid, a positive +PONG reply
* (we just check for "+") or an authentication error. * (we just check for "+") or an authentication error.
* Note that older versions of Redis replied with "operation not * Note that older versions of Redis replied with "operation not
...@@ -2652,6 +2667,7 @@ void syncWithMaster(connection *conn) { ...@@ -2652,6 +2667,7 @@ void syncWithMaster(connection *conn) {
/* Receive AUTH reply. */ /* Receive AUTH reply. */
if (server.repl_state == REPL_STATE_RECEIVE_AUTH_REPLY) { if (server.repl_state == REPL_STATE_RECEIVE_AUTH_REPLY) {
err = receiveSynchronousResponse(conn); err = receiveSynchronousResponse(conn);
if (err == NULL) goto no_response_error;
if (err[0] == '-') { if (err[0] == '-') {
serverLog(LL_WARNING,"Unable to AUTH to MASTER: %s",err); serverLog(LL_WARNING,"Unable to AUTH to MASTER: %s",err);
sdsfree(err); sdsfree(err);
...@@ -2666,6 +2682,7 @@ void syncWithMaster(connection *conn) { ...@@ -2666,6 +2682,7 @@ void syncWithMaster(connection *conn) {
/* Receive REPLCONF listening-port reply. */ /* Receive REPLCONF listening-port reply. */
if (server.repl_state == REPL_STATE_RECEIVE_PORT_REPLY) { if (server.repl_state == REPL_STATE_RECEIVE_PORT_REPLY) {
err = receiveSynchronousResponse(conn); err = receiveSynchronousResponse(conn);
if (err == NULL) goto no_response_error;
/* Ignore the error if any, not all the Redis versions support /* Ignore the error if any, not all the Redis versions support
* REPLCONF listening-port. */ * REPLCONF listening-port. */
if (err[0] == '-') { if (err[0] == '-') {
...@@ -2683,6 +2700,7 @@ void syncWithMaster(connection *conn) { ...@@ -2683,6 +2700,7 @@ void syncWithMaster(connection *conn) {
/* Receive REPLCONF ip-address reply. */ /* Receive REPLCONF ip-address reply. */
if (server.repl_state == REPL_STATE_RECEIVE_IP_REPLY) { if (server.repl_state == REPL_STATE_RECEIVE_IP_REPLY) {
err = receiveSynchronousResponse(conn); err = receiveSynchronousResponse(conn);
if (err == NULL) goto no_response_error;
/* Ignore the error if any, not all the Redis versions support /* Ignore the error if any, not all the Redis versions support
* REPLCONF ip-address. */ * REPLCONF ip-address. */
if (err[0] == '-') { if (err[0] == '-') {
...@@ -2697,6 +2715,7 @@ void syncWithMaster(connection *conn) { ...@@ -2697,6 +2715,7 @@ void syncWithMaster(connection *conn) {
/* Receive CAPA reply. */ /* Receive CAPA reply. */
if (server.repl_state == REPL_STATE_RECEIVE_CAPA_REPLY) { if (server.repl_state == REPL_STATE_RECEIVE_CAPA_REPLY) {
err = receiveSynchronousResponse(conn); err = receiveSynchronousResponse(conn);
if (err == NULL) goto no_response_error;
/* Ignore the error if any, not all the Redis versions support /* Ignore the error if any, not all the Redis versions support
* REPLCONF capa. */ * REPLCONF capa. */
if (err[0] == '-') { if (err[0] == '-') {
...@@ -2810,6 +2829,10 @@ void syncWithMaster(connection *conn) { ...@@ -2810,6 +2829,10 @@ void syncWithMaster(connection *conn) {
server.repl_transfer_lastio = server.unixtime; server.repl_transfer_lastio = server.unixtime;
return; return;
no_response_error: /* Handle receiveSynchronousResponse() error when master has no reply */
serverLog(LL_WARNING, "Master did not respond to command during SYNC handshake");
/* Fall through to regular error handling */
error: error:
if (dfd != -1) close(dfd); if (dfd != -1) close(dfd);
connClose(conn); connClose(conn);
......
...@@ -438,6 +438,20 @@ void rioSetAutoSync(rio *r, off_t bytes) { ...@@ -438,6 +438,20 @@ void rioSetAutoSync(rio *r, off_t bytes) {
r->io.file.autosync = bytes; r->io.file.autosync = bytes;
} }
/* Check the type of rio. */
uint8_t rioCheckType(rio *r) {
if (r->read == rioFileRead) {
return RIO_TYPE_FILE;
} else if (r->read == rioBufferRead) {
return RIO_TYPE_BUFFER;
} else if (r->read == rioConnRead) {
return RIO_TYPE_CONN;
} else {
/* r->read == rioFdRead */
return RIO_TYPE_FD;
}
}
/* --------------------------- Higher level interface -------------------------- /* --------------------------- Higher level interface --------------------------
* *
* The following higher level functions use lower level rio.c functions to help * The following higher level functions use lower level rio.c functions to help
......
...@@ -40,6 +40,11 @@ ...@@ -40,6 +40,11 @@
#define RIO_FLAG_READ_ERROR (1<<0) #define RIO_FLAG_READ_ERROR (1<<0)
#define RIO_FLAG_WRITE_ERROR (1<<1) #define RIO_FLAG_WRITE_ERROR (1<<1)
#define RIO_TYPE_FILE (1<<0)
#define RIO_TYPE_BUFFER (1<<1)
#define RIO_TYPE_CONN (1<<2)
#define RIO_TYPE_FD (1<<3)
struct _rio { struct _rio {
/* Backend functions. /* Backend functions.
* Since this functions do not tolerate short writes or reads the return * Since this functions do not tolerate short writes or reads the return
...@@ -174,5 +179,5 @@ int rioWriteBulkObject(rio *r, struct redisObject *obj); ...@@ -174,5 +179,5 @@ int rioWriteBulkObject(rio *r, struct redisObject *obj);
void rioGenericUpdateChecksum(rio *r, const void *buf, size_t len); void rioGenericUpdateChecksum(rio *r, const void *buf, size_t len);
void rioSetAutoSync(rio *r, off_t bytes); void rioSetAutoSync(rio *r, off_t bytes);
uint8_t rioCheckType(rio *r);
#endif #endif
...@@ -92,8 +92,8 @@ int scriptInterrupt(scriptRunCtx *run_ctx) { ...@@ -92,8 +92,8 @@ int scriptInterrupt(scriptRunCtx *run_ctx) {
serverLog(LL_WARNING, serverLog(LL_WARNING,
"Slow script detected: still in execution after %lld milliseconds. " "Slow script detected: still in execution after %lld milliseconds. "
"You can try killing the script using the %s command.", "You can try killing the script using the %s command. Script name is: %s.",
elapsed, (run_ctx->flags & SCRIPT_EVAL_MODE) ? "SCRIPT KILL" : "FUNCTION KILL"); elapsed, (run_ctx->flags & SCRIPT_EVAL_MODE) ? "SCRIPT KILL" : "FUNCTION KILL", run_ctx->funcname);
enterScriptTimedoutMode(run_ctx); enterScriptTimedoutMode(run_ctx);
/* Once the script timeouts we reenter the event loop to permit others /* Once the script timeouts we reenter the event loop to permit others
...@@ -108,6 +108,25 @@ int scriptInterrupt(scriptRunCtx *run_ctx) { ...@@ -108,6 +108,25 @@ int scriptInterrupt(scriptRunCtx *run_ctx) {
return (run_ctx->flags & SCRIPT_KILLED) ? SCRIPT_KILL : SCRIPT_CONTINUE; return (run_ctx->flags & SCRIPT_KILLED) ? SCRIPT_KILL : SCRIPT_CONTINUE;
} }
uint64_t scriptFlagsToCmdFlags(uint64_t cmd_flags, uint64_t script_flags) {
/* If the script declared flags, clear the ones from the command and use the ones it declared.*/
cmd_flags &= ~(CMD_STALE | CMD_DENYOOM | CMD_WRITE);
/* NO_WRITES implies ALLOW_OOM */
if (!(script_flags & (SCRIPT_FLAG_ALLOW_OOM | SCRIPT_FLAG_NO_WRITES)))
cmd_flags |= CMD_DENYOOM;
if (!(script_flags & SCRIPT_FLAG_NO_WRITES))
cmd_flags |= CMD_WRITE;
if (script_flags & SCRIPT_FLAG_ALLOW_STALE)
cmd_flags |= CMD_STALE;
/* In addition the MAY_REPLICATE flag is set for these commands, but
* if we have flags we know if it's gonna do any writes or not. */
cmd_flags &= ~CMD_MAY_REPLICATE;
return cmd_flags;
}
/* Prepare the given run ctx for execution */ /* Prepare the given run ctx for execution */
int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *caller, const char *funcname, uint64_t script_flags, int ro) { int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *caller, const char *funcname, uint64_t script_flags, int ro) {
serverAssert(!curr_run_ctx); serverAssert(!curr_run_ctx);
...@@ -123,12 +142,6 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca ...@@ -123,12 +142,6 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca
return C_ERR; return C_ERR;
} }
if (!(script_flags & SCRIPT_FLAG_ALLOW_OOM) && server.script_oom && server.maxmemory) {
addReplyError(caller, "-OOM allow-oom flag is not set on the script, "
"can not run it when used memory > 'maxmemory'");
return C_ERR;
}
if (running_stale && !(script_flags & SCRIPT_FLAG_ALLOW_STALE)) { if (running_stale && !(script_flags & SCRIPT_FLAG_ALLOW_STALE)) {
addReplyError(caller, "-MASTERDOWN Link with MASTER is down, " addReplyError(caller, "-MASTERDOWN Link with MASTER is down, "
"replica-serve-stale-data is set to 'no' " "replica-serve-stale-data is set to 'no' "
...@@ -142,7 +155,7 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca ...@@ -142,7 +155,7 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca
* 2. no disk error detected * 2. no disk error detected
* 3. command is not `fcall_ro`/`eval[sha]_ro` */ * 3. command is not `fcall_ro`/`eval[sha]_ro` */
if (server.masterhost && server.repl_slave_ro && !obey_client) { if (server.masterhost && server.repl_slave_ro && !obey_client) {
addReplyError(caller, "Can not run script with write flag on readonly replica"); addReplyError(caller, "-READONLY Can not run script with write flag on readonly replica");
return C_ERR; return C_ERR;
} }
...@@ -177,6 +190,17 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca ...@@ -177,6 +190,17 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca
return C_ERR; return C_ERR;
} }
} }
/* Check OOM state. the no-writes flag imply allow-oom. we tested it
* after the no-write error, so no need to mention it in the error reply. */
if (server.pre_command_oom_state && server.maxmemory &&
!(script_flags & (SCRIPT_FLAG_ALLOW_OOM|SCRIPT_FLAG_NO_WRITES)))
{
addReplyError(caller, "-OOM allow-oom flag is not set on the script, "
"can not run it when used memory > 'maxmemory'");
return C_ERR;
}
} else { } else {
/* Special handling for backwards compatibility (no shebang eval[sha]) mode */ /* Special handling for backwards compatibility (no shebang eval[sha]) mode */
if (running_stale) { if (running_stale) {
...@@ -202,8 +226,6 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca ...@@ -202,8 +226,6 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca
script_client->flags |= CLIENT_MULTI; script_client->flags |= CLIENT_MULTI;
} }
server.in_script = 1;
run_ctx->start_time = getMonotonicUs(); run_ctx->start_time = getMonotonicUs();
run_ctx->snapshot_time = mstime(); run_ctx->snapshot_time = mstime();
...@@ -216,6 +238,8 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca ...@@ -216,6 +238,8 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca
run_ctx->flags |= SCRIPT_READ_ONLY; run_ctx->flags |= SCRIPT_READ_ONLY;
} }
if (!(script_flags & SCRIPT_FLAG_EVAL_COMPAT_MODE) && (script_flags & SCRIPT_FLAG_ALLOW_OOM)) { if (!(script_flags & SCRIPT_FLAG_EVAL_COMPAT_MODE) && (script_flags & SCRIPT_FLAG_ALLOW_OOM)) {
/* Note: we don't need to test the no-writes flag here and set this run_ctx flag,
* since only write commands can are deny-oom. */
run_ctx->flags |= SCRIPT_ALLOW_OOM; run_ctx->flags |= SCRIPT_ALLOW_OOM;
} }
...@@ -236,7 +260,6 @@ void scriptResetRun(scriptRunCtx *run_ctx) { ...@@ -236,7 +260,6 @@ void scriptResetRun(scriptRunCtx *run_ctx) {
/* After the script done, remove the MULTI state. */ /* After the script done, remove the MULTI state. */
run_ctx->c->flags &= ~CLIENT_MULTI; run_ctx->c->flags &= ~CLIENT_MULTI;
server.in_script = 0;
server.script_caller = NULL; server.script_caller = NULL;
if (scriptIsTimedout()) { if (scriptIsTimedout()) {
...@@ -323,16 +346,23 @@ static int scriptVerifyACL(client *c, sds *err) { ...@@ -323,16 +346,23 @@ static int scriptVerifyACL(client *c, sds *err) {
} }
static int scriptVerifyWriteCommandAllow(scriptRunCtx *run_ctx, char **err) { static int scriptVerifyWriteCommandAllow(scriptRunCtx *run_ctx, char **err) {
if (!(run_ctx->c->cmd->flags & CMD_WRITE)) {
return C_OK;
}
if (run_ctx->flags & SCRIPT_READ_ONLY) { /* A write command, on an RO command or an RO script is rejected ASAP.
/* We know its a write command, on a read only run we do not allow it. */ * Note: For scripts, we consider may-replicate commands as write commands.
* This also makes it possible to allow read-only scripts to be run during
* CLIENT PAUSE WRITE. */
if (run_ctx->flags & SCRIPT_READ_ONLY &&
(run_ctx->c->cmd->flags & (CMD_WRITE|CMD_MAY_REPLICATE)))
{
*err = sdsnew("Write commands are not allowed from read-only scripts."); *err = sdsnew("Write commands are not allowed from read-only scripts.");
return C_ERR; return C_ERR;
} }
/* The other checks below are on the server state and are only relevant for
* write commands, return if this is not a write command. */
if (!(run_ctx->c->cmd->flags & CMD_WRITE))
return C_OK;
/* Write commands are forbidden against read-only slaves, or if a /* Write commands are forbidden against read-only slaves, or if a
* command marked as non-deterministic was already called in the context * command marked as non-deterministic was already called in the context
* of this script. */ * of this script. */
...@@ -362,16 +392,6 @@ static int scriptVerifyWriteCommandAllow(scriptRunCtx *run_ctx, char **err) { ...@@ -362,16 +392,6 @@ static int scriptVerifyWriteCommandAllow(scriptRunCtx *run_ctx, char **err) {
return C_OK; return C_OK;
} }
static int scriptVerifyMayReplicate(scriptRunCtx *run_ctx, char **err) {
if (run_ctx->c->cmd->flags & CMD_MAY_REPLICATE &&
server.client_pause_type == CLIENT_PAUSE_WRITE) {
*err = sdsnew("May-replicate commands are not allowed when client pause write.");
return C_ERR;
}
return C_OK;
}
static int scriptVerifyOOM(scriptRunCtx *run_ctx, char **err) { static int scriptVerifyOOM(scriptRunCtx *run_ctx, char **err) {
if (run_ctx->flags & SCRIPT_ALLOW_OOM) { if (run_ctx->flags & SCRIPT_ALLOW_OOM) {
/* Allow running any command even if OOM reached */ /* Allow running any command even if OOM reached */
...@@ -385,8 +405,8 @@ static int scriptVerifyOOM(scriptRunCtx *run_ctx, char **err) { ...@@ -385,8 +405,8 @@ static int scriptVerifyOOM(scriptRunCtx *run_ctx, char **err) {
if (server.maxmemory && /* Maxmemory is actually enabled. */ if (server.maxmemory && /* Maxmemory is actually enabled. */
!mustObeyClient(run_ctx->original_client) && /* Don't care about mem for replicas or AOF. */ !mustObeyClient(run_ctx->original_client) && /* Don't care about mem for replicas or AOF. */
!(run_ctx->flags & SCRIPT_WRITE_DIRTY) && /* Script had no side effects so far. */ !(run_ctx->flags & SCRIPT_WRITE_DIRTY) && /* Script had no side effects so far. */
server.script_oom && /* Detected OOM when script start. */ server.pre_command_oom_state && /* Detected OOM when script start. */
(run_ctx->c->cmd->flags & CMD_DENYOOM)) (run_ctx->c->cmd->flags & CMD_DENYOOM))
{ {
*err = sdsdup(shared.oomerr->ptr); *err = sdsdup(shared.oomerr->ptr);
...@@ -526,10 +546,6 @@ void scriptCall(scriptRunCtx *run_ctx, robj* *argv, int argc, sds *err) { ...@@ -526,10 +546,6 @@ void scriptCall(scriptRunCtx *run_ctx, robj* *argv, int argc, sds *err) {
goto error; goto error;
} }
if (scriptVerifyMayReplicate(run_ctx, err) != C_OK) {
goto error;
}
if (scriptVerifyOOM(run_ctx, err) != C_OK) { if (scriptVerifyOOM(run_ctx, err) != C_OK) {
goto error; goto error;
} }
......
...@@ -93,6 +93,7 @@ typedef struct scriptFlag { ...@@ -93,6 +93,7 @@ typedef struct scriptFlag {
extern scriptFlag scripts_flags_def[]; extern scriptFlag scripts_flags_def[];
uint64_t scriptFlagsToCmdFlags(uint64_t cmd_flags, uint64_t script_flags);
int scriptPrepareForRun(scriptRunCtx *r_ctx, client *engine_client, client *caller, const char *funcname, uint64_t script_flags, int ro); int scriptPrepareForRun(scriptRunCtx *r_ctx, client *engine_client, client *caller, const char *funcname, uint64_t script_flags, int ro);
void scriptResetRun(scriptRunCtx *r_ctx); void scriptResetRun(scriptRunCtx *r_ctx);
int scriptSetResp(scriptRunCtx *r_ctx, int resp); int scriptSetResp(scriptRunCtx *r_ctx, int resp);
......
...@@ -1651,8 +1651,11 @@ void luaCallFunction(scriptRunCtx* run_ctx, lua_State *lua, robj** keys, size_t ...@@ -1651,8 +1651,11 @@ void luaCallFunction(scriptRunCtx* run_ctx, lua_State *lua, robj** keys, size_t
* {err='<error msg>', source='<source file>', line=<line>} * {err='<error msg>', source='<source file>', line=<line>}
* We can construct the error message from this information */ * We can construct the error message from this information */
if (!lua_istable(lua, -1)) { if (!lua_istable(lua, -1)) {
/* Should not happened, and we should considered assert it */ const char *msg = "execution failure";
addReplyErrorFormat(c,"Error running script (call to %s)\n", run_ctx->funcname); if (lua_isstring(lua, -1)) {
msg = lua_tostring(lua, -1);
}
addReplyErrorFormat(c,"Error running script %s, %.100s\n", run_ctx->funcname, msg);
} else { } else {
errorInfo err_info = {0}; errorInfo err_info = {0};
sds final_msg = sdsempty(); sds final_msg = sdsempty();
......
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