Commit 7b9fc6fc authored by Oran Agra's avatar Oran Agra
Browse files

Merge 'origin/unstable' into 7.0 for 7.0.3

parents 05833959 693acc01
...@@ -282,7 +282,7 @@ int dictRehashMilliseconds(dict *d, int ms) { ...@@ -282,7 +282,7 @@ int dictRehashMilliseconds(dict *d, int ms) {
/* This function performs just a step of rehashing, and only if hashing has /* This function performs just a step of rehashing, and only if hashing has
* not been paused for our hash table. When we have iterators in the * not been paused for our hash table. When we have iterators in the
* middle of a rehashing we can't mess with the two hash tables otherwise * middle of a rehashing we can't mess with the two hash tables otherwise
* some element can be missed or duplicated. * some elements can be missed or duplicated.
* *
* This function is called by common lookup or update operations in the * This function is called by common lookup or update operations in the
* dictionary so that the hash table automatically migrates from H1 to H2 * dictionary so that the hash table automatically migrates from H1 to H2
......
...@@ -266,14 +266,14 @@ void evictionPoolPopulate(int dbid, dict *sampledict, dict *keydict, struct evic ...@@ -266,14 +266,14 @@ void evictionPoolPopulate(int dbid, dict *sampledict, dict *keydict, struct evic
* has a low value. * has a low value.
* *
* New keys don't start at zero, in order to have the ability to collect * New keys don't start at zero, in order to have the ability to collect
* some accesses before being trashed away, so they start at COUNTER_INIT_VAL. * some accesses before being trashed away, so they start at LFU_INIT_VAL.
* The logarithmic increment performed on LOG_C takes care of COUNTER_INIT_VAL * The logarithmic increment performed on LOG_C takes care of LFU_INIT_VAL
* when incrementing the key, so that keys starting at COUNTER_INIT_VAL * when incrementing the key, so that keys starting at LFU_INIT_VAL
* (or having a smaller value) have a very high chance of being incremented * (or having a smaller value) have a very high chance of being incremented
* on access. * on access.
* *
* During decrement, the value of the logarithmic counter is halved if * During decrement, the value of the logarithmic counter is halved if
* its current value is greater than two times the COUNTER_INIT_VAL, otherwise * its current value is greater than two times the LFU_INIT_VAL, otherwise
* it is just decremented by one. * it is just decremented by one.
* --------------------------------------------------------------------------*/ * --------------------------------------------------------------------------*/
...@@ -295,7 +295,7 @@ unsigned long LFUTimeElapsed(unsigned long ldt) { ...@@ -295,7 +295,7 @@ unsigned long LFUTimeElapsed(unsigned long ldt) {
} }
/* Logarithmically increment a counter. The greater is the current counter value /* Logarithmically increment a counter. The greater is the current counter value
* the less likely is that it gets really implemented. Saturate it at 255. */ * the less likely is that it gets really incremented. Saturate it at 255. */
uint8_t LFULogIncr(uint8_t counter) { uint8_t LFULogIncr(uint8_t counter) {
if (counter == 255) return 255; if (counter == 255) return 255;
double r = (double)rand()/RAND_MAX; double r = (double)rand()/RAND_MAX;
......
...@@ -674,7 +674,7 @@ void fcallroCommand(client *c) { ...@@ -674,7 +674,7 @@ void fcallroCommand(client *c) {
* * Engine name * * Engine name
* * Library description * * Library description
* * Library code * * Library code
* RDB_OPCODE_FUNCTION is saved before each library to present * RDB_OPCODE_FUNCTION2 is saved before each library to present
* that the payload is a library. * that the payload is a library.
* RDB version and crc64 is saved at the end of the payload. * RDB version and crc64 is saved at the end of the payload.
* The RDB version is saved for backward compatibility. * The RDB version is saved for backward compatibility.
......
/* Automatically generated by ./utils/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
...@@ -138,7 +138,7 @@ struct commandHelp { ...@@ -138,7 +138,7 @@ struct commandHelp {
"key GET encoding offset [GET 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.0.0" },
{ "BITOP", { "BITOP",
"operation destkey key [key ...]", "operation destkey key [key ...]",
"Perform bitwise operations between strings", "Perform bitwise operations between strings",
...@@ -355,7 +355,7 @@ struct commandHelp { ...@@ -355,7 +355,7 @@ struct commandHelp {
12, 12,
"7.0.0" }, "7.0.0" },
{ "CLUSTER MEET", { "CLUSTER MEET",
"ip port", "ip port [cluster_bus_port]",
"Force a node cluster to handshake with another node", "Force a node cluster to handshake with another node",
12, 12,
"3.0.0" }, "3.0.0" },
...@@ -1775,7 +1775,7 @@ struct commandHelp { ...@@ -1775,7 +1775,7 @@ struct commandHelp {
4, 4,
"6.2.0" }, "6.2.0" },
{ "ZRANGE", { "ZRANGE",
"key min max [BYSCORE|BYLEX] [REV] [LIMIT offset count] [WITHSCORES]", "key start stop [BYSCORE|BYLEX] [REV] [LIMIT offset count] [WITHSCORES]",
"Return a range of members in a sorted set", "Return a range of members in a sorted set",
4, 4,
"1.2.0" }, "1.2.0" },
......
...@@ -49,7 +49,6 @@ ...@@ -49,7 +49,6 @@
#define LP_HDR_NUMELE_UNKNOWN UINT16_MAX #define LP_HDR_NUMELE_UNKNOWN UINT16_MAX
#define LP_MAX_INT_ENCODING_LEN 9 #define LP_MAX_INT_ENCODING_LEN 9
#define LP_MAX_BACKLEN_SIZE 5 #define LP_MAX_BACKLEN_SIZE 5
#define LP_MAX_ENTRY_BACKLEN 34359738367ULL
#define LP_ENCODING_INT 0 #define LP_ENCODING_INT 0
#define LP_ENCODING_STRING 1 #define LP_ENCODING_STRING 1
...@@ -569,9 +568,7 @@ unsigned long lpLength(unsigned char *lp) { ...@@ -569,9 +568,7 @@ unsigned long lpLength(unsigned char *lp) {
* this lib. * this lib.
* *
* Similarly, there is no error returned since the listpack normally can be * Similarly, there is no error returned since the listpack normally can be
* assumed to be valid, so that would be a very high API cost. However a function * assumed to be valid, so that would be a very high API cost. */
* in order to check the integrity of the listpack at load time is provided,
* check lpIsValid(). */
static inline unsigned char *lpGetWithSize(unsigned char *p, int64_t *count, unsigned char *intbuf, uint64_t *entry_size) { static inline unsigned char *lpGetWithSize(unsigned char *p, int64_t *count, unsigned char *intbuf, uint64_t *entry_size) {
int64_t val; int64_t val;
uint64_t uval, negstart, negmax; uint64_t uval, negstart, negmax;
......
...@@ -2297,7 +2297,7 @@ RedisModuleString *RM_CreateStringPrintf(RedisModuleCtx *ctx, const char *fmt, . ...@@ -2297,7 +2297,7 @@ RedisModuleString *RM_CreateStringPrintf(RedisModuleCtx *ctx, const char *fmt, .
} }
/* Like RedisModule_CreatString(), but creates a string starting from a `long long` /* Like RedisModule_CreateString(), but creates a string starting from a `long long`
* integer instead of taking a buffer and its length. * integer instead of taking a buffer and its length.
* *
* The returned string must be released with RedisModule_FreeString() or by * The returned string must be released with RedisModule_FreeString() or by
...@@ -2311,7 +2311,21 @@ RedisModuleString *RM_CreateStringFromLongLong(RedisModuleCtx *ctx, long long ll ...@@ -2311,7 +2311,21 @@ RedisModuleString *RM_CreateStringFromLongLong(RedisModuleCtx *ctx, long long ll
return RM_CreateString(ctx,buf,len); return RM_CreateString(ctx,buf,len);
} }
/* Like RedisModule_CreatString(), but creates a string starting from a double /* Like RedisModule_CreateString(), but creates a string starting from a `unsigned long long`
* integer instead of taking a buffer and its length.
*
* The returned string must be released with RedisModule_FreeString() or by
* enabling automatic memory management.
*
* The passed context 'ctx' may be NULL if necessary, see the
* RedisModule_CreateString() documentation for more info. */
RedisModuleString *RM_CreateStringFromULongLong(RedisModuleCtx *ctx, unsigned long long ull) {
char buf[LONG_STR_SIZE];
size_t len = ull2string(buf,sizeof(buf),ull);
return RM_CreateString(ctx,buf,len);
}
/* Like RedisModule_CreateString(), but creates a string starting from a double
* instead of taking a buffer and its length. * instead of taking a buffer and its length.
* *
* The returned string must be released with RedisModule_FreeString() or by * The returned string must be released with RedisModule_FreeString() or by
...@@ -2322,7 +2336,7 @@ RedisModuleString *RM_CreateStringFromDouble(RedisModuleCtx *ctx, double d) { ...@@ -2322,7 +2336,7 @@ RedisModuleString *RM_CreateStringFromDouble(RedisModuleCtx *ctx, double d) {
return RM_CreateString(ctx,buf,len); return RM_CreateString(ctx,buf,len);
} }
/* Like RedisModule_CreatString(), but creates a string starting from a long /* Like RedisModule_CreateString(), but creates a string starting from a long
* double. * double.
* *
* The returned string must be released with RedisModule_FreeString() or by * The returned string must be released with RedisModule_FreeString() or by
...@@ -2337,7 +2351,7 @@ RedisModuleString *RM_CreateStringFromLongDouble(RedisModuleCtx *ctx, long doubl ...@@ -2337,7 +2351,7 @@ RedisModuleString *RM_CreateStringFromLongDouble(RedisModuleCtx *ctx, long doubl
return RM_CreateString(ctx,buf,len); return RM_CreateString(ctx,buf,len);
} }
/* Like RedisModule_CreatString(), but creates a string starting from another /* Like RedisModule_CreateString(), but creates a string starting from another
* RedisModuleString. * RedisModuleString.
* *
* The returned string must be released with RedisModule_FreeString() or by * The returned string must be released with RedisModule_FreeString() or by
...@@ -2519,6 +2533,14 @@ int RM_StringToLongLong(const RedisModuleString *str, long long *ll) { ...@@ -2519,6 +2533,14 @@ int RM_StringToLongLong(const RedisModuleString *str, long long *ll) {
REDISMODULE_ERR; REDISMODULE_ERR;
} }
/* Convert the string into a `unsigned long long` integer, storing it at `*ull`.
* Returns REDISMODULE_OK on success. If the string can't be parsed
* as a valid, strict `unsigned long long` (no spaces before/after), REDISMODULE_ERR
* is returned. */
int RM_StringToULongLong(const RedisModuleString *str, unsigned long long *ull) {
return string2ull(str->ptr,ull) ? REDISMODULE_OK : REDISMODULE_ERR;
}
/* Convert the string into a double, storing it at `*d`. /* Convert the string into a double, storing it at `*d`.
* Returns REDISMODULE_OK on success or REDISMODULE_ERR if the string is * Returns REDISMODULE_OK on success or REDISMODULE_ERR if the string is
* not a valid string representation of a double value. */ * not a valid string representation of a double value. */
...@@ -3327,8 +3349,8 @@ int modulePopulateReplicationInfoStructure(void *ri, int structver) { ...@@ -3327,8 +3349,8 @@ int modulePopulateReplicationInfoStructure(void *ri, int structver) {
* is returned. * is returned.
* *
* When the client exist and the `ci` pointer is not NULL, but points to * When the client exist and the `ci` pointer is not NULL, but points to
* a structure of type RedisModuleClientInfo, previously initialized with * a structure of type RedisModuleClientInfoV1, previously initialized with
* the correct REDISMODULE_CLIENTINFO_INITIALIZER, the structure is populated * the correct REDISMODULE_CLIENTINFO_INITIALIZER_V1, the structure is populated
* with the following fields: * with the following fields:
* *
* uint64_t flags; // REDISMODULE_CLIENTINFO_FLAG_* * uint64_t flags; // REDISMODULE_CLIENTINFO_FLAG_*
...@@ -3373,6 +3395,40 @@ int RM_GetClientInfoById(void *ci, uint64_t id) { ...@@ -3373,6 +3395,40 @@ int RM_GetClientInfoById(void *ci, uint64_t id) {
return modulePopulateClientInfoStructure(ci,client,structver); return modulePopulateClientInfoStructure(ci,client,structver);
} }
/* Returns the name of the client connection with the given ID.
*
* If the client ID does not exist or if the client has no name associated with
* it, NULL is returned. */
RedisModuleString *RM_GetClientNameById(RedisModuleCtx *ctx, uint64_t id) {
client *client = lookupClientByID(id);
if (client == NULL || client->name == NULL) return NULL;
robj *name = client->name;
incrRefCount(name);
autoMemoryAdd(ctx, REDISMODULE_AM_STRING, name);
return name;
}
/* Sets the name of the client with the given ID. This is equivalent to the client calling
* `CLIENT SETNAME name`.
*
* Returns REDISMODULE_OK on success. On failure, REDISMODULE_ERR is returned
* and errno is set as follows:
*
* - ENOENT if the client does not exist
* - EINVAL if the name contains invalid characters */
int RM_SetClientNameById(uint64_t id, RedisModuleString *name) {
client *client = lookupClientByID(id);
if (client == NULL) {
errno = ENOENT;
return REDISMODULE_ERR;
}
if (clientSetName(client, name) == C_ERR) {
errno = EINVAL;
return REDISMODULE_ERR;
}
return REDISMODULE_OK;
}
/* Publish a message to subscribers (see PUBLISH command). */ /* Publish a message to subscribers (see PUBLISH command). */
int RM_PublishMessage(RedisModuleCtx *ctx, RedisModuleString *channel, RedisModuleString *message) { int RM_PublishMessage(RedisModuleCtx *ctx, RedisModuleString *channel, RedisModuleString *message) {
UNUSED(ctx); UNUSED(ctx);
...@@ -5783,7 +5839,16 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch ...@@ -5783,7 +5839,16 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
if (flags & REDISMODULE_ARGV_RESPECT_DENY_OOM) { if (flags & REDISMODULE_ARGV_RESPECT_DENY_OOM) {
if (cmd->flags & CMD_DENYOOM) { if (cmd->flags & CMD_DENYOOM) {
if (server.pre_command_oom_state) { int oom_state;
if (ctx->flags & REDISMODULE_CTX_THREAD_SAFE) {
/* On background thread we can not count on server.pre_command_oom_state.
* Because it is only set on the main thread, in such case we will check
* the actual memory usage. */
oom_state = (getMaxmemoryState(NULL,NULL,NULL,NULL) == C_ERR);
} else {
oom_state = server.pre_command_oom_state;
}
if (oom_state) {
errno = ENOSPC; errno = ENOSPC;
if (error_as_call_replies) { if (error_as_call_replies) {
sds msg = sdsdup(shared.oomerr->ptr); sds msg = sdsdup(shared.oomerr->ptr);
...@@ -5823,7 +5888,7 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch ...@@ -5823,7 +5888,7 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
} }
int deny_write_type = writeCommandsDeniedByDiskError(); int deny_write_type = writeCommandsDeniedByDiskError();
int obey_client = mustObeyClient(server.current_client); int obey_client = (server.current_client && mustObeyClient(server.current_client));
if (deny_write_type != DISK_ERROR_TYPE_NONE && !obey_client) { if (deny_write_type != DISK_ERROR_TYPE_NONE && !obey_client) {
errno = ESPIPE; errno = ESPIPE;
...@@ -12378,6 +12443,7 @@ void moduleRegisterCoreAPI(void) { ...@@ -12378,6 +12443,7 @@ void moduleRegisterCoreAPI(void) {
REGISTER_API(ListInsert); REGISTER_API(ListInsert);
REGISTER_API(ListDelete); REGISTER_API(ListDelete);
REGISTER_API(StringToLongLong); REGISTER_API(StringToLongLong);
REGISTER_API(StringToULongLong);
REGISTER_API(StringToDouble); REGISTER_API(StringToDouble);
REGISTER_API(StringToLongDouble); REGISTER_API(StringToLongDouble);
REGISTER_API(StringToStreamID); REGISTER_API(StringToStreamID);
...@@ -12400,6 +12466,7 @@ void moduleRegisterCoreAPI(void) { ...@@ -12400,6 +12466,7 @@ void moduleRegisterCoreAPI(void) {
REGISTER_API(CreateStringFromCallReply); REGISTER_API(CreateStringFromCallReply);
REGISTER_API(CreateString); REGISTER_API(CreateString);
REGISTER_API(CreateStringFromLongLong); REGISTER_API(CreateStringFromLongLong);
REGISTER_API(CreateStringFromULongLong);
REGISTER_API(CreateStringFromDouble); REGISTER_API(CreateStringFromDouble);
REGISTER_API(CreateStringFromLongDouble); REGISTER_API(CreateStringFromLongDouble);
REGISTER_API(CreateStringFromString); REGISTER_API(CreateStringFromString);
...@@ -12592,6 +12659,8 @@ void moduleRegisterCoreAPI(void) { ...@@ -12592,6 +12659,8 @@ void moduleRegisterCoreAPI(void) {
REGISTER_API(ServerInfoGetFieldUnsigned); REGISTER_API(ServerInfoGetFieldUnsigned);
REGISTER_API(ServerInfoGetFieldDouble); REGISTER_API(ServerInfoGetFieldDouble);
REGISTER_API(GetClientInfoById); REGISTER_API(GetClientInfoById);
REGISTER_API(GetClientNameById);
REGISTER_API(SetClientNameById);
REGISTER_API(PublishMessage); REGISTER_API(PublishMessage);
REGISTER_API(PublishMessageShard); REGISTER_API(PublishMessageShard);
REGISTER_API(SubscribeToServerEvent); REGISTER_API(SubscribeToServerEvent);
......
...@@ -38,6 +38,7 @@ void initClientMultiState(client *c) { ...@@ -38,6 +38,7 @@ void initClientMultiState(client *c) {
c->mstate.cmd_flags = 0; c->mstate.cmd_flags = 0;
c->mstate.cmd_inv_flags = 0; c->mstate.cmd_inv_flags = 0;
c->mstate.argv_len_sums = 0; c->mstate.argv_len_sums = 0;
c->mstate.alloc_count = 0;
} }
/* Release all the resources associated with MULTI/EXEC state */ /* Release all the resources associated with MULTI/EXEC state */
...@@ -65,9 +66,16 @@ void queueMultiCommand(client *c, uint64_t cmd_flags) { ...@@ -65,9 +66,16 @@ void queueMultiCommand(client *c, uint64_t cmd_flags) {
* aborted. */ * aborted. */
if (c->flags & (CLIENT_DIRTY_CAS|CLIENT_DIRTY_EXEC)) if (c->flags & (CLIENT_DIRTY_CAS|CLIENT_DIRTY_EXEC))
return; return;
if (c->mstate.count == 0) {
c->mstate.commands = zrealloc(c->mstate.commands, /* If a client is using multi/exec, assuming it is used to execute at least
sizeof(multiCmd)*(c->mstate.count+1)); * two commands. Hence, creating by default size of 2. */
c->mstate.commands = zmalloc(sizeof(multiCmd)*2);
c->mstate.alloc_count = 2;
}
if (c->mstate.count == c->mstate.alloc_count) {
c->mstate.alloc_count = c->mstate.alloc_count < INT_MAX/2 ? c->mstate.alloc_count*2 : INT_MAX;
c->mstate.commands = zrealloc(c->mstate.commands, sizeof(multiCmd)*(c->mstate.alloc_count));
}
mc = c->mstate.commands+c->mstate.count; mc = c->mstate.commands+c->mstate.count;
mc->cmd = c->cmd; mc->cmd = c->cmd;
mc->argc = c->argc; mc->argc = c->argc;
...@@ -230,6 +238,7 @@ void execCommand(client *c) { ...@@ -230,6 +238,7 @@ void execCommand(client *c) {
/* Commands may alter argc/argv, restore mstate. */ /* Commands may alter argc/argv, restore mstate. */
c->mstate.commands[j].argc = c->argc; c->mstate.commands[j].argc = c->argc;
c->mstate.commands[j].argv = c->argv; c->mstate.commands[j].argv = c->argv;
c->mstate.commands[j].argv_len = c->argv_len;
c->mstate.commands[j].cmd = c->cmd; c->mstate.commands[j].cmd = c->cmd;
} }
...@@ -465,5 +474,7 @@ size_t multiStateMemOverhead(client *c) { ...@@ -465,5 +474,7 @@ size_t multiStateMemOverhead(client *c) {
size_t mem = c->mstate.argv_len_sums; size_t mem = c->mstate.argv_len_sums;
/* Add watched keys overhead, Note: this doesn't take into account the watched keys themselves, because they aren't managed per-client. */ /* Add watched keys overhead, Note: this doesn't take into account the watched keys themselves, because they aren't managed per-client. */
mem += listLength(c->watched_keys) * (sizeof(listNode) + sizeof(watchedKey)); mem += listLength(c->watched_keys) * (sizeof(listNode) + sizeof(watchedKey));
/* Reserved memory for queued multi commands. */
mem += c->mstate.alloc_count * sizeof(multiCmd);
return mem; return mem;
} }
...@@ -2785,7 +2785,7 @@ sds catClientInfoString(sds s, client *client) { ...@@ -2785,7 +2785,7 @@ sds catClientInfoString(sds s, client *client) {
} }
sds ret = sdscatfmt(s, sds ret = sdscatfmt(s,
"id=%U addr=%s laddr=%s %s name=%s age=%I idle=%I flags=%s db=%i sub=%i psub=%i multi=%i qbuf=%U qbuf-free=%U argv-mem=%U multi-mem=%U rbs=%U rbp=%U obl=%U oll=%U omem=%U tot-mem=%U events=%s cmd=%s user=%s redir=%I resp=%i", "id=%U addr=%s laddr=%s %s name=%s age=%I idle=%I flags=%s db=%i sub=%i psub=%i ssub=%i multi=%i qbuf=%U qbuf-free=%U argv-mem=%U multi-mem=%U rbs=%U rbp=%U obl=%U oll=%U omem=%U tot-mem=%U events=%s cmd=%s user=%s redir=%I resp=%i",
(unsigned long long) client->id, (unsigned long long) client->id,
getClientPeerId(client), getClientPeerId(client),
getClientSockname(client), getClientSockname(client),
...@@ -2797,6 +2797,7 @@ sds catClientInfoString(sds s, client *client) { ...@@ -2797,6 +2797,7 @@ sds catClientInfoString(sds s, client *client) {
client->db->id, client->db->id,
(int) dictSize(client->pubsub_channels), (int) dictSize(client->pubsub_channels),
(int) listLength(client->pubsub_patterns), (int) listLength(client->pubsub_patterns),
(int) dictSize(client->pubsubshard_channels),
(client->flags & CLIENT_MULTI) ? client->mstate.count : -1, (client->flags & CLIENT_MULTI) ? client->mstate.count : -1,
(unsigned long long) sdslen(client->querybuf), (unsigned long long) sdslen(client->querybuf),
(unsigned long long) sdsavail(client->querybuf), (unsigned long long) sdsavail(client->querybuf),
...@@ -2832,18 +2833,9 @@ sds getAllClientsInfoString(int type) { ...@@ -2832,18 +2833,9 @@ sds getAllClientsInfoString(int type) {
return o; return o;
} }
/* This function implements CLIENT SETNAME, including replying to the /* Returns C_OK if the name has been set or C_ERR if the name is invalid. */
* user with an error if the charset is wrong (in that case C_ERR is int clientSetName(client *c, robj *name) {
* returned). If the function succeeded C_OK is returned, and it's up int len = (name != NULL) ? sdslen(name->ptr) : 0;
* to the caller to send a reply if needed.
*
* Setting an empty string as name has the effect of unsetting the
* currently set name: the client will remain unnamed.
*
* This function is also used to implement the HELLO SETNAME option. */
int clientSetNameOrReply(client *c, robj *name) {
int len = sdslen(name->ptr);
char *p = name->ptr;
/* Setting the client name to an empty string actually removes /* Setting the client name to an empty string actually removes
* the current name. */ * the current name. */
...@@ -2856,11 +2848,9 @@ int clientSetNameOrReply(client *c, robj *name) { ...@@ -2856,11 +2848,9 @@ int clientSetNameOrReply(client *c, robj *name) {
/* Otherwise check if the charset is ok. We need to do this otherwise /* Otherwise check if the charset is ok. We need to do this otherwise
* CLIENT LIST format will break. You should always be able to * CLIENT LIST format will break. You should always be able to
* split by space to get the different fields. */ * split by space to get the different fields. */
char *p = name->ptr;
for (int j = 0; j < len; j++) { for (int j = 0; j < len; j++) {
if (p[j] < '!' || p[j] > '~') { /* ASCII is assumed. */ if (p[j] < '!' || p[j] > '~') { /* ASCII is assumed. */
addReplyError(c,
"Client names cannot contain spaces, "
"newlines or special characters.");
return C_ERR; return C_ERR;
} }
} }
...@@ -2870,6 +2860,25 @@ int clientSetNameOrReply(client *c, robj *name) { ...@@ -2870,6 +2860,25 @@ int clientSetNameOrReply(client *c, robj *name) {
return C_OK; return C_OK;
} }
/* This function implements CLIENT SETNAME, including replying to the
* user with an error if the charset is wrong (in that case C_ERR is
* returned). If the function succeeded C_OK is returned, and it's up
* to the caller to send a reply if needed.
*
* Setting an empty string as name has the effect of unsetting the
* currently set name: the client will remain unnamed.
*
* This function is also used to implement the HELLO SETNAME option. */
int clientSetNameOrReply(client *c, robj *name) {
int result = clientSetName(c, name);
if (result == C_ERR) {
addReplyError(c,
"Client names cannot contain spaces, "
"newlines or special characters.");
}
return result;
}
/* Reset the client state to resemble a newly connected client. /* Reset the client state to resemble a newly connected client.
*/ */
void resetCommand(client *c) { void resetCommand(client *c) {
...@@ -3660,9 +3669,7 @@ size_t getClientMemoryUsage(client *c, size_t *output_buffer_mem_usage) { ...@@ -3660,9 +3669,7 @@ size_t getClientMemoryUsage(client *c, size_t *output_buffer_mem_usage) {
/* Add memory overhead of pubsub channels and patterns. Note: this is just the overhead of the robj pointers /* Add memory overhead of pubsub channels and patterns. Note: this is just the overhead of the robj pointers
* to the strings themselves because they aren't stored per client. */ * to the strings themselves because they aren't stored per client. */
mem += listLength(c->pubsub_patterns) * sizeof(listNode); mem += pubsubMemOverhead(c);
mem += dictSize(c->pubsub_channels) * sizeof(dictEntry) +
dictSlots(c->pubsub_channels) * sizeof(dictEntry*);
/* Add memory overhead of the tracking prefixes, this is an underestimation so we don't need to traverse the entire rax */ /* Add memory overhead of the tracking prefixes, this is an underestimation so we don't need to traverse the entire rax */
if (c->client_tracking_prefixes) if (c->client_tracking_prefixes)
...@@ -3997,10 +4004,15 @@ void processEventsWhileBlocked(void) { ...@@ -3997,10 +4004,15 @@ void processEventsWhileBlocked(void) {
* ========================================================================== */ * ========================================================================== */
#define IO_THREADS_MAX_NUM 128 #define IO_THREADS_MAX_NUM 128
#define CACHE_LINE_SIZE 64
typedef struct __attribute__((aligned(CACHE_LINE_SIZE))) threads_pending {
redisAtomic unsigned long value;
} threads_pending;
pthread_t io_threads[IO_THREADS_MAX_NUM]; pthread_t io_threads[IO_THREADS_MAX_NUM];
pthread_mutex_t io_threads_mutex[IO_THREADS_MAX_NUM]; pthread_mutex_t io_threads_mutex[IO_THREADS_MAX_NUM];
redisAtomic unsigned long io_threads_pending[IO_THREADS_MAX_NUM]; threads_pending io_threads_pending[IO_THREADS_MAX_NUM];
int io_threads_op; /* IO_THREADS_OP_IDLE, IO_THREADS_OP_READ or IO_THREADS_OP_WRITE. */ // TODO: should access to this be atomic??! int io_threads_op; /* IO_THREADS_OP_IDLE, IO_THREADS_OP_READ or IO_THREADS_OP_WRITE. */ // TODO: should access to this be atomic??!
/* This is the list of clients each thread will serve when threaded I/O is /* This is the list of clients each thread will serve when threaded I/O is
...@@ -4010,12 +4022,12 @@ list *io_threads_list[IO_THREADS_MAX_NUM]; ...@@ -4010,12 +4022,12 @@ list *io_threads_list[IO_THREADS_MAX_NUM];
static inline unsigned long getIOPendingCount(int i) { static inline unsigned long getIOPendingCount(int i) {
unsigned long count = 0; unsigned long count = 0;
atomicGetWithSync(io_threads_pending[i], count); atomicGetWithSync(io_threads_pending[i].value, count);
return count; return count;
} }
static inline void setIOPendingCount(int i, unsigned long count) { static inline void setIOPendingCount(int i, unsigned long count) {
atomicSetWithSync(io_threads_pending[i], count); atomicSetWithSync(io_threads_pending[i].value, count);
} }
void *IOThreadMain(void *myid) { void *IOThreadMain(void *myid) {
......
...@@ -722,3 +722,15 @@ void sunsubscribeCommand(client *c) { ...@@ -722,3 +722,15 @@ void sunsubscribeCommand(client *c) {
} }
if (clientTotalPubSubSubscriptionCount(c) == 0) c->flags &= ~CLIENT_PUBSUB; if (clientTotalPubSubSubscriptionCount(c) == 0) c->flags &= ~CLIENT_PUBSUB;
} }
size_t pubsubMemOverhead(client *c) {
/* PubSub patterns */
size_t mem = listLength(c->pubsub_patterns) * sizeof(listNode);
/* Global PubSub channels */
mem += dictSize(c->pubsub_channels) * sizeof(dictEntry) +
dictSlots(c->pubsub_channels) * sizeof(dictEntry*);
/* Sharded PubSub channels */
mem += dictSize(c->pubsubshard_channels) * sizeof(dictEntry) +
dictSlots(c->pubsubshard_channels) * sizeof(dictEntry*);
return mem;
}
...@@ -1160,7 +1160,6 @@ ssize_t rdbSaveAuxFieldStrInt(rio *rdb, char *key, long long val) { ...@@ -1160,7 +1160,6 @@ ssize_t rdbSaveAuxFieldStrInt(rio *rdb, char *key, long long val) {
/* Save a few default AUX fields with information about the RDB generated. */ /* Save a few default AUX fields with information about the RDB generated. */
int rdbSaveInfoAuxFields(rio *rdb, int rdbflags, rdbSaveInfo *rsi) { int rdbSaveInfoAuxFields(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
UNUSED(rdbflags);
int redis_bits = (sizeof(void*) == 8) ? 64 : 32; int redis_bits = (sizeof(void*) == 8) ? 64 : 32;
int aof_base = (rdbflags & RDBFLAGS_AOF_PREAMBLE) != 0; int aof_base = (rdbflags & RDBFLAGS_AOF_PREAMBLE) != 0;
...@@ -1397,6 +1396,7 @@ int rdbSave(int req, char *filename, rdbSaveInfo *rsi) { ...@@ -1397,6 +1396,7 @@ int rdbSave(int req, char *filename, rdbSaveInfo *rsi) {
FILE *fp = NULL; FILE *fp = NULL;
rio rdb; rio rdb;
int error = 0; int error = 0;
char *err_op; /* For a detailed log */
snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid()); snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid());
fp = fopen(tmpfile,"w"); fp = fopen(tmpfile,"w");
...@@ -1420,13 +1420,14 @@ int rdbSave(int req, char *filename, rdbSaveInfo *rsi) { ...@@ -1420,13 +1420,14 @@ int rdbSave(int req, char *filename, rdbSaveInfo *rsi) {
if (rdbSaveRio(req,&rdb,&error,RDBFLAGS_NONE,rsi) == C_ERR) { if (rdbSaveRio(req,&rdb,&error,RDBFLAGS_NONE,rsi) == C_ERR) {
errno = error; errno = error;
err_op = "rdbSaveRio";
goto werr; goto werr;
} }
/* Make sure data will not remain on the OS's output buffers */ /* Make sure data will not remain on the OS's output buffers */
if (fflush(fp)) goto werr; if (fflush(fp)) { err_op = "fflush"; goto werr; }
if (fsync(fileno(fp))) goto werr; if (fsync(fileno(fp))) { err_op = "fsync"; goto werr; }
if (fclose(fp)) { fp = NULL; goto werr; } if (fclose(fp)) { fp = NULL; err_op = "fclose"; goto werr; }
fp = NULL; fp = NULL;
/* Use RENAME to make sure the DB file is changed atomically only /* Use RENAME to make sure the DB file is changed atomically only
...@@ -1445,6 +1446,7 @@ int rdbSave(int req, char *filename, rdbSaveInfo *rsi) { ...@@ -1445,6 +1446,7 @@ int rdbSave(int req, char *filename, rdbSaveInfo *rsi) {
stopSaving(0); stopSaving(0);
return C_ERR; return C_ERR;
} }
if (fsyncFileDir(filename) == -1) { err_op = "fsyncFileDir"; goto werr; }
serverLog(LL_NOTICE,"DB saved on disk"); serverLog(LL_NOTICE,"DB saved on disk");
server.dirty = 0; server.dirty = 0;
...@@ -1454,7 +1456,7 @@ int rdbSave(int req, char *filename, rdbSaveInfo *rsi) { ...@@ -1454,7 +1456,7 @@ int rdbSave(int req, char *filename, rdbSaveInfo *rsi) {
return C_OK; return C_OK;
werr: werr:
serverLog(LL_WARNING,"Write error saving DB on disk: %s", strerror(errno)); serverLog(LL_WARNING,"Write error saving DB on disk(%s): %s", err_op, strerror(errno));
if (fp) fclose(fp); if (fp) fclose(fp);
unlink(tmpfile); unlink(tmpfile);
stopSaving(0); stopSaving(0);
...@@ -3244,7 +3246,7 @@ eoferr: ...@@ -3244,7 +3246,7 @@ eoferr:
* to do the actual loading. Moreover the ETA displayed in the INFO * to do the actual loading. Moreover the ETA displayed in the INFO
* output is initialized and finalized. * output is initialized and finalized.
* *
* If you pass an 'rsi' structure initialized with RDB_SAVE_OPTION_INIT, the * If you pass an 'rsi' structure initialized with RDB_SAVE_INFO_INIT, the
* loading code will fill the information fields in the structure. */ * loading code will fill the information fields in the structure. */
int rdbLoad(char *filename, rdbSaveInfo *rsi, int rdbflags) { int rdbLoad(char *filename, rdbSaveInfo *rsi, int rdbflags) {
FILE *fp; FILE *fp;
......
...@@ -846,6 +846,10 @@ static client createClient(char *cmd, size_t len, client from, int thread_id) { ...@@ -846,6 +846,10 @@ static client createClient(char *cmd, size_t len, client from, int thread_id) {
} }
if (config.idlemode == 0) if (config.idlemode == 0)
aeCreateFileEvent(el,c->context->fd,AE_WRITABLE,writeHandler,c); aeCreateFileEvent(el,c->context->fd,AE_WRITABLE,writeHandler,c);
else
/* In idle mode, clients still need to register readHandler for catching errors */
aeCreateFileEvent(el,c->context->fd,AE_READABLE,readHandler,c);
listAddNodeTail(config.clients,c); listAddNodeTail(config.clients,c);
atomicIncr(config.liveclients, 1); atomicIncr(config.liveclients, 1);
atomicGet(config.slots_last_update, c->slots_last_update); atomicGet(config.slots_last_update, c->slots_last_update);
......
...@@ -283,7 +283,10 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) { ...@@ -283,7 +283,10 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) {
robj *auxkey, *auxval; robj *auxkey, *auxval;
rdbstate.doing = RDB_CHECK_DOING_READ_AUX; rdbstate.doing = RDB_CHECK_DOING_READ_AUX;
if ((auxkey = rdbLoadStringObject(&rdb)) == NULL) goto eoferr; if ((auxkey = rdbLoadStringObject(&rdb)) == NULL) goto eoferr;
if ((auxval = rdbLoadStringObject(&rdb)) == NULL) goto eoferr; if ((auxval = rdbLoadStringObject(&rdb)) == NULL) {
decrRefCount(auxkey);
goto eoferr;
}
rdbCheckInfo("AUX FIELD %s = '%s'", rdbCheckInfo("AUX FIELD %s = '%s'",
(char*)auxkey->ptr, (char*)auxval->ptr); (char*)auxkey->ptr, (char*)auxval->ptr);
...@@ -297,6 +300,10 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) { ...@@ -297,6 +300,10 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) {
if ((moduleid = rdbLoadLen(&rdb,NULL)) == RDB_LENERR) goto eoferr; if ((moduleid = rdbLoadLen(&rdb,NULL)) == RDB_LENERR) goto eoferr;
if ((when_opcode = rdbLoadLen(&rdb,NULL)) == RDB_LENERR) goto eoferr; if ((when_opcode = rdbLoadLen(&rdb,NULL)) == RDB_LENERR) goto eoferr;
if ((when = rdbLoadLen(&rdb,NULL)) == RDB_LENERR) goto eoferr; if ((when = rdbLoadLen(&rdb,NULL)) == RDB_LENERR) goto eoferr;
if (when_opcode != RDB_MODULE_OPCODE_UINT) {
rdbCheckError("bad when_opcode");
goto err;
}
char name[10]; char name[10];
moduleTypeNameByID(name,moduleid); moduleTypeNameByID(name,moduleid);
......
...@@ -81,6 +81,7 @@ ...@@ -81,6 +81,7 @@
#define REDIS_CLI_CLUSTER_YES_ENV "REDISCLI_CLUSTER_YES" #define REDIS_CLI_CLUSTER_YES_ENV "REDISCLI_CLUSTER_YES"
#define CLUSTER_MANAGER_SLOTS 16384 #define CLUSTER_MANAGER_SLOTS 16384
#define CLUSTER_MANAGER_PORT_INCR 10000 /* same as CLUSTER_PORT_INCR */
#define CLUSTER_MANAGER_MIGRATE_TIMEOUT 60000 #define CLUSTER_MANAGER_MIGRATE_TIMEOUT 60000
#define CLUSTER_MANAGER_MIGRATE_PIPELINE 10 #define CLUSTER_MANAGER_MIGRATE_PIPELINE 10
#define CLUSTER_MANAGER_REBALANCE_THRESHOLD 2 #define CLUSTER_MANAGER_REBALANCE_THRESHOLD 2
...@@ -2867,6 +2868,7 @@ typedef struct clusterManagerNode { ...@@ -2867,6 +2868,7 @@ typedef struct clusterManagerNode {
sds name; sds name;
char *ip; char *ip;
int port; int port;
int bus_port; /* cluster-port */
uint64_t current_epoch; uint64_t current_epoch;
time_t ping_sent; time_t ping_sent;
time_t ping_recv; time_t ping_recv;
...@@ -2937,7 +2939,7 @@ typedef int (*clusterManagerOnReplyError)(redisReply *reply, ...@@ -2937,7 +2939,7 @@ typedef int (*clusterManagerOnReplyError)(redisReply *reply,
/* Cluster Manager helper functions */ /* Cluster Manager helper functions */
static clusterManagerNode *clusterManagerNewNode(char *ip, int port); static clusterManagerNode *clusterManagerNewNode(char *ip, int port, int bus_port);
static clusterManagerNode *clusterManagerNodeByName(const char *name); static clusterManagerNode *clusterManagerNodeByName(const char *name);
static clusterManagerNode *clusterManagerNodeByAbbreviatedName(const char *n); static clusterManagerNode *clusterManagerNodeByAbbreviatedName(const char *n);
static void clusterManagerNodeResetSlots(clusterManagerNode *node); static void clusterManagerNodeResetSlots(clusterManagerNode *node);
...@@ -2997,15 +2999,15 @@ typedef struct clusterManagerCommandDef { ...@@ -2997,15 +2999,15 @@ typedef struct clusterManagerCommandDef {
clusterManagerCommandDef clusterManagerCommands[] = { clusterManagerCommandDef clusterManagerCommands[] = {
{"create", clusterManagerCommandCreate, -2, "host1:port1 ... hostN:portN", {"create", clusterManagerCommandCreate, -2, "host1:port1 ... hostN:portN",
"replicas <arg>"}, "replicas <arg>"},
{"check", clusterManagerCommandCheck, -1, "host:port", {"check", clusterManagerCommandCheck, -1, "<host:port> or <host> <port> - separated by either colon or space",
"search-multiple-owners"}, "search-multiple-owners"},
{"info", clusterManagerCommandInfo, -1, "host:port", NULL}, {"info", clusterManagerCommandInfo, -1, "<host:port> or <host> <port> - separated by either colon or space", NULL},
{"fix", clusterManagerCommandFix, -1, "host:port", {"fix", clusterManagerCommandFix, -1, "<host:port> or <host> <port> - separated by either colon or space",
"search-multiple-owners,fix-with-unreachable-masters"}, "search-multiple-owners,fix-with-unreachable-masters"},
{"reshard", clusterManagerCommandReshard, -1, "host:port", {"reshard", clusterManagerCommandReshard, -1, "<host:port> or <host> <port> - separated by either colon or space",
"from <arg>,to <arg>,slots <arg>,yes,timeout <arg>,pipeline <arg>," "from <arg>,to <arg>,slots <arg>,yes,timeout <arg>,pipeline <arg>,"
"replace"}, "replace"},
{"rebalance", clusterManagerCommandRebalance, -1, "host:port", {"rebalance", clusterManagerCommandRebalance, -1, "<host:port> or <host> <port> - separated by either colon or space",
"weight <node1=w1...nodeN=wN>,use-empty-masters," "weight <node1=w1...nodeN=wN>,use-empty-masters,"
"timeout <arg>,simulate,pipeline <arg>,threshold <arg>,replace"}, "timeout <arg>,simulate,pipeline <arg>,threshold <arg>,replace"},
{"add-node", clusterManagerCommandAddNode, 2, {"add-node", clusterManagerCommandAddNode, 2,
...@@ -3094,6 +3096,7 @@ static clusterManagerCommandProc *validateClusterManagerCommand(void) { ...@@ -3094,6 +3096,7 @@ static clusterManagerCommandProc *validateClusterManagerCommand(void) {
static int parseClusterNodeAddress(char *addr, char **ip_ptr, int *port_ptr, static int parseClusterNodeAddress(char *addr, char **ip_ptr, int *port_ptr,
int *bus_port_ptr) int *bus_port_ptr)
{ {
/* ip:port[@bus_port] */
char *c = strrchr(addr, '@'); char *c = strrchr(addr, '@');
if (c != NULL) { if (c != NULL) {
*c = '\0'; *c = '\0';
...@@ -3203,12 +3206,15 @@ static void freeClusterManager(void) { ...@@ -3203,12 +3206,15 @@ static void freeClusterManager(void) {
dictRelease(clusterManagerUncoveredSlots); dictRelease(clusterManagerUncoveredSlots);
} }
static clusterManagerNode *clusterManagerNewNode(char *ip, int port) { static clusterManagerNode *clusterManagerNewNode(char *ip, int port, int bus_port) {
clusterManagerNode *node = zmalloc(sizeof(*node)); clusterManagerNode *node = zmalloc(sizeof(*node));
node->context = NULL; node->context = NULL;
node->name = NULL; node->name = NULL;
node->ip = ip; node->ip = ip;
node->port = port; node->port = port;
/* We don't need to know the bus_port, at this point this value may be wrong.
* If it is used, it will be corrected in clusterManagerLoadInfoFromNode. */
node->bus_port = bus_port ? bus_port : port + CLUSTER_MANAGER_PORT_INCR;
node->current_epoch = 0; node->current_epoch = 0;
node->ping_sent = 0; node->ping_sent = 0;
node->ping_recv = 0; node->ping_recv = 0;
...@@ -4611,9 +4617,20 @@ static int clusterManagerNodeLoadInfo(clusterManagerNode *node, int opts, ...@@ -4611,9 +4617,20 @@ static int clusterManagerNodeLoadInfo(clusterManagerNode *node, int opts,
success = 0; success = 0;
goto cleanup; goto cleanup;
} }
char *ip = NULL;
int port = 0, bus_port = 0;
if (addr == NULL || !parseClusterNodeAddress(addr, &ip, &port, &bus_port)) {
fprintf(stderr, "Error: invalid CLUSTER NODES reply\n");
success = 0;
goto cleanup;
}
int myself = (strstr(flags, "myself") != NULL); int myself = (strstr(flags, "myself") != NULL);
clusterManagerNode *currentNode = NULL; clusterManagerNode *currentNode = NULL;
if (myself) { if (myself) {
/* bus-port could be wrong, correct it here, see clusterManagerNewNode. */
node->bus_port = bus_port;
node->flags |= CLUSTER_MANAGER_FLAG_MYSELF; node->flags |= CLUSTER_MANAGER_FLAG_MYSELF;
currentNode = node; currentNode = node;
clusterManagerNodeResetSlots(node); clusterManagerNodeResetSlots(node);
...@@ -4681,22 +4698,7 @@ static int clusterManagerNodeLoadInfo(clusterManagerNode *node, int opts, ...@@ -4681,22 +4698,7 @@ static int clusterManagerNodeLoadInfo(clusterManagerNode *node, int opts,
if (!(node->flags & CLUSTER_MANAGER_FLAG_MYSELF)) continue; if (!(node->flags & CLUSTER_MANAGER_FLAG_MYSELF)) continue;
else break; else break;
} else { } else {
if (addr == NULL) { currentNode = clusterManagerNewNode(sdsnew(ip), port, bus_port);
fprintf(stderr, "Error: invalid CLUSTER NODES reply\n");
success = 0;
goto cleanup;
}
char *c = strrchr(addr, '@');
if (c != NULL) *c = '\0';
c = strrchr(addr, ':');
if (c == NULL) {
fprintf(stderr, "Error: invalid CLUSTER NODES reply\n");
success = 0;
goto cleanup;
}
*c = '\0';
int port = atoi(++c);
currentNode = clusterManagerNewNode(sdsnew(addr), port);
currentNode->flags |= CLUSTER_MANAGER_FLAG_FRIEND; currentNode->flags |= CLUSTER_MANAGER_FLAG_FRIEND;
if (node->friends == NULL) node->friends = listCreate(); if (node->friends == NULL) node->friends = listCreate();
listAddNodeTail(node->friends, currentNode); listAddNodeTail(node->friends, currentNode);
...@@ -6110,17 +6112,14 @@ static int clusterManagerCommandCreate(int argc, char **argv) { ...@@ -6110,17 +6112,14 @@ static int clusterManagerCommandCreate(int argc, char **argv) {
cluster_manager.nodes = listCreate(); cluster_manager.nodes = listCreate();
for (i = 0; i < argc; i++) { for (i = 0; i < argc; i++) {
char *addr = argv[i]; char *addr = argv[i];
char *c = strrchr(addr, '@'); char *ip = NULL;
if (c != NULL) *c = '\0'; int port = 0;
c = strrchr(addr, ':'); if (!parseClusterNodeAddress(addr, &ip, &port, NULL)) {
if (c == NULL) {
fprintf(stderr, "Invalid address format: %s\n", addr); fprintf(stderr, "Invalid address format: %s\n", addr);
return 0; return 0;
} }
*c = '\0';
char *ip = addr; clusterManagerNode *node = clusterManagerNewNode(ip, port, 0);
int port = atoi(++c);
clusterManagerNode *node = clusterManagerNewNode(ip, port);
if (!clusterManagerNodeConnect(node)) { if (!clusterManagerNodeConnect(node)) {
freeClusterManagerNode(node); freeClusterManagerNode(node);
return 0; return 0;
...@@ -6327,8 +6326,16 @@ assign_replicas: ...@@ -6327,8 +6326,16 @@ assign_replicas:
continue; continue;
} }
redisReply *reply = NULL; redisReply *reply = NULL;
reply = CLUSTER_MANAGER_COMMAND(node, "cluster meet %s %d", if (first->bus_port == 0 || (first->bus_port == first->port + CLUSTER_MANAGER_PORT_INCR)) {
first_ip, first->port); /* CLUSTER MEET bus-port parameter was added in 4.0.
* So if (bus_port == 0) or (bus_port == port + CLUSTER_MANAGER_PORT_INCR),
* we just call CLUSTER MEET with 2 arguments, using the old form. */
reply = CLUSTER_MANAGER_COMMAND(node, "cluster meet %s %d",
first->ip, first->port);
} else {
reply = CLUSTER_MANAGER_COMMAND(node, "cluster meet %s %d %d",
first->ip, first->port, first->bus_port);
}
int is_err = 0; int is_err = 0;
if (reply != NULL) { if (reply != NULL) {
if ((is_err = reply->type == REDIS_REPLY_ERROR)) if ((is_err = reply->type == REDIS_REPLY_ERROR))
...@@ -6362,6 +6369,8 @@ assign_replicas: ...@@ -6362,6 +6369,8 @@ assign_replicas:
} }
success = 0; success = 0;
goto cleanup; goto cleanup;
} else if (err != NULL) {
zfree(err);
} }
} }
// Reset Nodes // Reset Nodes
...@@ -6405,7 +6414,7 @@ static int clusterManagerCommandAddNode(int argc, char **argv) { ...@@ -6405,7 +6414,7 @@ static int clusterManagerCommandAddNode(int argc, char **argv) {
clusterManagerLogInfo(">>> Adding node %s:%d to cluster %s:%d\n", ip, port, clusterManagerLogInfo(">>> Adding node %s:%d to cluster %s:%d\n", ip, port,
ref_ip, ref_port); ref_ip, ref_port);
// Check the existing cluster // Check the existing cluster
clusterManagerNode *refnode = clusterManagerNewNode(ref_ip, ref_port); clusterManagerNode *refnode = clusterManagerNewNode(ref_ip, ref_port, 0);
if (!clusterManagerLoadInfoFromNode(refnode)) return 0; if (!clusterManagerLoadInfoFromNode(refnode)) return 0;
if (!clusterManagerCheckCluster(0)) return 0; if (!clusterManagerCheckCluster(0)) return 0;
...@@ -6429,7 +6438,7 @@ static int clusterManagerCommandAddNode(int argc, char **argv) { ...@@ -6429,7 +6438,7 @@ static int clusterManagerCommandAddNode(int argc, char **argv) {
} }
// Add the new node // Add the new node
clusterManagerNode *new_node = clusterManagerNewNode(ip, port); clusterManagerNode *new_node = clusterManagerNewNode(ip, port, 0);
int added = 0; int added = 0;
if (!clusterManagerNodeConnect(new_node)) { if (!clusterManagerNodeConnect(new_node)) {
clusterManagerLogErr("[ERR] Sorry, can't connect to node %s:%d\n", clusterManagerLogErr("[ERR] Sorry, can't connect to node %s:%d\n",
...@@ -6507,8 +6516,18 @@ static int clusterManagerCommandAddNode(int argc, char **argv) { ...@@ -6507,8 +6516,18 @@ static int clusterManagerCommandAddNode(int argc, char **argv) {
success = 0; success = 0;
goto cleanup; goto cleanup;
} }
reply = CLUSTER_MANAGER_COMMAND(new_node, "CLUSTER MEET %s %d",
first_ip, first->port); if (first->bus_port == 0 || (first->bus_port == first->port + CLUSTER_MANAGER_PORT_INCR)) {
/* CLUSTER MEET bus-port parameter was added in 4.0.
* So if (bus_port == 0) or (bus_port == port + CLUSTER_MANAGER_PORT_INCR),
* we just call CLUSTER MEET with 2 arguments, using the old form. */
reply = CLUSTER_MANAGER_COMMAND(new_node, "CLUSTER MEET %s %d",
first_ip, first->port);
} else {
reply = CLUSTER_MANAGER_COMMAND(new_node, "CLUSTER MEET %s %d %d",
first->ip, first->port, first->bus_port);
}
if (!(success = clusterManagerCheckRedisReply(new_node, reply, NULL))) if (!(success = clusterManagerCheckRedisReply(new_node, reply, NULL)))
goto cleanup; goto cleanup;
...@@ -6545,7 +6564,7 @@ static int clusterManagerCommandDeleteNode(int argc, char **argv) { ...@@ -6545,7 +6564,7 @@ static int clusterManagerCommandDeleteNode(int argc, char **argv) {
char *node_id = argv[1]; char *node_id = argv[1];
clusterManagerLogInfo(">>> Removing node %s from cluster %s:%d\n", clusterManagerLogInfo(">>> Removing node %s from cluster %s:%d\n",
node_id, ip, port); node_id, ip, port);
clusterManagerNode *ref_node = clusterManagerNewNode(ip, port); clusterManagerNode *ref_node = clusterManagerNewNode(ip, port, 0);
clusterManagerNode *node = NULL; clusterManagerNode *node = NULL;
// Load cluster information // Load cluster information
...@@ -6607,7 +6626,7 @@ static int clusterManagerCommandInfo(int argc, char **argv) { ...@@ -6607,7 +6626,7 @@ static int clusterManagerCommandInfo(int argc, char **argv) {
int port = 0; int port = 0;
char *ip = NULL; char *ip = NULL;
if (!getClusterHostFromCmdArgs(argc, argv, &ip, &port)) goto invalid_args; if (!getClusterHostFromCmdArgs(argc, argv, &ip, &port)) goto invalid_args;
clusterManagerNode *node = clusterManagerNewNode(ip, port); clusterManagerNode *node = clusterManagerNewNode(ip, port, 0);
if (!clusterManagerLoadInfoFromNode(node)) return 0; if (!clusterManagerLoadInfoFromNode(node)) return 0;
clusterManagerShowClusterInfo(); clusterManagerShowClusterInfo();
return 1; return 1;
...@@ -6620,7 +6639,7 @@ static int clusterManagerCommandCheck(int argc, char **argv) { ...@@ -6620,7 +6639,7 @@ static int clusterManagerCommandCheck(int argc, char **argv) {
int port = 0; int port = 0;
char *ip = NULL; char *ip = NULL;
if (!getClusterHostFromCmdArgs(argc, argv, &ip, &port)) goto invalid_args; if (!getClusterHostFromCmdArgs(argc, argv, &ip, &port)) goto invalid_args;
clusterManagerNode *node = clusterManagerNewNode(ip, port); clusterManagerNode *node = clusterManagerNewNode(ip, port, 0);
if (!clusterManagerLoadInfoFromNode(node)) return 0; if (!clusterManagerLoadInfoFromNode(node)) return 0;
clusterManagerShowClusterInfo(); clusterManagerShowClusterInfo();
return clusterManagerCheckCluster(0); return clusterManagerCheckCluster(0);
...@@ -6638,7 +6657,7 @@ static int clusterManagerCommandReshard(int argc, char **argv) { ...@@ -6638,7 +6657,7 @@ static int clusterManagerCommandReshard(int argc, char **argv) {
int port = 0; int port = 0;
char *ip = NULL; char *ip = NULL;
if (!getClusterHostFromCmdArgs(argc, argv, &ip, &port)) goto invalid_args; if (!getClusterHostFromCmdArgs(argc, argv, &ip, &port)) goto invalid_args;
clusterManagerNode *node = clusterManagerNewNode(ip, port); clusterManagerNode *node = clusterManagerNewNode(ip, port, 0);
if (!clusterManagerLoadInfoFromNode(node)) return 0; if (!clusterManagerLoadInfoFromNode(node)) return 0;
clusterManagerCheckCluster(0); clusterManagerCheckCluster(0);
if (cluster_manager.errors && listLength(cluster_manager.errors) > 0) { if (cluster_manager.errors && listLength(cluster_manager.errors) > 0) {
...@@ -6827,7 +6846,7 @@ static int clusterManagerCommandRebalance(int argc, char **argv) { ...@@ -6827,7 +6846,7 @@ static int clusterManagerCommandRebalance(int argc, char **argv) {
clusterManagerNode **weightedNodes = NULL; clusterManagerNode **weightedNodes = NULL;
list *involved = NULL; list *involved = NULL;
if (!getClusterHostFromCmdArgs(argc, argv, &ip, &port)) goto invalid_args; if (!getClusterHostFromCmdArgs(argc, argv, &ip, &port)) goto invalid_args;
clusterManagerNode *node = clusterManagerNewNode(ip, port); clusterManagerNode *node = clusterManagerNewNode(ip, port, 0);
if (!clusterManagerLoadInfoFromNode(node)) return 0; if (!clusterManagerLoadInfoFromNode(node)) return 0;
int result = 1, i; int result = 1, i;
if (config.cluster_manager_command.weight != NULL) { if (config.cluster_manager_command.weight != NULL) {
...@@ -7028,7 +7047,7 @@ static int clusterManagerCommandSetTimeout(int argc, char **argv) { ...@@ -7028,7 +7047,7 @@ static int clusterManagerCommandSetTimeout(int argc, char **argv) {
return 0; return 0;
} }
// Load cluster information // Load cluster information
clusterManagerNode *node = clusterManagerNewNode(ip, port); clusterManagerNode *node = clusterManagerNewNode(ip, port, 0);
if (!clusterManagerLoadInfoFromNode(node)) return 0; if (!clusterManagerLoadInfoFromNode(node)) return 0;
int ok_count = 0, err_count = 0; int ok_count = 0, err_count = 0;
...@@ -7098,7 +7117,7 @@ static int clusterManagerCommandImport(int argc, char **argv) { ...@@ -7098,7 +7117,7 @@ static int clusterManagerCommandImport(int argc, char **argv) {
clusterManagerLogInfo(">>> Importing data from %s:%d to cluster %s:%d\n", clusterManagerLogInfo(">>> Importing data from %s:%d to cluster %s:%d\n",
src_ip, src_port, ip, port); src_ip, src_port, ip, port);
clusterManagerNode *refnode = clusterManagerNewNode(ip, port); clusterManagerNode *refnode = clusterManagerNewNode(ip, port, 0);
if (!clusterManagerLoadInfoFromNode(refnode)) return 0; if (!clusterManagerLoadInfoFromNode(refnode)) return 0;
if (!clusterManagerCheckCluster(0)) return 0; if (!clusterManagerCheckCluster(0)) return 0;
char *reply_err = NULL; char *reply_err = NULL;
...@@ -7233,7 +7252,7 @@ static int clusterManagerCommandCall(int argc, char **argv) { ...@@ -7233,7 +7252,7 @@ static int clusterManagerCommandCall(int argc, char **argv) {
int port = 0, i; int port = 0, i;
char *ip = NULL; char *ip = NULL;
if (!getClusterHostFromCmdArgs(1, argv, &ip, &port)) goto invalid_args; if (!getClusterHostFromCmdArgs(1, argv, &ip, &port)) goto invalid_args;
clusterManagerNode *refnode = clusterManagerNewNode(ip, port); clusterManagerNode *refnode = clusterManagerNewNode(ip, port, 0);
if (!clusterManagerLoadInfoFromNode(refnode)) return 0; if (!clusterManagerLoadInfoFromNode(refnode)) return 0;
argc--; argc--;
argv++; argv++;
...@@ -7278,7 +7297,7 @@ static int clusterManagerCommandBackup(int argc, char **argv) { ...@@ -7278,7 +7297,7 @@ static int clusterManagerCommandBackup(int argc, char **argv) {
int success = 1, port = 0; int success = 1, port = 0;
char *ip = NULL; char *ip = NULL;
if (!getClusterHostFromCmdArgs(1, argv, &ip, &port)) goto invalid_args; if (!getClusterHostFromCmdArgs(1, argv, &ip, &port)) goto invalid_args;
clusterManagerNode *refnode = clusterManagerNewNode(ip, port); clusterManagerNode *refnode = clusterManagerNewNode(ip, port, 0);
if (!clusterManagerLoadInfoFromNode(refnode)) return 0; if (!clusterManagerLoadInfoFromNode(refnode)) return 0;
int no_issues = clusterManagerCheckCluster(0); int no_issues = clusterManagerCheckCluster(0);
int cluster_errors_count = (no_issues ? 0 : int cluster_errors_count = (no_issues ? 0 :
...@@ -7489,7 +7508,7 @@ struct distsamples { ...@@ -7489,7 +7508,7 @@ struct distsamples {
* of the collected samples targeting an xterm 256 terminal. * of the collected samples targeting an xterm 256 terminal.
* *
* Takes an array of distsamples structures, ordered from smaller to bigger * Takes an array of distsamples structures, ordered from smaller to bigger
* 'max' value. Last sample max must be 0, to mean that it olds all the * 'max' value. Last sample max must be 0, to mean that it holds all the
* samples greater than the previous one, and is also the stop sentinel. * samples greater than the previous one, and is also the stop sentinel.
* *
* "tot' is the total number of samples in the different buckets, so it * "tot' is the total number of samples in the different buckets, so it
......
...@@ -43,7 +43,7 @@ ...@@ -43,7 +43,7 @@
#define assert(_e) (likely((_e))?(void)0 : (_serverAssert(#_e,__FILE__,__LINE__),redis_unreachable())) #define assert(_e) (likely((_e))?(void)0 : (_serverAssert(#_e,__FILE__,__LINE__),redis_unreachable()))
#define panic(...) _serverPanic(__FILE__,__LINE__,__VA_ARGS__),redis_unreachable() #define panic(...) _serverPanic(__FILE__,__LINE__,__VA_ARGS__),redis_unreachable()
void _serverAssert(char *estr, char *file, int line); void _serverAssert(const char *estr, const char *file, int line);
void _serverPanic(const char *file, int line, const char *msg, ...); void _serverPanic(const char *file, int line, const char *msg, ...);
#endif #endif
...@@ -656,6 +656,8 @@ typedef struct RedisModuleClientInfo { ...@@ -656,6 +656,8 @@ typedef struct RedisModuleClientInfo {
#define RedisModuleClientInfo RedisModuleClientInfoV1 #define RedisModuleClientInfo RedisModuleClientInfoV1
#define REDISMODULE_CLIENTINFO_INITIALIZER_V1 { .version = 1 }
#define REDISMODULE_REPLICATIONINFO_VERSION 1 #define REDISMODULE_REPLICATIONINFO_VERSION 1
typedef struct RedisModuleReplicationInfo { typedef struct RedisModuleReplicationInfo {
uint64_t version; /* Not used since this structure is never passed uint64_t version; /* Not used since this structure is never passed
...@@ -915,6 +917,7 @@ REDISMODULE_API size_t (*RedisModule_CallReplyLength)(RedisModuleCallReply *repl ...@@ -915,6 +917,7 @@ REDISMODULE_API size_t (*RedisModule_CallReplyLength)(RedisModuleCallReply *repl
REDISMODULE_API RedisModuleCallReply * (*RedisModule_CallReplyArrayElement)(RedisModuleCallReply *reply, size_t idx) REDISMODULE_ATTR; REDISMODULE_API RedisModuleCallReply * (*RedisModule_CallReplyArrayElement)(RedisModuleCallReply *reply, size_t idx) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleString * (*RedisModule_CreateString)(RedisModuleCtx *ctx, const char *ptr, size_t len) REDISMODULE_ATTR; REDISMODULE_API RedisModuleString * (*RedisModule_CreateString)(RedisModuleCtx *ctx, const char *ptr, size_t len) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringFromLongLong)(RedisModuleCtx *ctx, long long ll) REDISMODULE_ATTR; REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringFromLongLong)(RedisModuleCtx *ctx, long long ll) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringFromULongLong)(RedisModuleCtx *ctx, unsigned long long ull) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringFromDouble)(RedisModuleCtx *ctx, double d) REDISMODULE_ATTR; REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringFromDouble)(RedisModuleCtx *ctx, double d) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringFromLongDouble)(RedisModuleCtx *ctx, long double ld, int humanfriendly) REDISMODULE_ATTR; REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringFromLongDouble)(RedisModuleCtx *ctx, long double ld, int humanfriendly) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringFromString)(RedisModuleCtx *ctx, const RedisModuleString *str) REDISMODULE_ATTR; REDISMODULE_API RedisModuleString * (*RedisModule_CreateStringFromString)(RedisModuleCtx *ctx, const RedisModuleString *str) REDISMODULE_ATTR;
...@@ -948,6 +951,7 @@ REDISMODULE_API int (*RedisModule_ReplyWithDouble)(RedisModuleCtx *ctx, double d ...@@ -948,6 +951,7 @@ REDISMODULE_API int (*RedisModule_ReplyWithDouble)(RedisModuleCtx *ctx, double d
REDISMODULE_API int (*RedisModule_ReplyWithBigNumber)(RedisModuleCtx *ctx, const char *bignum, size_t len) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_ReplyWithBigNumber)(RedisModuleCtx *ctx, const char *bignum, size_t len) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_ReplyWithCallReply)(RedisModuleCtx *ctx, RedisModuleCallReply *reply) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_ReplyWithCallReply)(RedisModuleCtx *ctx, RedisModuleCallReply *reply) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_StringToLongLong)(const RedisModuleString *str, long long *ll) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_StringToLongLong)(const RedisModuleString *str, long long *ll) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_StringToULongLong)(const RedisModuleString *str, unsigned long long *ull) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_StringToDouble)(const RedisModuleString *str, double *d) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_StringToDouble)(const RedisModuleString *str, double *d) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_StringToLongDouble)(const RedisModuleString *str, long double *d) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_StringToLongDouble)(const RedisModuleString *str, long double *d) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_StringToStreamID)(const RedisModuleString *str, RedisModuleStreamID *id) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_StringToStreamID)(const RedisModuleString *str, RedisModuleStreamID *id) REDISMODULE_ATTR;
...@@ -1000,6 +1004,8 @@ REDISMODULE_API void (*RedisModule_ChannelAtPosWithFlags)(RedisModuleCtx *ctx, i ...@@ -1000,6 +1004,8 @@ REDISMODULE_API void (*RedisModule_ChannelAtPosWithFlags)(RedisModuleCtx *ctx, i
REDISMODULE_API unsigned long long (*RedisModule_GetClientId)(RedisModuleCtx *ctx) REDISMODULE_ATTR; REDISMODULE_API unsigned long long (*RedisModule_GetClientId)(RedisModuleCtx *ctx) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleString * (*RedisModule_GetClientUserNameById)(RedisModuleCtx *ctx, uint64_t id) REDISMODULE_ATTR; REDISMODULE_API RedisModuleString * (*RedisModule_GetClientUserNameById)(RedisModuleCtx *ctx, uint64_t id) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_GetClientInfoById)(void *ci, uint64_t id) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_GetClientInfoById)(void *ci, uint64_t id) REDISMODULE_ATTR;
REDISMODULE_API RedisModuleString * (*RedisModule_GetClientNameById)(RedisModuleCtx *ctx, uint64_t id) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_SetClientNameById)(uint64_t id, RedisModuleString *name) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_PublishMessage)(RedisModuleCtx *ctx, RedisModuleString *channel, RedisModuleString *message) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_PublishMessage)(RedisModuleCtx *ctx, RedisModuleString *channel, RedisModuleString *message) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_PublishMessageShard)(RedisModuleCtx *ctx, RedisModuleString *channel, RedisModuleString *message) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_PublishMessageShard)(RedisModuleCtx *ctx, RedisModuleString *channel, RedisModuleString *message) REDISMODULE_ATTR;
REDISMODULE_API int (*RedisModule_GetContextFlags)(RedisModuleCtx *ctx) REDISMODULE_ATTR; REDISMODULE_API int (*RedisModule_GetContextFlags)(RedisModuleCtx *ctx) REDISMODULE_ATTR;
...@@ -1258,6 +1264,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -1258,6 +1264,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(ListInsert); REDISMODULE_GET_API(ListInsert);
REDISMODULE_GET_API(ListDelete); REDISMODULE_GET_API(ListDelete);
REDISMODULE_GET_API(StringToLongLong); REDISMODULE_GET_API(StringToLongLong);
REDISMODULE_GET_API(StringToULongLong);
REDISMODULE_GET_API(StringToDouble); REDISMODULE_GET_API(StringToDouble);
REDISMODULE_GET_API(StringToLongDouble); REDISMODULE_GET_API(StringToLongDouble);
REDISMODULE_GET_API(StringToStreamID); REDISMODULE_GET_API(StringToStreamID);
...@@ -1280,6 +1287,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -1280,6 +1287,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(CreateStringFromCallReply); REDISMODULE_GET_API(CreateStringFromCallReply);
REDISMODULE_GET_API(CreateString); REDISMODULE_GET_API(CreateString);
REDISMODULE_GET_API(CreateStringFromLongLong); REDISMODULE_GET_API(CreateStringFromLongLong);
REDISMODULE_GET_API(CreateStringFromULongLong);
REDISMODULE_GET_API(CreateStringFromDouble); REDISMODULE_GET_API(CreateStringFromDouble);
REDISMODULE_GET_API(CreateStringFromLongDouble); REDISMODULE_GET_API(CreateStringFromLongDouble);
REDISMODULE_GET_API(CreateStringFromString); REDISMODULE_GET_API(CreateStringFromString);
...@@ -1426,6 +1434,8 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int ...@@ -1426,6 +1434,8 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API(ServerInfoGetFieldUnsigned); REDISMODULE_GET_API(ServerInfoGetFieldUnsigned);
REDISMODULE_GET_API(ServerInfoGetFieldDouble); REDISMODULE_GET_API(ServerInfoGetFieldDouble);
REDISMODULE_GET_API(GetClientInfoById); REDISMODULE_GET_API(GetClientInfoById);
REDISMODULE_GET_API(GetClientNameById);
REDISMODULE_GET_API(SetClientNameById);
REDISMODULE_GET_API(PublishMessage); REDISMODULE_GET_API(PublishMessage);
REDISMODULE_GET_API(PublishMessageShard); REDISMODULE_GET_API(PublishMessageShard);
REDISMODULE_GET_API(SubscribeToServerEvent); REDISMODULE_GET_API(SubscribeToServerEvent);
......
...@@ -2151,6 +2151,16 @@ void readSyncBulkPayload(connection *conn) { ...@@ -2151,6 +2151,16 @@ void readSyncBulkPayload(connection *conn) {
/* Close old rdb asynchronously. */ /* Close old rdb asynchronously. */
if (old_rdb_fd != -1) bioCreateCloseJob(old_rdb_fd); if (old_rdb_fd != -1) bioCreateCloseJob(old_rdb_fd);
/* Sync the directory to ensure rename is persisted */
if (fsyncFileDir(server.rdb_filename) == -1) {
serverLog(LL_WARNING,
"Failed trying to sync DB directory %s in "
"MASTER <-> REPLICA synchronization: %s",
server.rdb_filename, strerror(errno));
cancelReplicationHandshake(1);
return;
}
if (rdbLoad(server.rdb_filename,&rsi,RDBFLAGS_REPLICATION) != C_OK) { if (rdbLoad(server.rdb_filename,&rsi,RDBFLAGS_REPLICATION) != C_OK) {
serverLog(LL_WARNING, serverLog(LL_WARNING,
"Failed trying to load the MASTER synchronization " "Failed trying to load the MASTER synchronization "
...@@ -3310,7 +3320,7 @@ void replicationDiscardCachedMaster(void) { ...@@ -3310,7 +3320,7 @@ void replicationDiscardCachedMaster(void) {
* passed as argument as the socket for the new master. * passed as argument as the socket for the new master.
* *
* This function is called when successfully setup a partial resynchronization * This function is called when successfully setup a partial resynchronization
* so the stream of data that we'll receive will start from were this * so the stream of data that we'll receive will start from where this
* master left. */ * master left. */
void replicationResurrectCachedMaster(connection *conn) { void replicationResurrectCachedMaster(connection *conn) {
server.master = server.cached_master; server.master = server.cached_master;
......
...@@ -363,6 +363,11 @@ static int scriptVerifyWriteCommandAllow(scriptRunCtx *run_ctx, char **err) { ...@@ -363,6 +363,11 @@ static int scriptVerifyWriteCommandAllow(scriptRunCtx *run_ctx, char **err) {
if (!(run_ctx->c->cmd->flags & CMD_WRITE)) if (!(run_ctx->c->cmd->flags & CMD_WRITE))
return C_OK; return C_OK;
/* If the script already made a modification to the dataset, we can't
* fail it on unpredictable error state. */
if ((run_ctx->flags & SCRIPT_WRITE_DIRTY))
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. */
......
...@@ -48,6 +48,7 @@ ...@@ -48,6 +48,7 @@
#include <arpa/inet.h> #include <arpa/inet.h>
#include <sys/stat.h> #include <sys/stat.h>
#include <fcntl.h> #include <fcntl.h>
#include <sys/file.h>
#include <sys/time.h> #include <sys/time.h>
#include <sys/resource.h> #include <sys/resource.h>
#include <sys/uio.h> #include <sys/uio.h>
...@@ -634,7 +635,7 @@ void resetChildState() { ...@@ -634,7 +635,7 @@ void resetChildState() {
NULL); NULL);
} }
/* Return if child type is mutual exclusive with other fork children */ /* Return if child type is mutually exclusive with other fork children */
int isMutuallyExclusiveChildType(int type) { int isMutuallyExclusiveChildType(int type) {
return type == CHILD_TYPE_RDB || type == CHILD_TYPE_AOF || type == CHILD_TYPE_MODULE; return type == CHILD_TYPE_RDB || type == CHILD_TYPE_AOF || type == CHILD_TYPE_MODULE;
} }
...@@ -2793,8 +2794,23 @@ int populateArgsStructure(struct redisCommandArg *args) { ...@@ -2793,8 +2794,23 @@ int populateArgsStructure(struct redisCommandArg *args) {
return count; return count;
} }
/* Recursively populate the command structure. */ /* Recursively populate the command structure.
void populateCommandStructure(struct redisCommand *c) { *
* On success, the function return C_OK. Otherwise C_ERR is returned and we won't
* add this command in the commands dict. */
int populateCommandStructure(struct redisCommand *c) {
/* If the command marks with CMD_SENTINEL, it exists in sentinel. */
if (!(c->flags & CMD_SENTINEL) && server.sentinel_mode)
return C_ERR;
/* If the command marks with CMD_ONLY_SENTINEL, it only exists in sentinel. */
if (c->flags & CMD_ONLY_SENTINEL && !server.sentinel_mode)
return C_ERR;
/* Translate the command string flags description into an actual
* set of flags. */
setImplicitACLCategories(c);
/* Redis commands don't need more args than STATIC_KEY_SPECS_NUM (Number of keys /* Redis commands don't need more args than STATIC_KEY_SPECS_NUM (Number of keys
* specs can be greater than STATIC_KEY_SPECS_NUM only for module commands) */ * specs can be greater than STATIC_KEY_SPECS_NUM only for module commands) */
c->key_specs = c->key_specs_static; c->key_specs = c->key_specs_static;
...@@ -2828,14 +2844,15 @@ void populateCommandStructure(struct redisCommand *c) { ...@@ -2828,14 +2844,15 @@ void populateCommandStructure(struct redisCommand *c) {
for (int j = 0; c->subcommands[j].declared_name; j++) { for (int j = 0; c->subcommands[j].declared_name; j++) {
struct redisCommand *sub = c->subcommands+j; struct redisCommand *sub = c->subcommands+j;
/* Translate the command string flags description into an actual
* set of flags. */
setImplicitACLCategories(sub);
sub->fullname = catSubCommandFullname(c->declared_name, sub->declared_name); sub->fullname = catSubCommandFullname(c->declared_name, sub->declared_name);
populateCommandStructure(sub); if (populateCommandStructure(sub) == C_ERR)
continue;
commandAddSubcommand(c, sub, sub->declared_name); commandAddSubcommand(c, sub, sub->declared_name);
} }
} }
return C_OK;
} }
extern struct redisCommand redisCommandTable[]; extern struct redisCommand redisCommandTable[];
...@@ -2853,16 +2870,9 @@ void populateCommandTable(void) { ...@@ -2853,16 +2870,9 @@ void populateCommandTable(void) {
int retval1, retval2; int retval1, retval2;
setImplicitACLCategories(c);
if (!(c->flags & CMD_SENTINEL) && server.sentinel_mode)
continue;
if (c->flags & CMD_ONLY_SENTINEL && !server.sentinel_mode)
continue;
c->fullname = sdsnew(c->declared_name); c->fullname = sdsnew(c->declared_name);
populateCommandStructure(c); if (populateCommandStructure(c) == C_ERR)
continue;
retval1 = dictAdd(server.commands, sdsdup(c->fullname), c); retval1 = dictAdd(server.commands, sdsdup(c->fullname), c);
/* Populate an additional dictionary that will be unaffected /* Populate an additional dictionary that will be unaffected
...@@ -3584,8 +3594,8 @@ int commandCheckArity(client *c, sds *err) { ...@@ -3584,8 +3594,8 @@ int commandCheckArity(client *c, sds *err) {
* if C_ERR is returned the client was destroyed (i.e. after QUIT). */ * if C_ERR is returned the client was destroyed (i.e. after QUIT). */
int processCommand(client *c) { int processCommand(client *c) {
if (!scriptIsTimedout()) { if (!scriptIsTimedout()) {
/* Both EXEC and EVAL call call() directly so there should be /* Both EXEC and scripts call call() directly so there should be
* no way in_exec or in_eval is 1. * no way in_exec or scriptIsRunning() is 1.
* That is unless lua_timedout, in which case client may run * That is unless lua_timedout, in which case client may run
* some commands. */ * some commands. */
serverAssert(!server.in_exec); serverAssert(!server.in_exec);
...@@ -4207,6 +4217,12 @@ int finishShutdown(void) { ...@@ -4207,6 +4217,12 @@ int finishShutdown(void) {
/* Close the listening sockets. Apparently this allows faster restarts. */ /* Close the listening sockets. Apparently this allows faster restarts. */
closeListeningSockets(1); closeListeningSockets(1);
/* Unlock the cluster config file before shutdown */
if (server.cluster_enabled && server.cluster_config_file_lock_fd != -1) {
flock(server.cluster_config_file_lock_fd, LOCK_UN|LOCK_NB);
}
serverLog(LL_WARNING,"%s is now ready to exit, bye bye...", serverLog(LL_WARNING,"%s is now ready to exit, bye bye...",
server.sentinel_mode ? "Sentinel" : "Redis"); server.sentinel_mode ? "Sentinel" : "Redis");
return C_OK; return C_OK;
...@@ -5645,6 +5661,7 @@ sds genRedisInfoString(dict *section_dict, int all_sections, int everything) { ...@@ -5645,6 +5661,7 @@ sds genRedisInfoString(dict *section_dict, int all_sections, int everything) {
"keyspace_misses:%lld\r\n" "keyspace_misses:%lld\r\n"
"pubsub_channels:%ld\r\n" "pubsub_channels:%ld\r\n"
"pubsub_patterns:%lu\r\n" "pubsub_patterns:%lu\r\n"
"pubsubshard_channels:%lu\r\n"
"latest_fork_usec:%lld\r\n" "latest_fork_usec:%lld\r\n"
"total_forks:%lld\r\n" "total_forks:%lld\r\n"
"migrate_cached_sockets:%ld\r\n" "migrate_cached_sockets:%ld\r\n"
...@@ -5694,6 +5711,7 @@ sds genRedisInfoString(dict *section_dict, int all_sections, int everything) { ...@@ -5694,6 +5711,7 @@ sds genRedisInfoString(dict *section_dict, int all_sections, int everything) {
server.stat_keyspace_misses, server.stat_keyspace_misses,
dictSize(server.pubsub_channels), dictSize(server.pubsub_channels),
dictSize(server.pubsub_patterns), dictSize(server.pubsub_patterns),
dictSize(server.pubsubshard_channels),
server.stat_fork_time, server.stat_fork_time,
server.stat_total_forks, server.stat_total_forks,
dictSize(server.migrate_cached_sockets), dictSize(server.migrate_cached_sockets),
...@@ -6094,9 +6112,12 @@ void usage(void) { ...@@ -6094,9 +6112,12 @@ void usage(void) {
fprintf(stderr," ./redis-server - (read config from stdin)\n"); fprintf(stderr," ./redis-server - (read config from stdin)\n");
fprintf(stderr," ./redis-server -v or --version\n"); fprintf(stderr," ./redis-server -v or --version\n");
fprintf(stderr," ./redis-server -h or --help\n"); fprintf(stderr," ./redis-server -h or --help\n");
fprintf(stderr," ./redis-server --test-memory <megabytes>\n\n"); fprintf(stderr," ./redis-server --test-memory <megabytes>\n");
fprintf(stderr," ./redis-server --check-system\n");
fprintf(stderr,"\n");
fprintf(stderr,"Examples:\n"); fprintf(stderr,"Examples:\n");
fprintf(stderr," ./redis-server (run the server with default conf)\n"); fprintf(stderr," ./redis-server (run the server with default conf)\n");
fprintf(stderr," echo 'maxmemory 128mb' | ./redis-server -\n");
fprintf(stderr," ./redis-server /etc/redis/6379.conf\n"); fprintf(stderr," ./redis-server /etc/redis/6379.conf\n");
fprintf(stderr," ./redis-server --port 7777\n"); fprintf(stderr," ./redis-server --port 7777\n");
fprintf(stderr," ./redis-server --port 7777 --replicaof 127.0.0.1 8888\n"); fprintf(stderr," ./redis-server --port 7777 --replicaof 127.0.0.1 8888\n");
...@@ -6347,7 +6368,7 @@ int redisFork(int purpose) { ...@@ -6347,7 +6368,7 @@ int redisFork(int purpose) {
server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */ server.stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / server.stat_fork_time / (1024*1024*1024); /* GB per second. */
latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000); latencyAddSampleIfNeeded("fork",server.stat_fork_time/1000);
/* The child_pid and child_type are only for mutual exclusive children. /* The child_pid and child_type are only for mutually exclusive children.
* other child types should handle and store their pid's in dedicated variables. * other child types should handle and store their pid's in dedicated variables.
* *
* Today, we allows CHILD_TYPE_LDB to run in parallel with the other fork types: * Today, we allows CHILD_TYPE_LDB to run in parallel with the other fork types:
...@@ -6433,7 +6454,7 @@ void dismissClientMemory(client *c) { ...@@ -6433,7 +6454,7 @@ void dismissClientMemory(client *c) {
/* In the child process, we don't need some buffers anymore, and these are /* In the child process, we don't need some buffers anymore, and these are
* likely to change in the parent when there's heavy write traffic. * likely to change in the parent when there's heavy write traffic.
* We dismis them right away, to avoid CoW. * We dismiss them right away, to avoid CoW.
* see dismissMemeory(). */ * see dismissMemeory(). */
void dismissMemoryInChild(void) { void dismissMemoryInChild(void) {
/* madvise(MADV_DONTNEED) may not work if Transparent Huge Pages is enabled. */ /* madvise(MADV_DONTNEED) may not work if Transparent Huge Pages is enabled. */
...@@ -6880,6 +6901,8 @@ int main(int argc, char **argv) { ...@@ -6880,6 +6901,8 @@ int main(int argc, char **argv) {
server.exec_argv[1] = zstrdup(server.configfile); server.exec_argv[1] = zstrdup(server.configfile);
j = 2; // Skip this arg when parsing options j = 2; // Skip this arg when parsing options
} }
sds *argv_tmp;
int argc_tmp;
int handled_last_config_arg = 1; int handled_last_config_arg = 1;
while(j < argc) { while(j < argc) {
/* Either first or last argument - Should we read config from stdin? */ /* Either first or last argument - Should we read config from stdin? */
...@@ -6897,7 +6920,37 @@ int main(int argc, char **argv) { ...@@ -6897,7 +6920,37 @@ int main(int argc, char **argv) {
/* argv[j]+2 for removing the preceding `--` */ /* argv[j]+2 for removing the preceding `--` */
options = sdscat(options,argv[j]+2); options = sdscat(options,argv[j]+2);
options = sdscat(options," "); options = sdscat(options," ");
handled_last_config_arg = 0;
argv_tmp = sdssplitargs(argv[j], &argc_tmp);
if (argc_tmp == 1) {
/* Means that we only have one option name, like --port or "--port " */
handled_last_config_arg = 0;
if ((j != argc-1) && argv[j+1][0] == '-' && argv[j+1][1] == '-' &&
!strcasecmp(argv[j], "--save"))
{
/* Special case: handle some things like `--save --config value`.
* In this case, if next argument starts with `--`, we will reset
* handled_last_config_arg flag and append an empty "" config value
* to the options, so it will become `--save "" --config value`.
* We are doing it to be compatible with pre 7.0 behavior (which we
* break it in #10660, 7.0.1), since there might be users who generate
* a command line from an array and when it's empty that's what they produce. */
options = sdscat(options, "\"\"");
handled_last_config_arg = 1;
}
else if ((j == argc-1) && !strcasecmp(argv[j], "--save")) {
/* Special case: when empty save is the last argument.
* In this case, we append an empty "" config value to the options,
* so it will become `--save ""` and will follow the same reset thing. */
options = sdscat(options, "\"\"");
}
} else {
/* Means that we are passing both config name and it's value in the same arg,
* like "--port 6380", so we need to reset handled_last_config_arg flag. */
handled_last_config_arg = 1;
}
sdsfreesplitres(argv_tmp, argc_tmp);
} else { } else {
/* Option argument */ /* Option argument */
options = sdscatrepr(options,argv[j],strlen(argv[j])); options = sdscatrepr(options,argv[j],strlen(argv[j]));
......
...@@ -959,6 +959,7 @@ typedef struct multiState { ...@@ -959,6 +959,7 @@ typedef struct multiState {
is possible to know if all the commands have a is possible to know if all the commands have a
certain flag. */ certain flag. */
size_t argv_len_sums; /* mem used by all commands arguments */ size_t argv_len_sums; /* mem used by all commands arguments */
int alloc_count; /* total number of multiCmd struct memory reserved. */
} multiState; } multiState;
/* This structure holds the blocking operation state for a client. /* This structure holds the blocking operation state for a client.
...@@ -1857,7 +1858,7 @@ struct redisServer { ...@@ -1857,7 +1858,7 @@ struct redisServer {
dict *pubsub_patterns; /* A dict of pubsub_patterns */ dict *pubsub_patterns; /* A dict of pubsub_patterns */
int notify_keyspace_events; /* Events to propagate via Pub/Sub. This is an int notify_keyspace_events; /* Events to propagate via Pub/Sub. This is an
xor of NOTIFY_... flags. */ xor of NOTIFY_... flags. */
dict *pubsubshard_channels; /* Map channels to list of subscribed clients */ dict *pubsubshard_channels; /* Map shard channels to list of subscribed clients */
/* Cluster */ /* Cluster */
int cluster_enabled; /* Is cluster enabled? */ int cluster_enabled; /* Is cluster enabled? */
int cluster_port; /* Set the cluster port for a node. */ int cluster_port; /* Set the cluster port for a node. */
...@@ -2186,13 +2187,14 @@ typedef int redisGetKeysProc(struct redisCommand *cmd, robj **argv, int argc, ge ...@@ -2186,13 +2187,14 @@ typedef int redisGetKeysProc(struct redisCommand *cmd, robj **argv, int argc, ge
* or may just execute read commands. A command can not be marked * or may just execute read commands. A command can not be marked
* both CMD_WRITE and CMD_MAY_REPLICATE * both CMD_WRITE and CMD_MAY_REPLICATE
* *
* CMD_SENTINEL: This command is present in sentinel mode too. * CMD_SENTINEL: This command is present in sentinel mode.
* *
* CMD_SENTINEL_ONLY: This command is present only when in sentinel mode. * CMD_ONLY_SENTINEL: This command is present only when in sentinel mode.
* And should be removed from redis.
* *
* CMD_NO_MANDATORY_KEYS: This key arguments for this command are optional. * CMD_NO_MANDATORY_KEYS: This key arguments for this command are optional.
* *
* CMD_NO_MULTI: The command is nt allowed inside a transaction * CMD_NO_MULTI: The command is not allowed inside a transaction
* *
* The following additional flags are only used in order to put commands * The following additional flags are only used in order to put commands
* in a specific ACL category. Commands can have multiple ACL categories. * in a specific ACL category. Commands can have multiple ACL categories.
...@@ -2242,7 +2244,6 @@ struct redisCommand { ...@@ -2242,7 +2244,6 @@ struct redisCommand {
struct redisCommandArg *args; struct redisCommandArg *args;
/* Runtime populated data */ /* Runtime populated data */
/* What keys should be loaded in background when calling this command? */
long long microseconds, calls, rejected_calls, failed_calls; long long microseconds, calls, rejected_calls, failed_calls;
int id; /* Command ID. This is a progressive ID starting from 0 that int id; /* Command ID. This is a progressive ID starting from 0 that
is assigned at runtime, and is used in order to check is assigned at runtime, and is used in order to check
...@@ -2498,6 +2499,7 @@ char *getClientPeerId(client *client); ...@@ -2498,6 +2499,7 @@ char *getClientPeerId(client *client);
char *getClientSockName(client *client); char *getClientSockName(client *client);
sds catClientInfoString(sds s, client *client); sds catClientInfoString(sds s, client *client);
sds getAllClientsInfoString(int type); sds getAllClientsInfoString(int type);
int clientSetName(client *c, robj *name);
void rewriteClientCommandVector(client *c, int argc, ...); void rewriteClientCommandVector(client *c, int argc, ...);
void rewriteClientCommandArgument(client *c, int i, robj *newval); void rewriteClientCommandArgument(client *c, int i, robj *newval);
void replaceClientCommandVector(client *c, int argc, robj **argv); void replaceClientCommandVector(client *c, int argc, robj **argv);
...@@ -2607,7 +2609,6 @@ void decrRefCount(robj *o); ...@@ -2607,7 +2609,6 @@ void decrRefCount(robj *o);
void decrRefCountVoid(void *o); void decrRefCountVoid(void *o);
void incrRefCount(robj *o); void incrRefCount(robj *o);
robj *makeObjectShared(robj *o); robj *makeObjectShared(robj *o);
robj *resetRefCount(robj *obj);
void freeStringObject(robj *o); void freeStringObject(robj *o);
void freeListObject(robj *o); void freeListObject(robj *o);
void freeSetObject(robj *o); void freeSetObject(robj *o);
...@@ -2752,7 +2753,7 @@ void sendChildInfo(childInfoType info_type, size_t keys, char *pname); ...@@ -2752,7 +2753,7 @@ void sendChildInfo(childInfoType info_type, size_t keys, char *pname);
void receiveChildInfo(void); void receiveChildInfo(void);
/* Fork helpers */ /* Fork helpers */
int redisFork(int type); int redisFork(int purpose);
int hasActiveChildProcess(); int hasActiveChildProcess();
void resetChildState(); void resetChildState();
int isMutuallyExclusiveChildType(int type); int isMutuallyExclusiveChildType(int type);
...@@ -2998,6 +2999,7 @@ int pubsubPublishMessageAndPropagateToCluster(robj *channel, robj *message, int ...@@ -2998,6 +2999,7 @@ int pubsubPublishMessageAndPropagateToCluster(robj *channel, robj *message, int
void addReplyPubsubMessage(client *c, robj *channel, robj *msg, robj *message_bulk); void addReplyPubsubMessage(client *c, robj *channel, robj *msg, robj *message_bulk);
int serverPubsubSubscriptionCount(); int serverPubsubSubscriptionCount();
int serverPubsubShardSubscriptionCount(); int serverPubsubShardSubscriptionCount();
size_t pubsubMemOverhead(client *c);
/* Keyspace events notification */ /* Keyspace events notification */
void notifyKeyspaceEvent(int type, char *event, robj *key, int dbid); void notifyKeyspaceEvent(int type, char *event, robj *key, int dbid);
......
...@@ -150,7 +150,7 @@ int checkOvercommit(sds *error_msg) { ...@@ -150,7 +150,7 @@ int checkOvercommit(sds *error_msg) {
} }
fclose(fp); fclose(fp);
if (atoi(buf)) { if (strtol(buf, NULL, 10) == 0) {
*error_msg = sdsnew( *error_msg = sdsnew(
"overcommit_memory is set to 0! Background save may fail under low memory condition. " "overcommit_memory is set to 0! Background save may fail under low memory condition. "
"To fix this issue add 'vm.overcommit_memory = 1' to /etc/sysctl.conf and then reboot or run the " "To fix this issue add 'vm.overcommit_memory = 1' to /etc/sysctl.conf and then reboot or run the "
......
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