Unverified Commit 2667c412 authored by Oran Agra's avatar Oran Agra Committed by GitHub
Browse files

Merge pull request #10829 from oranagra/release-7.0.1

Release 7.0.1
parents d375595d 595e725d
......@@ -293,7 +293,7 @@ static void libraryUnlink(functionsLibCtx *lib_ctx, functionLibInfo* li) {
entry = dictUnlink(lib_ctx->libraries, li->name);
dictSetVal(lib_ctx->libraries, entry, NULL);
dictFreeUnlinkedEntry(lib_ctx->libraries, entry);
lib_ctx->cache_memory += libraryMallocSize(li);
lib_ctx->cache_memory -= libraryMallocSize(li);
/* update stats */
functionsLibEngineStats *stats = dictFetchValue(lib_ctx->engines_stats, li->ei->name);
......@@ -325,7 +325,7 @@ static void libraryLink(functionsLibCtx *lib_ctx, functionLibInfo* li) {
/* Takes all libraries from lib_ctx_src and add to lib_ctx_dst.
* On collision, if 'replace' argument is true, replace the existing library with the new one.
* Otherwise abort and leave 'lib_ctx_dst' and 'lib_ctx_src' untouched.
* Return C_OK on success and C_ERR if aborted. If C_ERR is retunred, set a relevant
* Return C_OK on success and C_ERR if aborted. If C_ERR is returned, set a relevant
* error message on the 'err' out parameter.
* */
static int libraryJoin(functionsLibCtx *functions_lib_ctx_dst, functionsLibCtx *functions_lib_ctx_src, int replace, sds *err) {
......@@ -599,10 +599,22 @@ void functionDeleteCommand(client *c) {
addReply(c, shared.ok);
}
/* FUNCTION KILL */
void functionKillCommand(client *c) {
scriptKill(c, 0);
}
/* Try to extract command flags if we can, returns the modified flags.
* Note that it does not guarantee the command arguments are right. */
uint64_t fcallGetCommandFlags(client *c, uint64_t cmd_flags) {
robj *function_name = c->argv[1];
functionInfo *fi = dictFetchValue(curr_functions_lib_ctx->functions, function_name->ptr);
if (!fi)
return cmd_flags;
uint64_t script_flags = fi->f_flags;
return scriptFlagsToCmdFlags(cmd_flags, script_flags);
}
static void fcallCommandGeneric(client *c, int ro) {
robj *function_name = c->argv[1];
functionInfo *fi = dictFetchValue(curr_functions_lib_ctx->functions, function_name->ptr);
......@@ -778,6 +790,7 @@ load_error:
}
}
/* FUNCTION FLUSH [ASYNC | SYNC] */
void functionFlushCommand(client *c) {
if (c->argc > 3) {
addReplySubcommandSyntaxError(c);
......@@ -803,6 +816,7 @@ void functionFlushCommand(client *c) {
addReply(c,shared.ok);
}
/* FUNCTION HELP */
void functionHelpCommand(client *c) {
const char *help[] = {
"LOAD <ENGINE NAME> <LIBRARY NAME> [REPLACE] [DESCRIPTION <LIBRARY DESCRIPTION>] <LIBRARY CODE>",
......@@ -960,6 +974,7 @@ sds functionsCreateWithLibraryCtx(sds code, int replace, sds* err, functionsLibC
old_li = dictFetchValue(lib_ctx->libraries, md.name);
if (old_li && !replace) {
old_li = NULL;
*err = sdscatfmt(sdsempty(), "Library '%S' already exists", md.name);
goto error;
}
......
/* 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
#define __REDIS_HELP_H
......@@ -135,7 +135,7 @@ struct commandHelp {
15,
"3.2.0" },
{ "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",
15,
"6.2.0" },
......@@ -255,7 +255,7 @@ struct commandHelp {
8,
"2.6.9" },
{ "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",
8,
"6.0.0" },
......@@ -1025,7 +1025,7 @@ struct commandHelp {
9,
"4.0.0" },
{ "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",
9,
"7.0.0" },
......@@ -1120,7 +1120,7 @@ struct commandHelp {
11,
"2.8.9" },
{ "PFDEBUG",
"",
"subcommand key",
"Internal commands for debugging HyperLogLog values",
11,
"2.8.9" },
......@@ -1195,7 +1195,7 @@ struct commandHelp {
6,
"7.0.0" },
{ "PUBSUB SHARDNUMSUB",
"[channel [channel ...]]",
"[shardchannel [shardchannel ...]]",
"Get the count of subscribers for shard channels",
6,
"7.0.0" },
......@@ -1255,7 +1255,7 @@ struct commandHelp {
0,
"2.6.0" },
{ "RESTORE-ASKING",
"",
"key ttl serialized-value [REPLACE] [ABSTTL] [IDLETIME seconds] [FREQ frequency]",
"An internal command for migrating keys in a cluster",
9,
"3.0.0" },
......@@ -1450,12 +1450,12 @@ struct commandHelp {
3,
"1.0.0" },
{ "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",
0,
"1.0.0" },
{ "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.",
0,
"7.0.0" },
......@@ -1465,7 +1465,7 @@ struct commandHelp {
3,
"1.0.0" },
{ "SPUBLISH",
"channel message",
"shardchannel message",
"Post a message to a shard channel",
6,
"7.0.0" },
......@@ -1485,7 +1485,7 @@ struct commandHelp {
3,
"2.8.0" },
{ "SSUBSCRIBE",
"channel [channel ...]",
"shardchannel [shardchannel ...]",
"Listen for messages published to the given shard channels",
6,
"7.0.0" },
......@@ -1515,7 +1515,7 @@ struct commandHelp {
3,
"1.0.0" },
{ "SUNSUBSCRIBE",
"[channel [channel ...]]",
"[shardchannel [shardchannel ...]]",
"Stop listening for messages posted to the given shard channels",
6,
"7.0.0" },
......
......@@ -1495,8 +1495,13 @@ cleanup:
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) {
char *cmd = c->argv[1]->ptr;
struct hllhdr *hdr;
......
......@@ -59,39 +59,6 @@ dictType latencyTimeSeriesDictType = {
/* ------------------------- 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
* value of the function is non-zero, the process is being targeted by
* THP support, and is likely to have memory usage / latency issues. */
......
......@@ -63,8 +63,6 @@ struct latencyStats {
void latencyMonitorInit(void);
void latencyAddSample(const char *event, mstime_t latency);
int THPIsEnabled(void);
int THPDisable(void);
/* Latency monitoring macros. */
......
This diff is collapsed.
......@@ -56,7 +56,7 @@ void freeClientMultiState(client *c) {
}
/* Add a new command into the MULTI commands queue */
void queueMultiCommand(client *c) {
void queueMultiCommand(client *c, uint64_t cmd_flags) {
multiCmd *mc;
/* No sense to waste memory if the transaction is already aborted.
......@@ -75,8 +75,8 @@ void queueMultiCommand(client *c) {
mc->argv_len = c->argv_len;
c->mstate.count++;
c->mstate.cmd_flags |= c->cmd->flags;
c->mstate.cmd_inv_flags |= ~c->cmd->flags;
c->mstate.cmd_flags |= cmd_flags;
c->mstate.cmd_inv_flags |= ~cmd_flags;
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
......
......@@ -1953,7 +1953,13 @@ int writeToClient(client *c, int handler_installed) {
zmalloc_used_memory() < server.maxmemory) &&
!(c->flags & CLIENT_SLAVE)) break;
}
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 (connGetState(c->conn) != CONN_STATE_CONNECTED) {
serverLog(LL_VERBOSE,
......@@ -2499,7 +2505,7 @@ int processInputBuffer(client *c) {
* condition on the slave. We want just to accumulate the replication
* stream (instead of replying -BUSY like we do with other clients) and
* 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
* written to the client. Make sure to not let the reply grow after
......@@ -2655,8 +2661,13 @@ void readQueryFromClient(connection *conn) {
if (c->querybuf_peak < qblen) c->querybuf_peak = qblen;
c->lastinteraction = server.unixtime;
if (c->flags & CLIENT_MASTER) c->read_reploff += nread;
if (c->flags & CLIENT_MASTER) {
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) {
sds ci = catClientInfoString(sdsempty(),c), bytes = sdsempty();
......@@ -3557,7 +3568,6 @@ void replaceClientCommandVector(client *c, int argc, robj **argv) {
int j;
retainOriginalCommandVector(c);
freeClientArgv(c);
zfree(c->argv);
c->argv = argv;
c->argc = argc;
c->argv_len_sum = 0;
......
......@@ -930,7 +930,6 @@ char *strEncoding(int encoding) {
case OBJ_ENCODING_INT: return "int";
case OBJ_ENCODING_HT: return "hashtable";
case OBJ_ENCODING_QUICKLIST: return "quicklist";
case OBJ_ENCODING_ZIPLIST: return "ziplist";
case OBJ_ENCODING_LISTPACK: return "listpack";
case OBJ_ENCODING_INTSET: return "intset";
case OBJ_ENCODING_SKIPLIST: return "skiplist";
......@@ -998,8 +997,6 @@ size_t objectComputeSize(robj *key, robj *o, size_t sample_size, int dbid) {
samples++;
} while ((node = node->next) && samples < sample_size);
asize += (double)elesize/samples*ql->len;
} else if (o->encoding == OBJ_ENCODING_ZIPLIST) {
asize = sizeof(*o)+zmalloc_size(o->ptr);
} else {
serverPanic("Unknown list encoding");
}
......
......@@ -39,6 +39,7 @@ typedef struct pubsubtype {
dict **serverPubSubChannels;
robj **subscribeMsg;
robj **unsubscribeMsg;
robj **messageBulk;
}pubsubtype;
/*
......@@ -78,6 +79,7 @@ pubsubtype pubSubType = {
.serverPubSubChannels = &server.pubsub_channels,
.subscribeMsg = &shared.subscribebulk,
.unsubscribeMsg = &shared.unsubscribebulk,
.messageBulk = &shared.messagebulk,
};
/*
......@@ -89,7 +91,8 @@ pubsubtype pubSubShardType = {
.subscriptionCount = clientShardSubscriptionsCount,
.serverPubSubChannels = &server.pubsubshard_channels,
.subscribeMsg = &shared.ssubscribebulk,
.unsubscribeMsg = &shared.sunsubscribebulk
.unsubscribeMsg = &shared.sunsubscribebulk,
.messageBulk = &shared.smessagebulk,
};
/*-----------------------------------------------------------------------------
......@@ -101,12 +104,12 @@ pubsubtype pubSubShardType = {
* 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
* 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)
addReply(c,shared.mbulkhdr[3]);
else
addReplyPushLen(c,3);
addReply(c,shared.messagebulk);
addReply(c,message_bulk);
addReplyBulk(c,channel);
if (msg) addReplyBulk(c,msg);
}
......@@ -461,7 +464,7 @@ int pubsubPublishMessageInternal(robj *channel, robj *message, pubsubtype type)
listRewind(list,&li);
while ((ln = listNext(&li)) != NULL) {
client *c = ln->value;
addReplyPubsubMessage(c,channel,message);
addReplyPubsubMessage(c,channel,message,*type.messageBulk);
updateClientMemUsage(c);
receivers++;
}
......@@ -604,10 +607,10 @@ void pubsubCommand(client *c) {
" Return number of subscriptions to patterns.",
"NUMSUB [<channel> ...]",
" Return the number of subscribers for the specified channels, excluding",
" pattern subscriptions(default: no channels)."
" pattern subscriptions(default: no channels).",
"SHARDCHANNELS [<pattern>]",
" 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)",
NULL
};
......@@ -639,7 +642,7 @@ NULL
sds pat = (c->argc == 2) ? NULL : c->argv[2]->ptr;
channelList(c,pat,server.pubsubshard_channels);
} 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;
addReplyArrayLen(c, (c->argc-2)*2);
......@@ -676,7 +679,7 @@ void channelList(client *c, sds pat, dict *pubsub_channels) {
setDeferredArrayLen(c,replylen,mblen);
}
/* SPUBLISH <channel> <message> */
/* SPUBLISH <shardchannel> <message> */
void spublishCommand(client *c) {
int receivers = pubsubPublishMessageAndPropagateToCluster(c->argv[1],c->argv[2],1);
if (!server.cluster_enabled)
......@@ -684,7 +687,7 @@ void spublishCommand(client *c) {
addReplyLongLong(c,receivers);
}
/* SSUBSCRIBE channel [channel ...] */
/* SSUBSCRIBE shardchannel [shardchannel ...] */
void ssubscribeCommand(client *c) {
if (c->flags & CLIENT_DENY_BLOCKING) {
/* A client that has CLIENT_DENY_BLOCKING flag on
......@@ -708,7 +711,7 @@ void ssubscribeCommand(client *c) {
}
/* SUNSUBSCRIBE [channel ...] */
/* SUNSUBSCRIBE [shardchannel [shardchannel ...]] */
void sunsubscribeCommand(client *c) {
if (c->argc == 1) {
pubsubUnsubscribeShardAllChannels(c, 1);
......
......@@ -43,7 +43,7 @@
#endif
/* 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. */
static const size_t optimization_level[] = {4096, 8192, 16384, 32768, 65536};
......
......@@ -39,7 +39,7 @@
* 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).
* 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.
* attempted_compress: 1 bit, boolean, used for verifying during testing.
* extra: 10 bits, free for future use; pads out the remainder of 32 bits */
......
......@@ -2746,7 +2746,7 @@ void stopLoading(int success) {
}
void startSaving(int rdbflags) {
/* Fire the persistence modules end event. */
/* Fire the persistence modules start event. */
int subevent;
if (rdbflags & RDBFLAGS_AOF_PREAMBLE && getpid() != server.pid)
subevent = REDISMODULE_SUBEVENT_PERSISTENCE_AOF_START;
......@@ -2782,6 +2782,9 @@ void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) {
processEventsWhileBlocked();
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.
......@@ -3476,6 +3479,9 @@ void saveCommand(client *c) {
addReplyError(c,"Background save already in progress");
return;
}
server.stat_rdb_saves++;
rdbSaveInfo rsi, *rsiptr;
rsiptr = rdbPopulateSaveInfo(&rsi);
if (rdbSave(SLAVE_REQ_NONE,server.rdb_filename,rsiptr) == C_OK) {
......
......@@ -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) {
unsigned long long sampled = 0, total_keys, totlen=0, *sizes=NULL, it=0, scan_loops = 0;
redisReply *reply, *keys;
......@@ -8237,6 +8242,7 @@ static void findBigKeys(int memkeys, unsigned memkeys_samples) {
typeinfo_add(types_dict, "zset", &type_zset);
typeinfo_add(types_dict, "stream", &type_stream);
signal(SIGINT, longStatLoopModeStop);
/* Total keys pre scanning */
total_keys = getDbSize();
......@@ -8315,14 +8321,14 @@ static void findBigKeys(int memkeys, unsigned memkeys_samples) {
}
freeReplyObject(reply);
} while(it != 0);
} while(force_cancel_loop == 0 && it != 0);
if(types) zfree(types);
if(sizes) zfree(sizes);
/* We're done */
printf("\n-------- summary -------\n\n");
if (force_cancel_loop) printf("[%05.2f%%] ", pct);
printf("Sampled %llu keys in the keyspace!\n", sampled);
printf("Total key length in bytes is %llu (avg len %.2f)\n\n",
totlen, totlen ? (double)totlen/sampled : 0);
......@@ -8401,6 +8407,7 @@ static void findHotKeys(void) {
unsigned int arrsize = 0, i, k;
double pct;
signal(SIGINT, longStatLoopModeStop);
/* Total keys pre scanning */
total_keys = getDbSize();
......@@ -8466,13 +8473,13 @@ static void findHotKeys(void) {
}
freeReplyObject(reply);
} while(it != 0);
} while(force_cancel_loop ==0 && it != 0);
if (freqs) zfree(freqs);
/* We're done */
printf("\n-------- summary -------\n\n");
if(force_cancel_loop)printf("[%05.2f%%] ",pct);
printf("Sampled %llu keys in the keyspace!\n", sampled);
for (i=1; i<= HOTKEYS_SAMPLE; i++) {
......@@ -8643,7 +8650,7 @@ static void statMode(void) {
static void scanMode(void) {
redisReply *reply;
unsigned long long cur = 0;
signal(SIGINT, longStatLoopModeStop);
do {
reply = sendScan(&cur);
for (unsigned int j = 0; j < reply->element[1]->elements; j++) {
......@@ -8658,7 +8665,7 @@ static void scanMode(void) {
}
freeReplyObject(reply);
if (config.interval) usleep(config.interval);
} while(cur != 0);
} while(force_cancel_loop == 0 && cur != 0);
exit(0);
}
......@@ -8785,11 +8792,6 @@ unsigned long compute_something_fast(void) {
return output;
}
static void intrinsicLatencyModeStop(int s) {
UNUSED(s);
force_cancel_loop = 1;
}
static void sigIntHandler(int s) {
UNUSED(s);
......@@ -8807,7 +8809,7 @@ static void intrinsicLatencyMode(void) {
run_time = (long long)config.intrinsic_latency_duration * 1000000;
test_end = ustime() + run_time;
signal(SIGINT, intrinsicLatencyModeStop);
signal(SIGINT, longStatLoopModeStop);
while(1) {
long long start, end, latency;
......
......@@ -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_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_NEW (1<<14) /* n, new key notification */
/* Next notification flag, must be updated when adding new flags above!
This flag should not be used directly by the module.
* 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 */
......@@ -248,7 +249,7 @@ typedef uint64_t RedisModuleTimerID;
/* When set, Redis will not call RedisModule_SignalModifiedKey(), implicitly in
* 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)
/* Declare that the module can handle diskless async replication with RedisModule_SetModuleOptions. */
......@@ -346,7 +347,7 @@ typedef struct {
const char *keyword;
/* An index in argv from which to start searching.
* 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;
} keyword;
} bs;
......@@ -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 float (*RedisModule_GetUsedMemoryRatio)() 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_MallocSizeDict)(RedisModuleDict* dict) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleUser * (*RedisModule_CreateModuleUser)(const char *name) REDISMODULE_ATTR;
......@@ -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_EventLoopDel)(int fd, int mask) 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_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;
......@@ -1493,6 +1495,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(KillForkChild);
REDISMODULE_GET_API(GetUsedMemoryRatio);
REDISMODULE_GET_API(MallocSize);
REDISMODULE_GET_API(MallocUsableSize);
REDISMODULE_GET_API(MallocSizeString);
REDISMODULE_GET_API(MallocSizeDict);
REDISMODULE_GET_API(CreateModuleUser);
......
......@@ -1358,7 +1358,7 @@ void sendBulkToSlave(connection *conn) {
freeClient(slave);
return;
}
atomicIncr(server.stat_net_output_bytes, nwritten);
atomicIncr(server.stat_net_repl_output_bytes, nwritten);
sdsrange(slave->replpreamble,nwritten,-1);
if (sdslen(slave->replpreamble) == 0) {
sdsfree(slave->replpreamble);
......@@ -1387,7 +1387,7 @@ void sendBulkToSlave(connection *conn) {
return;
}
slave->repldboff += nwritten;
atomicIncr(server.stat_net_output_bytes, nwritten);
atomicIncr(server.stat_net_repl_output_bytes, nwritten);
if (slave->repldboff == slave->repldbsize) {
close(slave->repldbfd);
slave->repldbfd = -1;
......@@ -1419,7 +1419,7 @@ void rdbPipeWriteHandlerConnRemoved(struct connection *conn) {
void rdbPipeWriteHandler(struct connection *conn) {
serverAssert(server.rdb_pipe_bufflen>0);
client *slave = connGetPrivateData(conn);
int nwritten;
ssize_t nwritten;
if ((nwritten = connWrite(conn, server.rdb_pipe_buff + slave->repldboff,
server.rdb_pipe_bufflen - slave->repldboff)) == -1)
{
......@@ -1431,7 +1431,7 @@ void rdbPipeWriteHandler(struct connection *conn) {
return;
} else {
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) {
slave->repl_last_partial_write = server.unixtime;
return; /* more data to write.. */
......@@ -1491,7 +1491,7 @@ void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData,
int stillAlive = 0;
for (i=0; i < server.rdb_pipe_numconns; i++)
{
int nwritten;
ssize_t nwritten;
connection *conn = server.rdb_pipe_conns[i];
if (!conn)
continue;
......@@ -1511,7 +1511,7 @@ void rdbPipeReadHandler(struct aeEventLoop *eventLoop, int fd, void *clientData,
/* Note: when use diskless replication, 'repldboff' is the offset
* of 'rdb_pipe_buff' sent rather than the offset of entire RDB. */
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,
* setup write handler (and disable pipe read handler, below) */
......@@ -1817,11 +1817,16 @@ void readSyncBulkPayload(connection *conn) {
/* If repl_transfer_size == -1 we still have to read the bulk length
* from the master reply. */
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,
"I/O error reading bulk count from MASTER: %s",
strerror(errno));
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] == '-') {
......@@ -1892,7 +1897,7 @@ void readSyncBulkPayload(connection *conn) {
cancelReplicationHandshake(1);
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
* writing the EOF mark into the file... */
......@@ -2223,8 +2228,8 @@ char *receiveSynchronousResponse(connection *conn) {
/* Read the reply from the server. */
if (connSyncReadLine(conn,buf,sizeof(buf),server.repl_syncio_timeout*1000) == -1)
{
return sdscatprintf(sdsempty(),"-Reading from master: %s",
strerror(errno));
serverLog(LL_WARNING, "Failed to read response from the server: %s", strerror(errno));
return NULL;
}
server.repl_transfer_lastio = server.unixtime;
return sdsnew(buf);
......@@ -2407,6 +2412,13 @@ int slaveTryPartialResynchronization(connection *conn, int read_reply) {
/* Reading half */
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) {
/* The master may send empty newlines after it receives PSYNC
* and before to reply, just to keep the connection alive. */
......@@ -2566,6 +2578,9 @@ void syncWithMaster(connection *conn) {
if (server.repl_state == REPL_STATE_RECEIVE_PING_REPLY) {
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 just check for "+") or an authentication error.
* Note that older versions of Redis replied with "operation not
......@@ -2652,6 +2667,7 @@ void syncWithMaster(connection *conn) {
/* Receive AUTH reply. */
if (server.repl_state == REPL_STATE_RECEIVE_AUTH_REPLY) {
err = receiveSynchronousResponse(conn);
if (err == NULL) goto no_response_error;
if (err[0] == '-') {
serverLog(LL_WARNING,"Unable to AUTH to MASTER: %s",err);
sdsfree(err);
......@@ -2666,6 +2682,7 @@ void syncWithMaster(connection *conn) {
/* Receive REPLCONF listening-port reply. */
if (server.repl_state == REPL_STATE_RECEIVE_PORT_REPLY) {
err = receiveSynchronousResponse(conn);
if (err == NULL) goto no_response_error;
/* Ignore the error if any, not all the Redis versions support
* REPLCONF listening-port. */
if (err[0] == '-') {
......@@ -2683,6 +2700,7 @@ void syncWithMaster(connection *conn) {
/* Receive REPLCONF ip-address reply. */
if (server.repl_state == REPL_STATE_RECEIVE_IP_REPLY) {
err = receiveSynchronousResponse(conn);
if (err == NULL) goto no_response_error;
/* Ignore the error if any, not all the Redis versions support
* REPLCONF ip-address. */
if (err[0] == '-') {
......@@ -2697,6 +2715,7 @@ void syncWithMaster(connection *conn) {
/* Receive CAPA reply. */
if (server.repl_state == REPL_STATE_RECEIVE_CAPA_REPLY) {
err = receiveSynchronousResponse(conn);
if (err == NULL) goto no_response_error;
/* Ignore the error if any, not all the Redis versions support
* REPLCONF capa. */
if (err[0] == '-') {
......@@ -2810,6 +2829,10 @@ void syncWithMaster(connection *conn) {
server.repl_transfer_lastio = server.unixtime;
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:
if (dfd != -1) close(dfd);
connClose(conn);
......
......@@ -438,6 +438,20 @@ void rioSetAutoSync(rio *r, off_t 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 --------------------------
*
* The following higher level functions use lower level rio.c functions to help
......
......@@ -40,6 +40,11 @@
#define RIO_FLAG_READ_ERROR (1<<0)
#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 {
/* Backend functions.
* Since this functions do not tolerate short writes or reads the return
......@@ -174,5 +179,5 @@ int rioWriteBulkObject(rio *r, struct redisObject *obj);
void rioGenericUpdateChecksum(rio *r, const void *buf, size_t len);
void rioSetAutoSync(rio *r, off_t bytes);
uint8_t rioCheckType(rio *r);
#endif
......@@ -92,8 +92,8 @@ int scriptInterrupt(scriptRunCtx *run_ctx) {
serverLog(LL_WARNING,
"Slow script detected: still in execution after %lld milliseconds. "
"You can try killing the script using the %s command.",
elapsed, (run_ctx->flags & SCRIPT_EVAL_MODE) ? "SCRIPT KILL" : "FUNCTION KILL");
"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", run_ctx->funcname);
enterScriptTimedoutMode(run_ctx);
/* Once the script timeouts we reenter the event loop to permit others
......@@ -108,6 +108,25 @@ int scriptInterrupt(scriptRunCtx *run_ctx) {
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 */
int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *caller, const char *funcname, uint64_t script_flags, int ro) {
serverAssert(!curr_run_ctx);
......@@ -123,12 +142,6 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca
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)) {
addReplyError(caller, "-MASTERDOWN Link with MASTER is down, "
"replica-serve-stale-data is set to 'no' "
......@@ -142,7 +155,7 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca
* 2. no disk error detected
* 3. command is not `fcall_ro`/`eval[sha]_ro` */
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;
}
......@@ -177,6 +190,17 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca
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 {
/* Special handling for backwards compatibility (no shebang eval[sha]) mode */
if (running_stale) {
......@@ -202,8 +226,6 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca
script_client->flags |= CLIENT_MULTI;
}
server.in_script = 1;
run_ctx->start_time = getMonotonicUs();
run_ctx->snapshot_time = mstime();
......@@ -216,6 +238,8 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca
run_ctx->flags |= SCRIPT_READ_ONLY;
}
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;
}
......@@ -236,7 +260,6 @@ void scriptResetRun(scriptRunCtx *run_ctx) {
/* After the script done, remove the MULTI state. */
run_ctx->c->flags &= ~CLIENT_MULTI;
server.in_script = 0;
server.script_caller = NULL;
if (scriptIsTimedout()) {
......@@ -323,16 +346,23 @@ static int scriptVerifyACL(client *c, sds *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) {
/* We know its a write command, on a read only run we do not allow it. */
/* A write command, on an RO command or an RO script is rejected ASAP.
* 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.");
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
* command marked as non-deterministic was already called in the context
* of this script. */
......@@ -362,16 +392,6 @@ static int scriptVerifyWriteCommandAllow(scriptRunCtx *run_ctx, char **err) {
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) {
if (run_ctx->flags & SCRIPT_ALLOW_OOM) {
/* Allow running any command even if OOM reached */
......@@ -386,7 +406,7 @@ static int scriptVerifyOOM(scriptRunCtx *run_ctx, char **err) {
if (server.maxmemory && /* Maxmemory is actually enabled. */
!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. */
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))
{
*err = sdsdup(shared.oomerr->ptr);
......@@ -526,10 +546,6 @@ void scriptCall(scriptRunCtx *run_ctx, robj* *argv, int argc, sds *err) {
goto error;
}
if (scriptVerifyMayReplicate(run_ctx, err) != C_OK) {
goto error;
}
if (scriptVerifyOOM(run_ctx, err) != C_OK) {
goto error;
}
......
......@@ -93,6 +93,7 @@ typedef struct scriptFlag {
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);
void scriptResetRun(scriptRunCtx *r_ctx);
int scriptSetResp(scriptRunCtx *r_ctx, int resp);
......
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