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

Merge branch 'unstable' into fixChildInfoPipeFdLeak

parents 440385de cfdc800a
/*
* Copyright (c) 2018, Salvatore Sanfilippo <antirez at gmail dot com>
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
*
* * Redistributions of source code must retain the above copyright notice,
* this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* * Neither the name of Redis nor the names of its contributors may be used
* to endorse or promote products derived from this software without
* specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGE.
*/
#include "server.h"
/* =============================================================================
* Global state for ACLs
* ==========================================================================*/
rax *Users; /* Table mapping usernames to user structures. */
user *DefaultUser; /* Global reference to the default user.
Every new connection is associated to it, if no
AUTH or HELLO is used to authenticate with a
different user. */
/* =============================================================================
* Helper functions for the rest of the ACL implementation
* ==========================================================================*/
/* Return zero if strings are the same, non-zero if they are not.
* The comparison is performed in a way that prevents an attacker to obtain
* information about the nature of the strings just monitoring the execution
* time of the function.
*
* Note that limiting the comparison length to strings up to 512 bytes we
* can avoid leaking any information about the password length and any
* possible branch misprediction related leak.
*/
int time_independent_strcmp(char *a, char *b) {
char bufa[CONFIG_AUTHPASS_MAX_LEN], bufb[CONFIG_AUTHPASS_MAX_LEN];
/* The above two strlen perform len(a) + len(b) operations where either
* a or b are fixed (our password) length, and the difference is only
* relative to the length of the user provided string, so no information
* leak is possible in the following two lines of code. */
unsigned int alen = strlen(a);
unsigned int blen = strlen(b);
unsigned int j;
int diff = 0;
/* We can't compare strings longer than our static buffers.
* Note that this will never pass the first test in practical circumstances
* so there is no info leak. */
if (alen > sizeof(bufa) || blen > sizeof(bufb)) return 1;
memset(bufa,0,sizeof(bufa)); /* Constant time. */
memset(bufb,0,sizeof(bufb)); /* Constant time. */
/* Again the time of the following two copies is proportional to
* len(a) + len(b) so no info is leaked. */
memcpy(bufa,a,alen);
memcpy(bufb,b,blen);
/* Always compare all the chars in the two buffers without
* conditional expressions. */
for (j = 0; j < sizeof(bufa); j++) {
diff |= (bufa[j] ^ bufb[j]);
}
/* Length must be equal as well. */
diff |= alen ^ blen;
return diff; /* If zero strings are the same. */
}
/* =============================================================================
* Low level ACL API
* ==========================================================================*/
/* Method for passwords/pattern comparison used for the user->passwords list
* so that we can search for items with listSearchKey(). */
int ACLListMatchSds(void *a, void *b) {
return sdscmp(a,b) == 0;
}
/* Create a new user with the specified name, store it in the list
* of users (the Users global radix tree), and returns a reference to
* the structure representing the user.
*
* If the user with such name already exists NULL is returned. */
user *ACLCreateUser(const char *name, size_t namelen) {
if (raxFind(Users,(unsigned char*)name,namelen) != raxNotFound) return NULL;
user *u = zmalloc(sizeof(*u));
u->name = sdsnewlen(name,namelen);
u->flags = 0;
u->allowed_subcommands = NULL;
u->passwords = listCreate();
u->patterns = listCreate();
listSetMatchMethod(u->passwords,ACLListMatchSds);
listSetMatchMethod(u->patterns,ACLListMatchSds);
memset(u->allowed_commands,0,sizeof(u->allowed_commands));
raxInsert(Users,(unsigned char*)name,namelen,u,NULL);
return u;
}
/* Set user properties according to the string "op". The following
* is a description of what different strings will do:
*
* on Enable the user: it is possible to authenticate as this user.
* off Disable the user: it's no longer possible to authenticate
* with this user, however the already authenticated connections
* will still work.
* +<command> Allow the execution of that command
* -<command> Disallow the execution of that command
* +@<category> Allow the execution of all the commands in such category
* with valid categories being @set, @sortedset, @list, @hash,
* @string, @bitmap, @hyperloglog,
* @stream, @admin, @readonly,
* @readwrite, @fast, @slow,
* @pubsub.
* The special category @all means all the commands.
* +<command>|subcommand Allow a specific subcommand of an otherwise
* disabled command. Note that this form is not
* allowed as negative like -DEBUG|SEGFAULT, but
* only additive starting with "+".
* ~<pattern> Add a pattern of keys that can be mentioned as part of
* commands. For instance ~* allows all the keys. The pattern
* is a glob-style pattern like the one of KEYS.
* It is possible to specify multiple patterns.
* ><password> Add this passowrd to the list of valid password for the user.
* For example >mypass will add "mypass" to the list.
* This directive clears the "nopass" flag (see later).
* <<password> Remove this password from the list of valid passwords.
* nopass All the set passwords of the user are removed, and the user
* is flagged as requiring no password: it means that every
* password will work against this user. If this directive is
* used for the default user, every new connection will be
* immediately authenticated with the default user without
* any explicit AUTH command required. Note that the "resetpass"
* directive will clear this condition.
* allcommands Alias for +@all
* allkeys Alias for ~*
* resetpass Flush the list of allowed passwords. Moreover removes the
* "nopass" status. After "resetpass" the user has no associated
* passwords and there is no way to authenticate without adding
* some password (or setting it as "nopass" later).
* resetkeys Flush the list of allowed keys patterns.
* reset Performs the following actions: resetpass, resetkeys, off,
* -@all. The user returns to the same state it has immediately
* after its creation.
*
* The 'op' string must be null terminated. The 'oplen' argument should
* specify the length of the 'op' string in case the caller requires to pass
* binary data (for instance the >password form may use a binary password).
* Otherwise the field can be set to -1 and the function will use strlen()
* to determine the length.
*
* The function returns C_OK if the action to perform was understood because
* the 'op' string made sense. Otherwise C_ERR is returned if the operation
* is unknown or has some syntax error.
*/
int ACLSetUser(user *u, const char *op, ssize_t oplen) {
if (oplen == -1) oplen = strlen(op);
if (!strcasecmp(op,"on")) {
u->flags |= USER_FLAG_ENABLED;
} else if (!strcasecmp(op,"off")) {
u->flags &= ~USER_FLAG_ENABLED;
} else if (!strcasecmp(op,"allkeys") ||
!strcasecmp(op,"~*"))
{
u->flags |= USER_FLAG_ALLKEYS;
if (u->patterns) listEmpty(u->patterns);
} else if (!strcasecmp(op,"allcommands") ||
!strcasecmp(op,"+@all"))
{
memset(u->allowed_commands,255,sizeof(u->allowed_commands));
u->flags |= USER_FLAG_ALLCOMMANDS;
} else if (!strcasecmp(op,"nopass")) {
u->flags |= USER_FLAG_NOPASS;
listEmpty(u->passwords);
} else if (!strcasecmp(op,"resetpass")) {
u->flags &= ~USER_FLAG_NOPASS;
listEmpty(u->passwords);
} else if (op[0] == '>') {
sds newpass = sdsnewlen(op+1,oplen-1);
listNode *ln = listSearchKey(u->passwords,newpass);
/* Avoid re-adding the same password multiple times. */
if (ln == NULL) listAddNodeTail(u->passwords,newpass);
u->flags &= ~USER_FLAG_NOPASS;
} else if (op[0] == '<') {
sds delpass = sdsnewlen(op+1,oplen-1);
listNode *ln = listSearchKey(u->passwords,delpass);
if (ln) listDelNode(u->passwords,ln);
sdsfree(delpass);
} else if (op[0] == '~') {
sds newpat = sdsnewlen(op+1,oplen-1);
listNode *ln = listSearchKey(u->patterns,newpat);
/* Avoid re-adding the same pattern multiple times. */
if (ln == NULL) listAddNodeTail(u->patterns,newpat);
u->flags &= ~USER_FLAG_ALLKEYS;
} else {
return C_ERR;
}
return C_OK;
}
/* Return the first password of the default user or NULL.
* This function is needed for backward compatibility with the old
* directive "requirepass" when Redis supported a single global
* password. */
sds ACLDefaultUserFirstPassword(void) {
if (listLength(DefaultUser->passwords) == 0) return NULL;
listNode *first = listFirst(DefaultUser->passwords);
return listNodeValue(first);
}
/* Initialization of the ACL subsystem. */
void ACLInit(void) {
Users = raxNew();
DefaultUser = ACLCreateUser("default",7);
ACLSetUser(DefaultUser,"+@all",-1);
ACLSetUser(DefaultUser,"~*",-1);
ACLSetUser(DefaultUser,"on",-1);
ACLSetUser(DefaultUser,"nopass",-1);
}
/* Check the username and password pair and return C_OK if they are valid,
* otherwise C_ERR is returned and errno is set to:
*
* EINVAL: if the username-password do not match.
* ENONENT: if the specified user does not exist at all.
*/
int ACLCheckUserCredentials(robj *username, robj *password) {
user *u = ACLGetUserByName(username->ptr,sdslen(username->ptr));
if (u == NULL) {
errno = ENOENT;
return C_ERR;
}
/* Disabled users can't login. */
if ((u->flags & USER_FLAG_ENABLED) == 0) {
errno = EINVAL;
return C_ERR;
}
/* If the user is configured to don't require any password, we
* are already fine here. */
if (u->flags & USER_FLAG_NOPASS) return C_OK;
/* Check all the user passwords for at least one to match. */
listIter li;
listNode *ln;
listRewind(u->passwords,&li);
while((ln = listNext(&li))) {
sds thispass = listNodeValue(ln);
if (!time_independent_strcmp(password->ptr, thispass))
return C_OK;
}
/* If we reached this point, no password matched. */
errno = EINVAL;
return C_ERR;
}
/* For ACL purposes, every user has a bitmap with the commands that such
* user is allowed to execute. In order to populate the bitmap, every command
* should have an assigned ID (that is used to index the bitmap). This function
* creates such an ID: it uses sequential IDs, reusing the same ID for the same
* command name, so that a command retains the same ID in case of modules that
* are unloaded and later reloaded. */
unsigned long ACLGetCommandID(const char *cmdname) {
static rax *map = NULL;
static unsigned long nextid = 0;
if (map == NULL) map = raxNew();
void *id = raxFind(map,(unsigned char*)cmdname,strlen(cmdname));
if (id != raxNotFound) return (unsigned long)id;
raxInsert(map,(unsigned char*)cmdname,strlen(cmdname),(void*)nextid,NULL);
return nextid++;
}
/* Return an username by its name, or NULL if the user does not exist. */
user *ACLGetUserByName(const char *name, size_t namelen) {
void *myuser = raxFind(Users,(unsigned char*)name,namelen);
if (myuser == raxNotFound) return NULL;
return myuser;
}
/* Check if the command ready to be excuted in the client 'c', and already
* referenced by c->cmd, can be executed by this client according to the
* ACls associated to the client user c->user.
*
* If the user can execute the command ACL_OK is returned, otherwise
* ACL_DENIED_CMD or ACL_DENIED_KEY is returned: the first in case the
* command cannot be executed because the user is not allowed to run such
* command, the second if the command is denied because the user is trying
* to access keys that are not among the specified patterns. */
int ACLCheckCommandPerm(client *c) {
user *u = c->user;
uint64_t id = c->cmd->id;
/* If there is no associated user, the connection can run anything. */
if (u == NULL) return ACL_OK;
/* We have to deny every command with an ID that overflows the Redis
* internal structures. Very unlikely to happen. */
if (c->cmd->id >= USER_MAX_COMMAND_BIT) return ACL_DENIED_CMD;
/* Check if the user can execute this command. */
if (!(u->flags & USER_FLAG_ALLCOMMANDS) &&
c->cmd->proc != authCommand)
{
uint64_t wordid = id / sizeof(u->allowed_commands[0]) / 8;
uint64_t bit = 1 << (id % (sizeof(u->allowed_commands[0] * 8)));
/* If the bit is not set we have to check further, in case the
* command is allowed just with that specific subcommand. */
if (!(u->allowed_commands[wordid] & bit)) {
/* Check if the subcommand matches. */
if (u->allowed_subcommands == NULL || c->argc < 2)
return ACL_DENIED_CMD;
long subid = 0;
while (1) {
if (u->allowed_subcommands[id][subid] == NULL)
return ACL_DENIED_CMD;
if (!strcasecmp(c->argv[1]->ptr,
u->allowed_subcommands[id][subid]))
break; /* Subcommand match found. Stop here. */
subid++;
}
}
}
/* Check if the user can execute commands explicitly touching the keys
* mentioned in the command arguments. */
if (!(c->user->flags & USER_FLAG_ALLKEYS) &&
(c->cmd->getkeys_proc || c->cmd->firstkey))
{
int numkeys;
int *keyidx = getKeysFromCommand(c->cmd,c->argv,c->argc,&numkeys);
for (int j = 0; j < numkeys; j++) {
listIter li;
listNode *ln;
listRewind(u->passwords,&li);
/* Test this key against every pattern. */
int match = 0;
while((ln = listNext(&li))) {
sds pattern = listNodeValue(ln);
size_t plen = sdslen(pattern);
int idx = keyidx[j];
if (stringmatchlen(pattern,plen,c->argv[idx]->ptr,
sdslen(c->argv[idx]->ptr),0))
{
match = 1;
break;
}
}
if (!match) return ACL_DENIED_KEY;
}
getKeysFreeResult(keyidx);
}
/* If we survived all the above checks, the user can execute the
* command. */
return ACL_OK;
}
/* =============================================================================
* ACL related commands
* ==========================================================================*/
/* ACL -- show and modify the configuration of ACL users.
* ACL HELP
* ACL LIST
* ACL SETUSER <username> ... user attribs ...
* ACL DELUSER <username>
* ACL GETUSER <username>
*/
void aclCommand(client *c) {
char *sub = c->argv[1]->ptr;
if (!strcasecmp(sub,"setuser") && c->argc >= 3) {
sds username = c->argv[2]->ptr;
user *u = ACLGetUserByName(username,sdslen(username));
if (!u) u = ACLCreateUser(username,sdslen(username));
serverAssert(u != NULL);
for (int j = 3; j < c->argc; j++) {
if (ACLSetUser(u,c->argv[j]->ptr,sdslen(c->argv[j]->ptr)) != C_OK) {
addReplyErrorFormat(c,
"Syntax error in ACL SETUSER modifier '%s'",
c->argv[j]->ptr);
return;
}
}
addReply(c,shared.ok);
} else if (!strcasecmp(sub,"whoami")) {
if (c->user != NULL) {
addReplyBulkCBuffer(c,c->user->name,sdslen(c->user->name));
} else {
addReplyNull(c);
}
} else if (!strcasecmp(sub,"getuser") && c->argc == 3) {
user *u = ACLGetUserByName(c->argv[2]->ptr,sdslen(c->argv[2]->ptr));
addReplyMapLen(c,2);
/* Flags */
addReplyBulkCString(c,"flags");
void *deflen = addReplyDeferredLen(c);
int numflags = 0;
if (u->flags & USER_FLAG_ENABLED) {
addReplyBulkCString(c,"on");
numflags++;
} else {
addReplyBulkCString(c,"off");
numflags++;
}
if (u->flags & USER_FLAG_ALLKEYS) {
addReplyBulkCString(c,"allkeys");
numflags++;
}
if (u->flags & USER_FLAG_ALLCOMMANDS) {
addReplyBulkCString(c,"allcommnads");
numflags++;
}
if (u->flags & USER_FLAG_NOPASS) {
addReplyBulkCString(c,"nopass");
numflags++;
}
setDeferredSetLen(c,deflen,numflags);
/* Passwords */
addReplyBulkCString(c,"passwords");
addReplyArrayLen(c,listLength(u->passwords));
listIter li;
listNode *ln;
listRewind(u->passwords,&li);
while((ln = listNext(&li))) {
sds thispass = listNodeValue(ln);
addReplyBulkCBuffer(c,thispass,sdslen(thispass));
}
} else if (!strcasecmp(sub,"help")) {
const char *help[] = {
"LIST -- List all the registered users.",
"SETUSER <username> [attribs ...] -- Create or modify a user.",
"DELUSER <username> -- Delete a user.",
"GETUSER <username> -- Get the user details.",
"WHOAMI -- Return the current username.",
NULL
};
addReplyHelp(c,help);
} else {
addReplySubcommandSyntaxError(c);
}
}
......@@ -351,8 +351,8 @@ static int processTimeEvents(aeEventLoop *eventLoop) {
* if flags has AE_FILE_EVENTS set, file events are processed.
* if flags has AE_TIME_EVENTS set, time events are processed.
* if flags has AE_DONT_WAIT set the function returns ASAP until all
* if flags has AE_CALL_AFTER_SLEEP set, the aftersleep callback is called.
* the events that's possible to process without to wait are processed.
* if flags has AE_CALL_AFTER_SLEEP set, the aftersleep callback is called.
*
* The function returns the number of events processed. */
int aeProcessEvents(aeEventLoop *eventLoop, int flags)
......
......@@ -222,6 +222,7 @@ static void killAppendOnlyChild(void) {
/* Close pipes used for IPC between the two processes. */
aofClosePipes();
closeChildInfoPipe();
updateDictResizePolicy();
}
/* Called when the user switches from "appendonly yes" to "appendonly no"
......@@ -646,6 +647,8 @@ struct client *createFakeClient(void) {
c->obuf_soft_limit_reached_time = 0;
c->watched_keys = listCreate();
c->peerid = NULL;
c->resp = 2;
c->user = NULL;
listSetFreeMethod(c->reply,freeClientReplyValue);
listSetDupMethod(c->reply,dupClientReplyValue);
initClientMultiState(c);
......
......@@ -1002,7 +1002,7 @@ void bitfieldCommand(client *c) {
highest_write_offset)) == NULL) return;
}
addReplyMultiBulkLen(c,numops);
addReplyArrayLen(c,numops);
/* Actually process the operations. */
for (j = 0; j < numops; j++) {
......@@ -1047,7 +1047,7 @@ void bitfieldCommand(client *c) {
setSignedBitfield(o->ptr,thisop->offset,
thisop->bits,newval);
} else {
addReply(c,shared.nullbulk);
addReplyNull(c);
}
} else {
uint64_t oldval, newval, wrapped, retval;
......@@ -1076,7 +1076,7 @@ void bitfieldCommand(client *c) {
setUnsignedBitfield(o->ptr,thisop->offset,
thisop->bits,newval);
} else {
addReply(c,shared.nullbulk);
addReplyNull(c);
}
}
changes++;
......
......@@ -187,7 +187,7 @@ void replyToBlockedClientTimedOut(client *c) {
if (c->btype == BLOCKED_LIST ||
c->btype == BLOCKED_ZSET ||
c->btype == BLOCKED_STREAM) {
addReply(c,shared.nullmultibulk);
addReplyNullArray(c);
} else if (c->btype == BLOCKED_WAIT) {
addReplyLongLong(c,replicationCountAcksByOffset(c->bpop.reploffset));
} else if (c->btype == BLOCKED_MODULE) {
......@@ -436,8 +436,12 @@ void handleClientsBlockedOnKeys(void) {
* the name of the stream and the data we
* extracted from it. Wrapped in a single-item
* array, since we have just one key. */
addReplyMultiBulkLen(receiver,1);
addReplyMultiBulkLen(receiver,2);
if (receiver->resp == 2) {
addReplyArrayLen(receiver,1);
addReplyArrayLen(receiver,2);
} else {
addReplyMapLen(receiver,1);
}
addReplyBulk(receiver,rl->key);
streamPropInfo pi = {
......
......@@ -4126,7 +4126,7 @@ void clusterReplyMultiBulkSlots(client *c) {
*/
int num_masters = 0;
void *slot_replylen = addDeferredMultiBulkLength(c);
void *slot_replylen = addReplyDeferredLen(c);
dictEntry *de;
dictIterator *di = dictGetSafeIterator(server.cluster->nodes);
......@@ -4146,7 +4146,7 @@ void clusterReplyMultiBulkSlots(client *c) {
}
if (start != -1 && (!bit || j == CLUSTER_SLOTS-1)) {
int nested_elements = 3; /* slots (2) + master addr (1). */
void *nested_replylen = addDeferredMultiBulkLength(c);
void *nested_replylen = addReplyDeferredLen(c);
if (bit && j == CLUSTER_SLOTS-1) j++;
......@@ -4162,7 +4162,7 @@ void clusterReplyMultiBulkSlots(client *c) {
start = -1;
/* First node reply position is always the master */
addReplyMultiBulkLen(c, 3);
addReplyArrayLen(c, 3);
addReplyBulkCString(c, node->ip);
addReplyLongLong(c, node->port);
addReplyBulkCBuffer(c, node->name, CLUSTER_NAMELEN);
......@@ -4172,19 +4172,19 @@ void clusterReplyMultiBulkSlots(client *c) {
/* This loop is copy/pasted from clusterGenNodeDescription()
* with modifications for per-slot node aggregation */
if (nodeFailed(node->slaves[i])) continue;
addReplyMultiBulkLen(c, 3);
addReplyArrayLen(c, 3);
addReplyBulkCString(c, node->slaves[i]->ip);
addReplyLongLong(c, node->slaves[i]->port);
addReplyBulkCBuffer(c, node->slaves[i]->name, CLUSTER_NAMELEN);
nested_elements++;
}
setDeferredMultiBulkLength(c, nested_replylen, nested_elements);
setDeferredArrayLen(c, nested_replylen, nested_elements);
num_masters++;
}
}
}
dictReleaseIterator(di);
setDeferredMultiBulkLength(c, slot_replylen, num_masters);
setDeferredArrayLen(c, slot_replylen, num_masters);
}
void clusterCommand(client *c) {
......@@ -4548,7 +4548,7 @@ NULL
keys = zmalloc(sizeof(robj*)*maxkeys);
numkeys = getKeysInSlot(slot, keys, maxkeys);
addReplyMultiBulkLen(c,numkeys);
addReplyArrayLen(c,numkeys);
for (j = 0; j < numkeys; j++) {
addReplyBulk(c,keys[j]);
decrRefCount(keys[j]);
......@@ -4627,7 +4627,7 @@ NULL
return;
}
addReplyMultiBulkLen(c,n->numslaves);
addReplyArrayLen(c,n->numslaves);
for (j = 0; j < n->numslaves; j++) {
sds ni = clusterGenNodeDescription(n->slaves[j]);
addReplyBulkCString(c,ni);
......@@ -4836,7 +4836,7 @@ void dumpCommand(client *c) {
/* Check if the key is here. */
if ((o = lookupKeyRead(c->db,c->argv[1])) == NULL) {
addReply(c,shared.nullbulk);
addReplyNull(c);
return;
}
......
......@@ -531,7 +531,12 @@ void loadServerConfigFromString(char *config) {
err = "Password is longer than CONFIG_AUTHPASS_MAX_LEN";
goto loaderr;
}
server.requirepass = argv[1][0] ? zstrdup(argv[1]) : NULL;
/* The old "requirepass" directive just translates to setting
* a password to the default user. */
ACLSetUser(DefaultUser,"resetpass",-1);
sds aclop = sdscatprintf(sdsempty(),">%s",argv[1]);
ACLSetUser(DefaultUser,aclop,sdslen(aclop));
sdsfree(aclop);
} else if (!strcasecmp(argv[0],"pidfile") && argc == 2) {
zfree(server.pidfile);
server.pidfile = zstrdup(argv[1]);
......@@ -919,8 +924,12 @@ void configSetCommand(client *c) {
server.rdb_filename = zstrdup(o->ptr);
} config_set_special_field("requirepass") {
if (sdslen(o->ptr) > CONFIG_AUTHPASS_MAX_LEN) goto badfmt;
zfree(server.requirepass);
server.requirepass = ((char*)o->ptr)[0] ? zstrdup(o->ptr) : NULL;
/* The old "requirepass" directive just translates to setting
* a password to the default user. */
ACLSetUser(DefaultUser,"resetpass",-1);
sds aclop = sdscatprintf(sdsempty(),">%s",o->ptr);
ACLSetUser(DefaultUser,aclop,sdslen(aclop));
sdsfree(aclop);
} config_set_special_field("masterauth") {
zfree(server.masterauth);
server.masterauth = ((char*)o->ptr)[0] ? zstrdup(o->ptr) : NULL;
......@@ -1324,7 +1333,7 @@ badfmt: /* Bad format errors */
void configGetCommand(client *c) {
robj *o = c->argv[2];
void *replylen = addDeferredMultiBulkLength(c);
void *replylen = addReplyDeferredLen(c);
char *pattern = o->ptr;
char buf[128];
int matches = 0;
......@@ -1332,7 +1341,6 @@ void configGetCommand(client *c) {
/* String values */
config_get_string_field("dbfilename",server.rdb_filename);
config_get_string_field("requirepass",server.requirepass);
config_get_string_field("masterauth",server.masterauth);
config_get_string_field("cluster-announce-ip",server.cluster_announce_ip);
config_get_string_field("unixsocket",server.unixsocket);
......@@ -1571,7 +1579,17 @@ void configGetCommand(client *c) {
sdsfree(aux);
matches++;
}
setDeferredMultiBulkLength(c,replylen,matches*2);
if (stringmatch(pattern,"requirepass",1)) {
addReplyBulkCString(c,"requirepass");
sds password = ACLDefaultUserFirstPassword();
if (password) {
addReplyBulkCBuffer(c,password,sdslen(password));
} else {
addReplyBulkCString(c,"");
}
matches++;
}
setDeferredMapLen(c,replylen,matches);
}
/*-----------------------------------------------------------------------------
......@@ -1981,6 +1999,26 @@ void rewriteConfigBindOption(struct rewriteConfigState *state) {
rewriteConfigRewriteLine(state,option,line,force);
}
/* Rewrite the requirepass option. */
void rewriteConfigRequirepassOption(struct rewriteConfigState *state, char *option) {
int force = 1;
sds line;
sds password = ACLDefaultUserFirstPassword();
/* If there is no password set, we don't want the requirepass option
* to be present in the configuration at all. */
if (password == NULL) {
rewriteConfigMarkAsProcessed(state,option);
return;
}
line = sdsnew(option);
line = sdscatlen(line, " ", 1);
line = sdscatsds(line, password);
rewriteConfigRewriteLine(state,option,line,force);
}
/* Glue together the configuration lines in the current configuration
* rewrite state into a single string, stripping multiple empty lines. */
sds rewriteConfigGetContentFromState(struct rewriteConfigState *state) {
......@@ -2161,7 +2199,7 @@ int rewriteConfig(char *path) {
rewriteConfigNumericalOption(state,"replica-priority",server.slave_priority,CONFIG_DEFAULT_SLAVE_PRIORITY);
rewriteConfigNumericalOption(state,"min-replicas-to-write",server.repl_min_slaves_to_write,CONFIG_DEFAULT_MIN_SLAVES_TO_WRITE);
rewriteConfigNumericalOption(state,"min-replicas-max-lag",server.repl_min_slaves_max_lag,CONFIG_DEFAULT_MIN_SLAVES_MAX_LAG);
rewriteConfigStringOption(state,"requirepass",server.requirepass,NULL);
rewriteConfigRequirepassOption(state,"requirepass");
rewriteConfigNumericalOption(state,"maxclients",server.maxclients,CONFIG_DEFAULT_MAX_CLIENTS);
rewriteConfigBytesOption(state,"maxmemory",server.maxmemory,CONFIG_DEFAULT_MAXMEMORY);
rewriteConfigBytesOption(state,"proto-max-bulk-len",server.proto_max_bulk_len,CONFIG_DEFAULT_PROTO_MAX_BULK_LEN);
......
......@@ -452,6 +452,7 @@ void flushallCommand(client *c) {
kill(server.rdb_child_pid,SIGUSR1);
rdbRemoveTempFile(server.rdb_child_pid);
closeChildInfoPipe();
updateDictResizePolicy();
}
if (server.saveparamslen > 0) {
/* Normally rdbSave() will reset dirty, but we don't want this here
......@@ -526,7 +527,7 @@ void randomkeyCommand(client *c) {
robj *key;
if ((key = dbRandomKey(c->db)) == NULL) {
addReply(c,shared.nullbulk);
addReplyNull(c);
return;
}
......@@ -540,7 +541,7 @@ void keysCommand(client *c) {
sds pattern = c->argv[1]->ptr;
int plen = sdslen(pattern), allkeys;
unsigned long numkeys = 0;
void *replylen = addDeferredMultiBulkLength(c);
void *replylen = addReplyDeferredLen(c);
di = dictGetSafeIterator(c->db->dict);
allkeys = (pattern[0] == '*' && pattern[1] == '\0');
......@@ -558,7 +559,7 @@ void keysCommand(client *c) {
}
}
dictReleaseIterator(di);
setDeferredMultiBulkLength(c,replylen,numkeys);
setDeferredArrayLen(c,replylen,numkeys);
}
/* This callback is used by scanGenericCommand in order to collect elements
......@@ -783,10 +784,10 @@ void scanGenericCommand(client *c, robj *o, unsigned long cursor) {
}
/* Step 4: Reply to the client. */
addReplyMultiBulkLen(c, 2);
addReplyArrayLen(c, 2);
addReplyBulkLongLong(c,cursor);
addReplyMultiBulkLen(c, listLength(keys));
addReplyArrayLen(c, listLength(keys));
while ((node = listFirst(keys)) != NULL) {
robj *kobj = listNodeValue(node);
addReplyBulk(c, kobj);
......
......@@ -517,7 +517,7 @@ NULL
sdsfree(d);
} else if (!strcasecmp(c->argv[1]->ptr,"digest-value") && c->argc >= 2) {
/* DEBUG DIGEST-VALUE key key key ... key. */
addReplyMultiBulkLen(c,c->argc-2);
addReplyArrayLen(c,c->argc-2);
for (int j = 2; j < c->argc; j++) {
unsigned char digest[20];
memset(digest,0,20); /* Start with a clean result */
......@@ -529,6 +529,58 @@ NULL
addReplyStatus(c,d);
sdsfree(d);
}
} else if (!strcasecmp(c->argv[1]->ptr,"protocol") && c->argc == 3) {
/* DEBUG PROTOCOL [string|integer|double|bignum|null|array|set|map|
* attrib|push|verbatim|true|false|state|err|bloberr] */
char *name = c->argv[2]->ptr;
if (!strcasecmp(name,"string")) {
addReplyBulkCString(c,"Hello World");
} else if (!strcasecmp(name,"integer")) {
addReplyLongLong(c,12345);
} else if (!strcasecmp(name,"double")) {
addReplyDouble(c,3.14159265359);
} else if (!strcasecmp(name,"bignum")) {
addReplyProto(c,"(1234567999999999999999999999999999999\r\n",40);
} else if (!strcasecmp(name,"null")) {
addReplyNull(c);
} else if (!strcasecmp(name,"array")) {
addReplyArrayLen(c,3);
for (int j = 0; j < 3; j++) addReplyLongLong(c,j);
} else if (!strcasecmp(name,"set")) {
addReplySetLen(c,3);
for (int j = 0; j < 3; j++) addReplyLongLong(c,j);
} else if (!strcasecmp(name,"map")) {
addReplyMapLen(c,3);
for (int j = 0; j < 3; j++) {
addReplyLongLong(c,j);
addReplyBool(c, j == 1);
}
} else if (!strcasecmp(name,"attrib")) {
addReplyAttributeLen(c,1);
addReplyBulkCString(c,"key-popularity");
addReplyArrayLen(c,2);
addReplyBulkCString(c,"key:123");
addReplyLongLong(c,90);
/* Attributes are not real replies, so a well formed reply should
* also have a normal reply type after the attribute. */
addReplyBulkCString(c,"Some real reply following the attribute");
} else if (!strcasecmp(name,"push")) {
addReplyPushLen(c,2);
addReplyBulkCString(c,"server-cpu-usage");
addReplyLongLong(c,42);
/* Push replies are not synchronous replies, so we emit also a
* normal reply in order for blocking clients just discarding the
* push reply, to actually consume the reply and continue. */
addReplyBulkCString(c,"Some real reply following the push reply");
} else if (!strcasecmp(name,"true")) {
addReplyBool(c,1);
} else if (!strcasecmp(name,"false")) {
addReplyBool(c,0);
} else if (!strcasecmp(name,"verbatim")) {
addReplyVerbatim(c,"This is a verbatim\nstring",25,"txt");
} else {
addReplyError(c,"Wrong protocol type name. Please use one of the following: string|integer|double|bignum|null|array|set|map|attrib|push|verbatim|true|false|state|err|bloberr");
}
} else if (!strcasecmp(c->argv[1]->ptr,"sleep") && c->argc == 3) {
double dtime = strtod(c->argv[2]->ptr,NULL);
long long utime = dtime*1000000;
......
......@@ -466,7 +466,7 @@ void georadiusGeneric(client *c, int flags) {
/* Look up the requested zset */
robj *zobj = NULL;
if ((zobj = lookupKeyReadOrReply(c, key, shared.emptymultibulk)) == NULL ||
if ((zobj = lookupKeyReadOrReply(c, key, shared.null[c->resp])) == NULL ||
checkType(c, zobj, OBJ_ZSET)) {
return;
}
......@@ -566,7 +566,7 @@ void georadiusGeneric(client *c, int flags) {
/* If no matching results, the user gets an empty reply. */
if (ga->used == 0 && storekey == NULL) {
addReply(c, shared.emptymultibulk);
addReplyNull(c);
geoArrayFree(ga);
return;
}
......@@ -597,11 +597,11 @@ void georadiusGeneric(client *c, int flags) {
if (withhash)
option_length++;
/* The multibulk len we send is exactly result_length. The result is
/* The array len we send is exactly result_length. The result is
* either all strings of just zset members *or* a nested multi-bulk
* reply containing the zset member string _and_ all the additional
* options the user enabled for this request. */
addReplyMultiBulkLen(c, returned_items);
addReplyArrayLen(c, returned_items);
/* Finally send results back to the caller */
int i;
......@@ -613,7 +613,7 @@ void georadiusGeneric(client *c, int flags) {
* as a nested multi-bulk. Add 1 to account for result value
* itself. */
if (option_length)
addReplyMultiBulkLen(c, option_length + 1);
addReplyArrayLen(c, option_length + 1);
addReplyBulkSds(c,gp->member);
gp->member = NULL;
......@@ -625,7 +625,7 @@ void georadiusGeneric(client *c, int flags) {
addReplyLongLong(c, gp->score);
if (withcoords) {
addReplyMultiBulkLen(c, 2);
addReplyArrayLen(c, 2);
addReplyHumanLongDouble(c, gp->longitude);
addReplyHumanLongDouble(c, gp->latitude);
}
......@@ -706,11 +706,11 @@ void geohashCommand(client *c) {
/* Geohash elements one after the other, using a null bulk reply for
* missing elements. */
addReplyMultiBulkLen(c,c->argc-2);
addReplyArrayLen(c,c->argc-2);
for (j = 2; j < c->argc; j++) {
double score;
if (!zobj || zsetScore(zobj, c->argv[j]->ptr, &score) == C_ERR) {
addReply(c,shared.nullbulk);
addReplyNull(c);
} else {
/* The internal format we use for geocoding is a bit different
* than the standard, since we use as initial latitude range
......@@ -721,7 +721,7 @@ void geohashCommand(client *c) {
/* Decode... */
double xy[2];
if (!decodeGeohash(score,xy)) {
addReply(c,shared.nullbulk);
addReplyNull(c);
continue;
}
......@@ -759,19 +759,19 @@ void geoposCommand(client *c) {
/* Report elements one after the other, using a null bulk reply for
* missing elements. */
addReplyMultiBulkLen(c,c->argc-2);
addReplyArrayLen(c,c->argc-2);
for (j = 2; j < c->argc; j++) {
double score;
if (!zobj || zsetScore(zobj, c->argv[j]->ptr, &score) == C_ERR) {
addReply(c,shared.nullmultibulk);
addReplyNullArray(c);
} else {
/* Decode... */
double xy[2];
if (!decodeGeohash(score,xy)) {
addReply(c,shared.nullmultibulk);
addReplyNullArray(c);
continue;
}
addReplyMultiBulkLen(c,2);
addReplyArrayLen(c,2);
addReplyHumanLongDouble(c,xy[0]);
addReplyHumanLongDouble(c,xy[1]);
}
......@@ -797,7 +797,7 @@ void geodistCommand(client *c) {
/* Look up the requested zset */
robj *zobj = NULL;
if ((zobj = lookupKeyReadOrReply(c, c->argv[1], shared.nullbulk))
if ((zobj = lookupKeyReadOrReply(c, c->argv[1], shared.null[c->resp]))
== NULL || checkType(c, zobj, OBJ_ZSET)) return;
/* Get the scores. We need both otherwise NULL is returned. */
......@@ -805,13 +805,13 @@ void geodistCommand(client *c) {
if (zsetScore(zobj, c->argv[2]->ptr, &score1) == C_ERR ||
zsetScore(zobj, c->argv[3]->ptr, &score2) == C_ERR)
{
addReply(c,shared.nullbulk);
addReplyNull(c);
return;
}
/* Decode & compute the distance. */
if (!decodeGeohash(score1,xyxy) || !decodeGeohash(score2,xyxy+2))
addReply(c,shared.nullbulk);
addReplyNull(c);
else
addReplyDoubleDistance(c,
geohashGetDistance(xyxy[0],xyxy[1],xyxy[2],xyxy[3]) / to_meter);
......
......@@ -1512,7 +1512,7 @@ void pfdebugCommand(client *c) {
}
hdr = o->ptr;
addReplyMultiBulkLen(c,HLL_REGISTERS);
addReplyArrayLen(c,HLL_REGISTERS);
for (j = 0; j < HLL_REGISTERS; j++) {
uint8_t val;
......
......@@ -476,19 +476,19 @@ sds createLatencyReport(void) {
/* latencyCommand() helper to produce a time-delay reply for all the samples
* in memory for the specified time series. */
void latencyCommandReplyWithSamples(client *c, struct latencyTimeSeries *ts) {
void *replylen = addDeferredMultiBulkLength(c);
void *replylen = addReplyDeferredLen(c);
int samples = 0, j;
for (j = 0; j < LATENCY_TS_LEN; j++) {
int i = (ts->idx + j) % LATENCY_TS_LEN;
if (ts->samples[i].time == 0) continue;
addReplyMultiBulkLen(c,2);
addReplyArrayLen(c,2);
addReplyLongLong(c,ts->samples[i].time);
addReplyLongLong(c,ts->samples[i].latency);
samples++;
}
setDeferredMultiBulkLength(c,replylen,samples);
setDeferredArrayLen(c,replylen,samples);
}
/* latencyCommand() helper to produce the reply for the LATEST subcommand,
......@@ -497,14 +497,14 @@ void latencyCommandReplyWithLatestEvents(client *c) {
dictIterator *di;
dictEntry *de;
addReplyMultiBulkLen(c,dictSize(server.latency_events));
addReplyArrayLen(c,dictSize(server.latency_events));
di = dictGetIterator(server.latency_events);
while((de = dictNext(di)) != NULL) {
char *event = dictGetKey(de);
struct latencyTimeSeries *ts = dictGetVal(de);
int last = (ts->idx + LATENCY_TS_LEN - 1) % LATENCY_TS_LEN;
addReplyMultiBulkLen(c,4);
addReplyArrayLen(c,4);
addReplyBulkCString(c,event);
addReplyLongLong(c,ts->samples[last].time);
addReplyLongLong(c,ts->samples[last].latency);
......@@ -583,7 +583,7 @@ NULL
/* LATENCY HISTORY <event> */
ts = dictFetchValue(server.latency_events,c->argv[2]->ptr);
if (ts == NULL) {
addReplyMultiBulkLen(c,0);
addReplyArrayLen(c,0);
} else {
latencyCommandReplyWithSamples(c,ts);
}
......
......@@ -1123,10 +1123,10 @@ int RM_ReplyWithArray(RedisModuleCtx *ctx, long len) {
ctx->postponed_arrays = zrealloc(ctx->postponed_arrays,sizeof(void*)*
(ctx->postponed_arrays_count+1));
ctx->postponed_arrays[ctx->postponed_arrays_count] =
addDeferredMultiBulkLength(c);
addReplyDeferredLen(c);
ctx->postponed_arrays_count++;
} else {
addReplyMultiBulkLen(c,len);
addReplyArrayLen(c,len);
}
return REDISMODULE_OK;
}
......@@ -1169,7 +1169,7 @@ void RM_ReplySetArrayLength(RedisModuleCtx *ctx, long len) {
return;
}
ctx->postponed_arrays_count--;
setDeferredMultiBulkLength(c,
setDeferredArrayLen(c,
ctx->postponed_arrays[ctx->postponed_arrays_count],
len);
if (ctx->postponed_arrays_count == 0) {
......@@ -1205,7 +1205,7 @@ int RM_ReplyWithString(RedisModuleCtx *ctx, RedisModuleString *str) {
int RM_ReplyWithNull(RedisModuleCtx *ctx) {
client *c = moduleGetReplyClient(ctx);
if (c == NULL) return REDISMODULE_OK;
addReply(c,shared.nullbulk);
addReplyNull(c);
return REDISMODULE_OK;
}
......@@ -3669,8 +3669,8 @@ void moduleHandleBlockedClients(void) {
* free the temporary client we just used for the replies. */
if (c) {
if (bc->reply_client->bufpos)
addReplyString(c,bc->reply_client->buf,
bc->reply_client->bufpos);
addReplyProto(c,bc->reply_client->buf,
bc->reply_client->bufpos);
if (listLength(bc->reply_client->reply))
listJoin(c->reply,bc->reply_client->reply);
c->reply_bytes += bc->reply_client->reply_bytes;
......@@ -4818,6 +4818,25 @@ int moduleUnload(sds name) {
return REDISMODULE_OK;
}
/* Helper function for the MODULE and HELLO command: send the list of the
* loaded modules to the client. */
void addReplyLoadedModules(client *c) {
dictIterator *di = dictGetIterator(modules);
dictEntry *de;
addReplyArrayLen(c,dictSize(modules));
while ((de = dictNext(di)) != NULL) {
sds name = dictGetKey(de);
struct RedisModule *module = dictGetVal(de);
addReplyMapLen(c,2);
addReplyBulkCString(c,"name");
addReplyBulkCBuffer(c,name,sdslen(name));
addReplyBulkCString(c,"ver");
addReplyLongLong(c,module->ver);
}
dictReleaseIterator(di);
}
/* Redis MODULE command.
*
* MODULE LOAD <path> [args...] */
......@@ -4865,20 +4884,7 @@ NULL
addReplyErrorFormat(c,"Error unloading module: %s",errmsg);
}
} else if (!strcasecmp(subcmd,"list") && c->argc == 2) {
dictIterator *di = dictGetIterator(modules);
dictEntry *de;
addReplyMultiBulkLen(c,dictSize(modules));
while ((de = dictNext(di)) != NULL) {
sds name = dictGetKey(de);
struct RedisModule *module = dictGetVal(de);
addReplyMultiBulkLen(c,4);
addReplyBulkCString(c,"name");
addReplyBulkCBuffer(c,name,sdslen(name));
addReplyBulkCString(c,"ver");
addReplyLongLong(c,module->ver);
}
dictReleaseIterator(di);
addReplyLoadedModules(c);
} else {
addReplySubcommandSyntaxError(c);
return;
......
......@@ -134,7 +134,7 @@ void execCommand(client *c) {
* in the second an EXECABORT error is returned. */
if (c->flags & (CLIENT_DIRTY_CAS|CLIENT_DIRTY_EXEC)) {
addReply(c, c->flags & CLIENT_DIRTY_EXEC ? shared.execaborterr :
shared.nullmultibulk);
shared.nullarray[c->resp]);
discardTransaction(c);
goto handle_monitor;
}
......@@ -159,7 +159,7 @@ void execCommand(client *c) {
orig_argv = c->argv;
orig_argc = c->argc;
orig_cmd = c->cmd;
addReplyMultiBulkLen(c,c->mstate.count);
addReplyArrayLen(c,c->mstate.count);
for (j = 0; j < c->mstate.count; j++) {
c->argc = c->mstate.commands[j].argc;
c->argv = c->mstate.commands[j].argv;
......
......@@ -107,6 +107,7 @@ client *createClient(int fd) {
uint64_t client_id;
atomicGetIncr(server.next_client_id,client_id,1);
c->id = client_id;
c->resp = 2;
c->fd = fd;
c->name = NULL;
c->bufpos = 0;
......@@ -118,12 +119,15 @@ client *createClient(int fd) {
c->argc = 0;
c->argv = NULL;
c->cmd = c->lastcmd = NULL;
c->user = DefaultUser;
c->multibulklen = 0;
c->bulklen = -1;
c->sentlen = 0;
c->flags = 0;
c->ctime = c->lastinteraction = server.unixtime;
c->authenticated = 0;
/* If the default user does not require authentication, the user is
* directly authenticated. */
c->authenticated = (c->user->flags & USER_FLAG_NOPASS) != 0;
c->replstate = REPL_STATE_NONE;
c->repl_put_online_on_ack = 0;
c->reploff = 0;
......@@ -252,7 +256,7 @@ int _addReplyToBuffer(client *c, const char *s, size_t len) {
return C_OK;
}
void _addReplyStringToList(client *c, const char *s, size_t len) {
void _addReplyProtoToList(client *c, const char *s, size_t len) {
if (c->flags & CLIENT_CLOSE_AFTER_REPLY) return;
listNode *ln = listLast(c->reply);
......@@ -299,7 +303,7 @@ void addReply(client *c, robj *obj) {
if (sdsEncodedObject(obj)) {
if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != C_OK)
_addReplyStringToList(c,obj->ptr,sdslen(obj->ptr));
_addReplyProtoToList(c,obj->ptr,sdslen(obj->ptr));
} else if (obj->encoding == OBJ_ENCODING_INT) {
/* For integer encoded strings we just convert it into a string
* using our optimized function, and attach the resulting string
......@@ -307,7 +311,7 @@ void addReply(client *c, robj *obj) {
char buf[32];
size_t len = ll2string(buf,sizeof(buf),(long)obj->ptr);
if (_addReplyToBuffer(c,buf,len) != C_OK)
_addReplyStringToList(c,buf,len);
_addReplyProtoToList(c,buf,len);
} else {
serverPanic("Wrong obj->encoding in addReply()");
}
......@@ -322,7 +326,7 @@ void addReplySds(client *c, sds s) {
return;
}
if (_addReplyToBuffer(c,s,sdslen(s)) != C_OK)
_addReplyStringToList(c,s,sdslen(s));
_addReplyProtoToList(c,s,sdslen(s));
sdsfree(s);
}
......@@ -332,12 +336,12 @@ void addReplySds(client *c, sds s) {
*
* It is efficient because does not create an SDS object nor an Redis object
* if not needed. The object will only be created by calling
* _addReplyStringToList() if we fail to extend the existing tail object
* _addReplyProtoToList() if we fail to extend the existing tail object
* in the list of objects. */
void addReplyString(client *c, const char *s, size_t len) {
void addReplyProto(client *c, const char *s, size_t len) {
if (prepareClientToWrite(c) != C_OK) return;
if (_addReplyToBuffer(c,s,len) != C_OK)
_addReplyStringToList(c,s,len);
_addReplyProtoToList(c,s,len);
}
/* Low level function called by the addReplyError...() functions.
......@@ -351,9 +355,9 @@ void addReplyString(client *c, const char *s, size_t len) {
void addReplyErrorLength(client *c, const char *s, size_t len) {
/* If the string already starts with "-..." then the error code
* is provided by the caller. Otherwise we use "-ERR". */
if (!len || s[0] != '-') addReplyString(c,"-ERR ",5);
addReplyString(c,s,len);
addReplyString(c,"\r\n",2);
if (!len || s[0] != '-') addReplyProto(c,"-ERR ",5);
addReplyProto(c,s,len);
addReplyProto(c,"\r\n",2);
/* Sometimes it could be normal that a slave replies to a master with
* an error and this function gets called. Actually the error will never
......@@ -396,9 +400,9 @@ void addReplyErrorFormat(client *c, const char *fmt, ...) {
}
void addReplyStatusLength(client *c, const char *s, size_t len) {
addReplyString(c,"+",1);
addReplyString(c,s,len);
addReplyString(c,"\r\n",2);
addReplyProto(c,"+",1);
addReplyProto(c,s,len);
addReplyProto(c,"\r\n",2);
}
void addReplyStatus(client *c, const char *status) {
......@@ -416,28 +420,28 @@ void addReplyStatusFormat(client *c, const char *fmt, ...) {
/* Adds an empty object to the reply list that will contain the multi bulk
* length, which is not known when this function is called. */
void *addDeferredMultiBulkLength(client *c) {
void *addReplyDeferredLen(client *c) {
/* Note that we install the write event here even if the object is not
* ready to be sent, since we are sure that before returning to the
* event loop setDeferredMultiBulkLength() will be called. */
* event loop setDeferredAggregateLen() will be called. */
if (prepareClientToWrite(c) != C_OK) return NULL;
listAddNodeTail(c->reply,NULL); /* NULL is our placeholder. */
return listLast(c->reply);
}
/* Populate the length object and try gluing it to the next chunk. */
void setDeferredMultiBulkLength(client *c, void *node, long length) {
void setDeferredAggregateLen(client *c, void *node, long length, char prefix) {
listNode *ln = (listNode*)node;
clientReplyBlock *next;
char lenstr[128];
size_t lenstr_len = sprintf(lenstr, "*%ld\r\n", length);
size_t lenstr_len = sprintf(lenstr, "%c%ld\r\n", prefix, length);
/* Abort when *node is NULL: when the client should not accept writes
* we return NULL in addDeferredMultiBulkLength() */
* we return NULL in addReplyDeferredLen() */
if (node == NULL) return;
serverAssert(!listNodeValue(ln));
/* Normally we fill this dummy NULL node, added by addDeferredMultiBulkLength(),
/* Normally we fill this dummy NULL node, added by addReplyDeferredLen(),
* with a new buffer structure containing the protocol needed to specify
* the length of the array following. However sometimes when there is
* little memory to move, we may instead remove this NULL node, and prefix
......@@ -467,18 +471,55 @@ void setDeferredMultiBulkLength(client *c, void *node, long length) {
asyncCloseClientOnOutputBufferLimitReached(c);
}
void setDeferredArrayLen(client *c, void *node, long length) {
setDeferredAggregateLen(c,node,length,'*');
}
void setDeferredMapLen(client *c, void *node, long length) {
int prefix = c->resp == 2 ? '*' : '%';
if (c->resp == 2) length *= 2;
setDeferredAggregateLen(c,node,length,prefix);
}
void setDeferredSetLen(client *c, void *node, long length) {
int prefix = c->resp == 2 ? '*' : '~';
setDeferredAggregateLen(c,node,length,prefix);
}
void setDeferredAttributeLen(client *c, void *node, long length) {
int prefix = c->resp == 2 ? '*' : '|';
if (c->resp == 2) length *= 2;
setDeferredAggregateLen(c,node,length,prefix);
}
void setDeferredPushLen(client *c, void *node, long length) {
int prefix = c->resp == 2 ? '*' : '>';
setDeferredAggregateLen(c,node,length,prefix);
}
/* Add a double as a bulk reply */
void addReplyDouble(client *c, double d) {
char dbuf[128], sbuf[128];
int dlen, slen;
if (isinf(d)) {
/* Libc in odd systems (Hi Solaris!) will format infinite in a
* different way, so better to handle it in an explicit way. */
addReplyBulkCString(c, d > 0 ? "inf" : "-inf");
if (c->resp == 2) {
addReplyBulkCString(c, d > 0 ? "inf" : "-inf");
} else {
addReplyProto(c, d > 0 ? ",inf\r\n" : "-inf\r\n",
d > 0 ? 6 : 7);
}
} else {
dlen = snprintf(dbuf,sizeof(dbuf),"%.17g",d);
slen = snprintf(sbuf,sizeof(sbuf),"$%d\r\n%s\r\n",dlen,dbuf);
addReplyString(c,sbuf,slen);
char dbuf[MAX_LONG_DOUBLE_CHARS+3],
sbuf[MAX_LONG_DOUBLE_CHARS+32];
int dlen, slen;
if (c->resp == 2) {
dlen = snprintf(dbuf,sizeof(dbuf),"%.17g",d);
slen = snprintf(sbuf,sizeof(sbuf),"$%d\r\n%s\r\n",dlen,dbuf);
addReplyProto(c,sbuf,slen);
} else {
dlen = snprintf(dbuf,sizeof(dbuf),",%.17g\r\n",d);
addReplyProto(c,dbuf,dlen);
}
}
}
......@@ -486,9 +527,17 @@ void addReplyDouble(client *c, double d) {
* of the double instead of exposing the crude behavior of doubles to the
* dear user. */
void addReplyHumanLongDouble(client *c, long double d) {
robj *o = createStringObjectFromLongDouble(d,1);
addReplyBulk(c,o);
decrRefCount(o);
if (c->resp == 2) {
robj *o = createStringObjectFromLongDouble(d,1);
addReplyBulk(c,o);
decrRefCount(o);
} else {
char buf[MAX_LONG_DOUBLE_CHARS];
int len = ld2string(buf,sizeof(buf),d,1);
addReplyProto(c,",",1);
addReplyProto(c,buf,len);
addReplyProto(c,"\r\n",2);
}
}
/* Add a long long as integer reply or bulk len / multi bulk count.
......@@ -512,7 +561,7 @@ void addReplyLongLongWithPrefix(client *c, long long ll, char prefix) {
len = ll2string(buf+1,sizeof(buf)-1,ll);
buf[len+1] = '\r';
buf[len+2] = '\n';
addReplyString(c,buf,len+3);
addReplyProto(c,buf,len+3);
}
void addReplyLongLong(client *c, long long ll) {
......@@ -524,11 +573,65 @@ void addReplyLongLong(client *c, long long ll) {
addReplyLongLongWithPrefix(c,ll,':');
}
void addReplyMultiBulkLen(client *c, long length) {
if (length < OBJ_SHARED_BULKHDR_LEN)
void addReplyAggregateLen(client *c, long length, int prefix) {
if (prefix == '*' && length < OBJ_SHARED_BULKHDR_LEN)
addReply(c,shared.mbulkhdr[length]);
else
addReplyLongLongWithPrefix(c,length,'*');
addReplyLongLongWithPrefix(c,length,prefix);
}
void addReplyArrayLen(client *c, long length) {
addReplyAggregateLen(c,length,'*');
}
void addReplyMapLen(client *c, long length) {
int prefix = c->resp == 2 ? '*' : '%';
if (c->resp == 2) length *= 2;
addReplyAggregateLen(c,length,prefix);
}
void addReplySetLen(client *c, long length) {
int prefix = c->resp == 2 ? '*' : '~';
addReplyAggregateLen(c,length,prefix);
}
void addReplyAttributeLen(client *c, long length) {
int prefix = c->resp == 2 ? '*' : '|';
if (c->resp == 2) length *= 2;
addReplyAggregateLen(c,length,prefix);
}
void addReplyPushLen(client *c, long length) {
int prefix = c->resp == 2 ? '*' : '>';
addReplyAggregateLen(c,length,prefix);
}
void addReplyNull(client *c) {
if (c->resp == 2) {
addReplyProto(c,"$-1\r\n",5);
} else {
addReplyProto(c,"_\r\n",3);
}
}
void addReplyBool(client *c, int b) {
if (c->resp == 2) {
addReply(c, b ? shared.cone : shared.czero);
} else {
addReplyProto(c, b ? "#t\r\n" : "#f\r\n",4);
}
}
/* A null array is a concept that no longer exists in RESP3. However
* RESP2 had it, so API-wise we have this call, that will emit the correct
* RESP2 protocol, however for RESP3 the reply will always be just the
* Null type "_\r\n". */
void addReplyNullArray(client *c) {
if (c->resp == 2) {
addReplyProto(c,"*-1\r\n",5);
} else {
addReplyProto(c,"_\r\n",3);
}
}
/* Create the length prefix of a bulk reply, example: $2234 */
......@@ -567,7 +670,7 @@ void addReplyBulk(client *c, robj *obj) {
/* Add a C buffer as bulk reply */
void addReplyBulkCBuffer(client *c, const void *p, size_t len) {
addReplyLongLongWithPrefix(c,len,'$');
addReplyString(c,p,len);
addReplyProto(c,p,len);
addReply(c,shared.crlf);
}
......@@ -581,7 +684,7 @@ void addReplyBulkSds(client *c, sds s) {
/* Add a C null term string as bulk reply */
void addReplyBulkCString(client *c, const char *s) {
if (s == NULL) {
addReply(c,shared.nullbulk);
addReplyNull(c);
} else {
addReplyBulkCBuffer(c,s,strlen(s));
}
......@@ -596,13 +699,42 @@ void addReplyBulkLongLong(client *c, long long ll) {
addReplyBulkCBuffer(c,buf,len);
}
/* Reply with a verbatim type having the specified extension.
*
* The 'ext' is the "extension" of the file, actually just a three
* character type that describes the format of the verbatim string.
* For instance "txt" means it should be interpreted as a text only
* file by the receiver, "md " as markdown, and so forth. Only the
* three first characters of the extension are used, and if the
* provided one is shorter than that, the remaining is filled with
* spaces. */
void addReplyVerbatim(client *c, const char *s, size_t len, const char *ext) {
if (c->resp == 2) {
addReplyBulkCBuffer(c,s,len);
} else {
char buf[32];
size_t preflen = snprintf(buf,sizeof(buf),"=%zu\r\nxxx:",len+4);
char *p = buf+preflen-4;
for (int i = 0; i < 3; i++) {
if (*ext == '\0') {
p[i] = ' ';
} else {
p[i] = *ext++;
}
}
addReplyProto(c,buf,preflen);
addReplyProto(c,s,len);
addReplyProto(c,"\r\n",2);
}
}
/* Add an array of C strings as status replies with a heading.
* This function is typically invoked by from commands that support
* subcommands in response to the 'help' subcommand. The help array
* is terminated by NULL sentinel. */
void addReplyHelp(client *c, const char **help) {
sds cmd = sdsnew((char*) c->argv[0]->ptr);
void *blenp = addDeferredMultiBulkLength(c);
void *blenp = addReplyDeferredLen(c);
int blen = 0;
sdstoupper(cmd);
......@@ -613,7 +745,7 @@ void addReplyHelp(client *c, const char **help) {
while (help[blen]) addReplyStatus(c,help[blen++]);
blen++; /* Account for the header line(s). */
setDeferredMultiBulkLength(c,blenp,blen);
setDeferredArrayLen(c,blenp,blen);
}
/* Add a suggestive error reply.
......@@ -678,7 +810,7 @@ static void acceptCommonHandler(int fd, int flags, char *ip) {
* user what to do to fix it if needed. */
if (server.protected_mode &&
server.bindaddr_count == 0 &&
server.requirepass == NULL &&
DefaultUser->flags & USER_FLAG_NOPASS &&
!(flags & CLIENT_UNIX_SOCKET) &&
ip != NULL)
{
......@@ -1893,7 +2025,7 @@ NULL
if (c->name)
addReplyBulk(c,c->name);
else
addReply(c,shared.nullbulk);
addReplyNull(c);
} else if (!strcasecmp(c->argv[1]->ptr,"pause") && c->argc == 3) {
long long duration;
......@@ -1906,6 +2038,63 @@ NULL
}
}
/* HELLO <protocol-version> [AUTH <user> <password>] */
void helloCommand(client *c) {
long long ver;
if (getLongLongFromObject(c->argv[1],&ver) != C_OK ||
ver < 2 || ver > 3)
{
addReplyError(c,"-NOPROTO unsupported protocol version");
return;
}
/* Switching to protocol v2 is not allowed. But we send a specific
* error message in this case. */
if (ver == 2) {
addReplyError(c,"Switching to RESP version 2 is not allowed.");
return;
}
/* At this point we need to be authenticated to continue. */
if (!c->authenticated) {
addReplyError(c,"-NOAUTH HELLO must be called with the client already "
"authenticated, otherwise the HELLO AUTH <user> <pass> "
"option can be used to authenticate the client and "
"select the RESP protocol version at the same time");
return;
}
/* Let's switch to RESP3 mode. */
c->resp = 3;
addReplyMapLen(c,7);
addReplyBulkCString(c,"server");
addReplyBulkCString(c,"redis");
addReplyBulkCString(c,"version");
addReplyBulkCString(c,REDIS_VERSION);
addReplyBulkCString(c,"proto");
addReplyLongLong(c,3);
addReplyBulkCString(c,"id");
addReplyLongLong(c,c->id);
addReplyBulkCString(c,"mode");
if (server.sentinel_mode) addReplyBulkCString(c,"sentinel");
if (server.cluster_enabled) addReplyBulkCString(c,"cluster");
else addReplyBulkCString(c,"standalone");
if (!server.sentinel_mode) {
addReplyBulkCString(c,"role");
addReplyBulkCString(c,server.masterhost ? "replica" : "master");
}
addReplyBulkCString(c,"modules");
addReplyLoadedModules(c);
}
/* This callback is bound to POST and "Host:" command names. Those are not
* really commands, but are used in security attacks in order to talk to
* Redis instances via HTTP, with a technique called "cross protocol scripting"
......
......@@ -1248,15 +1248,15 @@ NULL
};
addReplyHelp(c, help);
} else if (!strcasecmp(c->argv[1]->ptr,"refcount") && c->argc == 3) {
if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.nullbulk))
if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.null[c->resp]))
== NULL) return;
addReplyLongLong(c,o->refcount);
} else if (!strcasecmp(c->argv[1]->ptr,"encoding") && c->argc == 3) {
if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.nullbulk))
if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.null[c->resp]))
== NULL) return;
addReplyBulkCString(c,strEncoding(o->encoding));
} else if (!strcasecmp(c->argv[1]->ptr,"idletime") && c->argc == 3) {
if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.nullbulk))
if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.null[c->resp]))
== NULL) return;
if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
addReplyError(c,"An LFU maxmemory policy is selected, idle time not tracked. Please note that when switching between policies at runtime LRU and LFU data will take some time to adjust.");
......@@ -1264,7 +1264,7 @@ NULL
}
addReplyLongLong(c,estimateObjectIdleTime(o)/1000);
} else if (!strcasecmp(c->argv[1]->ptr,"freq") && c->argc == 3) {
if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.nullbulk))
if ((o = objectCommandLookupOrReply(c,c->argv[2],shared.null[c->resp]))
== NULL) return;
if (!(server.maxmemory_policy & MAXMEMORY_FLAG_LFU)) {
addReplyError(c,"An LFU maxmemory policy is not selected, access frequency not tracked. Please note that when switching between policies at runtime LRU and LFU data will take some time to adjust.");
......@@ -1316,7 +1316,7 @@ NULL
}
}
if ((de = dictFind(c->db->dict,c->argv[2]->ptr)) == NULL) {
addReply(c, shared.nullbulk);
addReplyNull(c);
return;
}
size_t usage = objectComputeSize(dictGetVal(de),samples);
......@@ -1326,7 +1326,7 @@ NULL
} else if (!strcasecmp(c->argv[1]->ptr,"stats") && c->argc == 2) {
struct redisMemOverhead *mh = getMemoryOverheadData();
addReplyMultiBulkLen(c,(25+mh->num_dbs)*2);
addReplyMapLen(c,25+mh->num_dbs);
addReplyBulkCString(c,"peak.allocated");
addReplyLongLong(c,mh->peak_allocated);
......@@ -1356,7 +1356,7 @@ NULL
char dbname[32];
snprintf(dbname,sizeof(dbname),"db.%zd",mh->db[j].dbid);
addReplyBulkCString(c,dbname);
addReplyMultiBulkLen(c,4);
addReplyMapLen(c,2);
addReplyBulkCString(c,"overhead.hashtable.main");
addReplyLongLong(c,mh->db[j].overhead_ht_main);
......
......@@ -29,6 +29,93 @@
#include "server.h"
int clientSubscriptionsCount(client *c);
/*-----------------------------------------------------------------------------
* Pubsub client replies API
*----------------------------------------------------------------------------*/
/* Send a pubsub message of type "message" to the client. */
void addReplyPubsubMessage(client *c, robj *channel, robj *msg) {
if (c->resp == 2)
addReply(c,shared.mbulkhdr[3]);
else
addReplyPushLen(c,3);
addReply(c,shared.messagebulk);
addReplyBulk(c,channel);
addReplyBulk(c,msg);
}
/* Send a pubsub message of type "pmessage" to the client. The difference
* with the "message" type delivered by addReplyPubsubMessage() is that
* this message format also includes the pattern that matched the message. */
void addReplyPubsubPatMessage(client *c, robj *pat, robj *channel, robj *msg) {
if (c->resp == 2)
addReply(c,shared.mbulkhdr[4]);
else
addReplyPushLen(c,4);
addReply(c,shared.pmessagebulk);
addReplyBulk(c,pat);
addReplyBulk(c,channel);
addReplyBulk(c,msg);
}
/* Send the pubsub subscription notification to the client. */
void addReplyPubsubSubscribed(client *c, robj *channel) {
if (c->resp == 2)
addReply(c,shared.mbulkhdr[3]);
else
addReplyPushLen(c,3);
addReply(c,shared.subscribebulk);
addReplyBulk(c,channel);
addReplyLongLong(c,clientSubscriptionsCount(c));
}
/* Send the pubsub unsubscription notification to the client.
* Channel can be NULL: this is useful when the client sends a mass
* unsubscribe command but there are no channels to unsubscribe from: we
* still send a notification. */
void addReplyPubsubUnsubscribed(client *c, robj *channel) {
if (c->resp == 2)
addReply(c,shared.mbulkhdr[3]);
else
addReplyPushLen(c,3);
addReply(c,shared.unsubscribebulk);
if (channel)
addReplyBulk(c,channel);
else
addReplyNull(c);
addReplyLongLong(c,clientSubscriptionsCount(c));
}
/* Send the pubsub pattern subscription notification to the client. */
void addReplyPubsubPatSubscribed(client *c, robj *pattern) {
if (c->resp == 2)
addReply(c,shared.mbulkhdr[3]);
else
addReplyPushLen(c,3);
addReply(c,shared.psubscribebulk);
addReplyBulk(c,pattern);
addReplyLongLong(c,clientSubscriptionsCount(c));
}
/* Send the pubsub pattern unsubscription notification to the client.
* Pattern can be NULL: this is useful when the client sends a mass
* punsubscribe command but there are no pattern to unsubscribe from: we
* still send a notification. */
void addReplyPubsubPatUnsubscribed(client *c, robj *pattern) {
if (c->resp == 2)
addReply(c,shared.mbulkhdr[3]);
else
addReplyPushLen(c,3);
addReply(c,shared.punsubscribebulk);
if (pattern)
addReplyBulk(c,pattern);
else
addReplyNull(c);
addReplyLongLong(c,clientSubscriptionsCount(c));
}
/*-----------------------------------------------------------------------------
* Pubsub low level API
*----------------------------------------------------------------------------*/
......@@ -76,10 +163,7 @@ int pubsubSubscribeChannel(client *c, robj *channel) {
listAddNodeTail(clients,c);
}
/* Notify the client */
addReply(c,shared.mbulkhdr[3]);
addReply(c,shared.subscribebulk);
addReplyBulk(c,channel);
addReplyLongLong(c,clientSubscriptionsCount(c));
addReplyPubsubSubscribed(c,channel);
return retval;
}
......@@ -111,14 +195,7 @@ int pubsubUnsubscribeChannel(client *c, robj *channel, int notify) {
}
}
/* Notify the client */
if (notify) {
addReply(c,shared.mbulkhdr[3]);
addReply(c,shared.unsubscribebulk);
addReplyBulk(c,channel);
addReplyLongLong(c,dictSize(c->pubsub_channels)+
listLength(c->pubsub_patterns));
}
if (notify) addReplyPubsubUnsubscribed(c,channel);
decrRefCount(channel); /* it is finally safe to release it */
return retval;
}
......@@ -138,10 +215,7 @@ int pubsubSubscribePattern(client *c, robj *pattern) {
listAddNodeTail(server.pubsub_patterns,pat);
}
/* Notify the client */
addReply(c,shared.mbulkhdr[3]);
addReply(c,shared.psubscribebulk);
addReplyBulk(c,pattern);
addReplyLongLong(c,clientSubscriptionsCount(c));
addReplyPubsubPatSubscribed(c,pattern);
return retval;
}
......@@ -162,13 +236,7 @@ int pubsubUnsubscribePattern(client *c, robj *pattern, int notify) {
listDelNode(server.pubsub_patterns,ln);
}
/* Notify the client */
if (notify) {
addReply(c,shared.mbulkhdr[3]);
addReply(c,shared.punsubscribebulk);
addReplyBulk(c,pattern);
addReplyLongLong(c,dictSize(c->pubsub_channels)+
listLength(c->pubsub_patterns));
}
if (notify) addReplyPubsubPatUnsubscribed(c,pattern);
decrRefCount(pattern);
return retval;
}
......@@ -186,13 +254,7 @@ int pubsubUnsubscribeAllChannels(client *c, int notify) {
count += pubsubUnsubscribeChannel(c,channel,notify);
}
/* We were subscribed to nothing? Still reply to the client. */
if (notify && count == 0) {
addReply(c,shared.mbulkhdr[3]);
addReply(c,shared.unsubscribebulk);
addReply(c,shared.nullbulk);
addReplyLongLong(c,dictSize(c->pubsub_channels)+
listLength(c->pubsub_patterns));
}
if (notify && count == 0) addReplyPubsubUnsubscribed(c,NULL);
dictReleaseIterator(di);
return count;
}
......@@ -210,14 +272,7 @@ int pubsubUnsubscribeAllPatterns(client *c, int notify) {
count += pubsubUnsubscribePattern(c,pattern,notify);
}
if (notify && count == 0) {
/* We were subscribed to nothing? Still reply to the client. */
addReply(c,shared.mbulkhdr[3]);
addReply(c,shared.punsubscribebulk);
addReply(c,shared.nullbulk);
addReplyLongLong(c,dictSize(c->pubsub_channels)+
listLength(c->pubsub_patterns));
}
if (notify && count == 0) addReplyPubsubPatUnsubscribed(c,NULL);
return count;
}
......@@ -238,11 +293,7 @@ int pubsubPublishMessage(robj *channel, robj *message) {
listRewind(list,&li);
while ((ln = listNext(&li)) != NULL) {
client *c = ln->value;
addReply(c,shared.mbulkhdr[3]);
addReply(c,shared.messagebulk);
addReplyBulk(c,channel);
addReplyBulk(c,message);
addReplyPubsubMessage(c,channel,message);
receivers++;
}
}
......@@ -256,12 +307,10 @@ int pubsubPublishMessage(robj *channel, robj *message) {
if (stringmatchlen((char*)pat->pattern->ptr,
sdslen(pat->pattern->ptr),
(char*)channel->ptr,
sdslen(channel->ptr),0)) {
addReply(pat->client,shared.mbulkhdr[4]);
addReply(pat->client,shared.pmessagebulk);
addReplyBulk(pat->client,pat->pattern);
addReplyBulk(pat->client,channel);
addReplyBulk(pat->client,message);
sdslen(channel->ptr),0))
{
addReplyPubsubPatMessage(pat->client,
pat->pattern,channel,message);
receivers++;
}
}
......@@ -343,7 +392,7 @@ NULL
long mblen = 0;
void *replylen;
replylen = addDeferredMultiBulkLength(c);
replylen = addReplyDeferredLen(c);
while((de = dictNext(di)) != NULL) {
robj *cobj = dictGetKey(de);
sds channel = cobj->ptr;
......@@ -356,12 +405,12 @@ NULL
}
}
dictReleaseIterator(di);
setDeferredMultiBulkLength(c,replylen,mblen);
setDeferredArrayLen(c,replylen,mblen);
} else if (!strcasecmp(c->argv[1]->ptr,"numsub") && c->argc >= 2) {
/* PUBSUB NUMSUB [Channel_1 ... Channel_N] */
int j;
addReplyMultiBulkLen(c,(c->argc-2)*2);
addReplyArrayLen(c,(c->argc-2)*2);
for (j = 2; j < c->argc; j++) {
list *l = dictFetchValue(server.pubsub_channels,c->argv[j]);
......
......@@ -810,6 +810,9 @@ static sds cliFormatReplyTTY(redisReply *r, char *prefix) {
case REDIS_REPLY_INTEGER:
out = sdscatprintf(out,"(integer) %lld\n",r->integer);
break;
case REDIS_REPLY_DOUBLE:
out = sdscatprintf(out,"(double) %s\n",r->str);
break;
case REDIS_REPLY_STRING:
/* If you are producing output for the standard output we want
* a more interesting output with quoted characters and so forth */
......@@ -819,9 +822,21 @@ static sds cliFormatReplyTTY(redisReply *r, char *prefix) {
case REDIS_REPLY_NIL:
out = sdscat(out,"(nil)\n");
break;
case REDIS_REPLY_BOOL:
out = sdscat(out,r->integer ? "(true)\n" : "(false)\n");
break;
case REDIS_REPLY_ARRAY:
case REDIS_REPLY_MAP:
case REDIS_REPLY_SET:
if (r->elements == 0) {
out = sdscat(out,"(empty list or set)\n");
if (r->type == REDIS_REPLY_ARRAY)
out = sdscat(out,"(empty array)\n");
else if (r->type == REDIS_REPLY_MAP)
out = sdscat(out,"(empty hash)\n");
else if (r->type == REDIS_REPLY_SET)
out = sdscat(out,"(empty set)\n");
else
out = sdscat(out,"(empty aggregate type)\n");
} else {
unsigned int i, idxlen = 0;
char _prefixlen[16];
......@@ -831,6 +846,7 @@ static sds cliFormatReplyTTY(redisReply *r, char *prefix) {
/* Calculate chars needed to represent the largest index */
i = r->elements;
if (r->type == REDIS_REPLY_MAP) i /= 2;
do {
idxlen++;
i /= 10;
......@@ -842,17 +858,35 @@ static sds cliFormatReplyTTY(redisReply *r, char *prefix) {
_prefix = sdscat(sdsnew(prefix),_prefixlen);
/* Setup prefix format for every entry */
snprintf(_prefixfmt,sizeof(_prefixfmt),"%%s%%%ud) ",idxlen);
char numsep;
if (r->type == REDIS_REPLY_SET) numsep = '~';
else if (r->type == REDIS_REPLY_MAP) numsep = '#';
else numsep = ')';
snprintf(_prefixfmt,sizeof(_prefixfmt),"%%s%%%ud%c ",idxlen,numsep);
for (i = 0; i < r->elements; i++) {
unsigned int human_idx = (r->type == REDIS_REPLY_MAP) ?
i/2 : i;
human_idx++; /* Make it 1-based. */
/* Don't use the prefix for the first element, as the parent
* caller already prepended the index number. */
out = sdscatprintf(out,_prefixfmt,i == 0 ? "" : prefix,i+1);
out = sdscatprintf(out,_prefixfmt,i == 0 ? "" : prefix,human_idx);
/* Format the multi bulk entry */
tmp = cliFormatReplyTTY(r->element[i],_prefix);
out = sdscatlen(out,tmp,sdslen(tmp));
sdsfree(tmp);
/* For maps, format the value as well. */
if (r->type == REDIS_REPLY_MAP) {
i++;
sdsrange(out,0,-2);
out = sdscat(out," => ");
tmp = cliFormatReplyTTY(r->element[i],_prefix);
out = sdscatlen(out,tmp,sdslen(tmp));
sdsfree(tmp);
}
}
sdsfree(_prefix);
}
......@@ -2934,6 +2968,70 @@ static int clusterManagerSetSlotOwner(clusterManagerNode *owner,
return success;
}
/* Get the hash for the values of the specified keys in *keys_reply for the
* specified nodes *n1 and *n2, by calling DEBUG DIGEST-VALUE redis command
* on both nodes. Every key with same name on both nodes but having different
* values will be added to the *diffs list. Return 0 in case of reply
* error. */
static int clusterManagerCompareKeysValues(clusterManagerNode *n1,
clusterManagerNode *n2,
redisReply *keys_reply,
list *diffs)
{
size_t i, argc = keys_reply->elements + 2;
static const char *hash_zero = "0000000000000000000000000000000000000000";
char **argv = zcalloc(argc * sizeof(char *));
size_t *argv_len = zcalloc(argc * sizeof(size_t));
argv[0] = "DEBUG";
argv_len[0] = 5;
argv[1] = "DIGEST-VALUE";
argv_len[1] = 12;
for (i = 0; i < keys_reply->elements; i++) {
redisReply *entry = keys_reply->element[i];
int idx = i + 2;
argv[idx] = entry->str;
argv_len[idx] = entry->len;
}
int success = 0;
void *_reply1 = NULL, *_reply2 = NULL;
redisReply *r1 = NULL, *r2 = NULL;
redisAppendCommandArgv(n1->context,argc, (const char**)argv,argv_len);
success = (redisGetReply(n1->context, &_reply1) == REDIS_OK);
if (!success) goto cleanup;
r1 = (redisReply *) _reply1;
redisAppendCommandArgv(n2->context,argc, (const char**)argv,argv_len);
success = (redisGetReply(n2->context, &_reply2) == REDIS_OK);
if (!success) goto cleanup;
r2 = (redisReply *) _reply2;
success = (r1->type != REDIS_REPLY_ERROR && r2->type != REDIS_REPLY_ERROR);
if (r1->type == REDIS_REPLY_ERROR) {
CLUSTER_MANAGER_PRINT_REPLY_ERROR(n1, r1->str);
success = 0;
}
if (r2->type == REDIS_REPLY_ERROR) {
CLUSTER_MANAGER_PRINT_REPLY_ERROR(n2, r2->str);
success = 0;
}
if (!success) goto cleanup;
assert(keys_reply->elements == r1->elements &&
keys_reply->elements == r2->elements);
for (i = 0; i < keys_reply->elements; i++) {
char *key = keys_reply->element[i]->str;
char *hash1 = r1->element[i]->str;
char *hash2 = r2->element[i]->str;
/* Ignore keys that don't exist in both nodes. */
if (strcmp(hash1, hash_zero) == 0 || strcmp(hash2, hash_zero) == 0)
continue;
if (strcmp(hash1, hash2) != 0) listAddNodeTail(diffs, key);
}
cleanup:
if (r1) freeReplyObject(r1);
if (r2) freeReplyObject(r2);
zfree(argv);
zfree(argv_len);
return success;
}
/* Migrate keys taken from reply->elements. It returns the reply from the
* MIGRATE command, or NULL if something goes wrong. If the argument 'dots'
* is not NULL, a dot will be printed for every migrated key. */
......@@ -3014,8 +3112,10 @@ static int clusterManagerMigrateKeysInSlot(clusterManagerNode *source,
char **err)
{
int success = 1;
int retry = (config.cluster_manager_command.flags &
(CLUSTER_MANAGER_CMD_FLAG_FIX | CLUSTER_MANAGER_CMD_FLAG_REPLACE));
int do_fix = config.cluster_manager_command.flags &
CLUSTER_MANAGER_CMD_FLAG_FIX;
int do_replace = config.cluster_manager_command.flags &
CLUSTER_MANAGER_CMD_FLAG_REPLACE;
while (1) {
char *dots = NULL;
redisReply *reply = NULL, *migrate_reply = NULL;
......@@ -3049,6 +3149,8 @@ static int clusterManagerMigrateKeysInSlot(clusterManagerNode *source,
int is_busy = strstr(migrate_reply->str, "BUSYKEY") != NULL;
int not_served = 0;
if (!is_busy) {
/* Check if the slot is unassigned (not served) in the
* source node's configuration. */
char *get_owner_err = NULL;
clusterManagerNode *served_by =
clusterManagerGetSlotOwner(source, slot, &get_owner_err);
......@@ -3061,20 +3163,68 @@ static int clusterManagerMigrateKeysInSlot(clusterManagerNode *source,
}
}
}
if (retry && (is_busy || not_served)) {
/* If the key already exists, try to migrate keys
* adding REPLACE option.
* If the key's slot is not served, try to assign slot
/* Try to handle errors. */
if (is_busy || not_served) {
/* If the key's slot is not served, try to assign slot
* to the target node. */
if (not_served) {
if (do_fix && not_served) {
clusterManagerLogWarn("*** Slot was not served, setting "
"owner to node %s:%d.\n",
target->ip, target->port);
clusterManagerSetSlot(source, target, slot, "node", NULL);
}
/* If the key already exists in the target node (BUSYKEY),
* check whether its value is the same in both nodes.
* In case of equal values, retry migration with the
* REPLACE option.
* In case of different values:
* - If the migration is requested by the fix command, stop
* and warn the user.
* - In other cases (ie. reshard), proceed only if the user
* launched the command with the --cluster-replace option.*/
if (is_busy) {
clusterManagerLogWarn("*** Target key exists. "
"Replacing it for FIX.\n");
clusterManagerLogWarn("\n*** Target key exists\n");
if (!do_replace) {
clusterManagerLogWarn("*** Checking key values on "
"both nodes...\n");
list *diffs = listCreate();
success = clusterManagerCompareKeysValues(source,
target, reply, diffs);
if (!success) {
clusterManagerLogErr("*** Value check failed!\n");
listRelease(diffs);
goto next;
}
if (listLength(diffs) > 0) {
success = 0;
clusterManagerLogErr(
"*** Found %d key(s) in both source node and "
"target node having different values.\n"
" Source node: %s:%d\n"
" Target node: %s:%d\n"
" Keys(s):\n",
listLength(diffs),
source->ip, source->port,
target->ip, target->port);
listIter dli;
listNode *dln;
listRewind(diffs, &dli);
while((dln = listNext(&dli)) != NULL) {
char *k = dln->value;
clusterManagerLogErr(" - %s\n", k);
}
clusterManagerLogErr("Please fix the above key(s) "
"manually and try again "
"or relaunch the command \n"
"with --cluster-replace "
"option to force key "
"overriding.\n");
listRelease(diffs);
goto next;
}
listRelease(diffs);
}
clusterManagerLogWarn("*** Replacing target keys...\n");
}
freeReplyObject(migrate_reply);
migrate_reply = clusterManagerMigrateKeysInReply(source,
......@@ -5068,9 +5218,10 @@ static int clusterManagerCommandDeleteNode(int argc, char **argv) {
if (!success) return 0;
}
// Finally shutdown the node
clusterManagerLogInfo(">>> SHUTDOWN the node.\n");
redisReply *r = redisCommand(node->context, "SHUTDOWN");
/* Finally send CLUSTER RESET to the node. */
clusterManagerLogInfo(">>> Sending CLUSTER RESET SOFT to the "
"deleted node.\n");
redisReply *r = redisCommand(node->context, "CLUSTER RESET %s", "SOFT");
success = clusterManagerCheckRedisReply(node, r, NULL);
if (r) freeReplyObject(r);
return success;
......
......@@ -263,7 +263,7 @@ void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
* or are already in sync with the master. */
/* Add the multi bulk length. */
addReplyMultiBulkLen(slave,argc);
addReplyArrayLen(slave,argc);
/* Finally any additional argument that was not stored inside the
* static buffer if any (from j to argc). */
......@@ -296,7 +296,7 @@ void replicationFeedSlavesFromMasterStream(list *slaves, char *buf, size_t bufle
/* Don't feed slaves that are still waiting for BGSAVE to start */
if (slave->replstate == SLAVE_STATE_WAIT_BGSAVE_START) continue;
addReplyString(slave,buf,buflen);
addReplyProto(slave,buf,buflen);
}
}
......@@ -1080,6 +1080,7 @@ void replicationCreateMasterClient(int fd, int dbid) {
server.master->authenticated = 1;
server.master->reploff = server.master_initial_offset;
server.master->read_reploff = server.master->reploff;
server.master->user = NULL; /* This client can do everything. */
memcpy(server.master->replid, server.master_replid,
sizeof(server.master_replid));
/* If master offset is set to -1, this master is old and is not
......@@ -1256,6 +1257,7 @@ void readSyncBulkPayload(aeEventLoop *el, int fd, void *privdata, int mask) {
kill(server.rdb_child_pid,SIGUSR1);
rdbRemoveTempFile(server.rdb_child_pid);
closeChildInfoPipe();
updateDictResizePolicy();
}
if (rename(server.repl_transfer_tmpfile,server.rdb_filename) == -1) {
......@@ -2063,10 +2065,10 @@ void roleCommand(client *c) {
void *mbcount;
int slaves = 0;
addReplyMultiBulkLen(c,3);
addReplyArrayLen(c,3);
addReplyBulkCBuffer(c,"master",6);
addReplyLongLong(c,server.master_repl_offset);
mbcount = addDeferredMultiBulkLength(c);
mbcount = addReplyDeferredLen(c);
listRewind(server.slaves,&li);
while((ln = listNext(&li))) {
client *slave = ln->value;
......@@ -2078,17 +2080,17 @@ void roleCommand(client *c) {
slaveip = ip;
}
if (slave->replstate != SLAVE_STATE_ONLINE) continue;
addReplyMultiBulkLen(c,3);
addReplyArrayLen(c,3);
addReplyBulkCString(c,slaveip);
addReplyBulkLongLong(c,slave->slave_listening_port);
addReplyBulkLongLong(c,slave->repl_ack_off);
slaves++;
}
setDeferredMultiBulkLength(c,mbcount,slaves);
setDeferredArrayLen(c,mbcount,slaves);
} else {
char *slavestate = NULL;
addReplyMultiBulkLen(c,5);
addReplyArrayLen(c,5);
addReplyBulkCBuffer(c,"slave",5);
addReplyBulkCString(c,server.masterhost);
addReplyLongLong(c,server.masterport);
......@@ -2117,7 +2119,7 @@ void replicationSendAck(void) {
if (c != NULL) {
c->flags |= CLIENT_MASTER_FORCE_REPLY;
addReplyMultiBulkLen(c,3);
addReplyArrayLen(c,3);
addReplyBulkCString(c,"REPLCONF");
addReplyBulkCString(c,"ACK");
addReplyBulkLongLong(c,c->reploff);
......
......@@ -42,7 +42,7 @@ char *redisProtocolToLuaType_Int(lua_State *lua, char *reply);
char *redisProtocolToLuaType_Bulk(lua_State *lua, char *reply);
char *redisProtocolToLuaType_Status(lua_State *lua, char *reply);
char *redisProtocolToLuaType_Error(lua_State *lua, char *reply);
char *redisProtocolToLuaType_MultiBulk(lua_State *lua, char *reply);
char *redisProtocolToLuaType_MultiBulk(lua_State *lua, char *reply, int atype);
int redis_math_random (lua_State *L);
int redis_math_randomseed (lua_State *L);
void ldbInit(void);
......@@ -132,7 +132,9 @@ char *redisProtocolToLuaType(lua_State *lua, char* reply) {
case '$': p = redisProtocolToLuaType_Bulk(lua,reply); break;
case '+': p = redisProtocolToLuaType_Status(lua,reply); break;
case '-': p = redisProtocolToLuaType_Error(lua,reply); break;
case '*': p = redisProtocolToLuaType_MultiBulk(lua,reply); break;
case '*': p = redisProtocolToLuaType_MultiBulk(lua,reply,*p); break;
case '%': p = redisProtocolToLuaType_MultiBulk(lua,reply,*p); break;
case '~': p = redisProtocolToLuaType_MultiBulk(lua,reply,*p); break;
}
return p;
}
......@@ -180,22 +182,38 @@ char *redisProtocolToLuaType_Error(lua_State *lua, char *reply) {
return p+2;
}
char *redisProtocolToLuaType_MultiBulk(lua_State *lua, char *reply) {
char *redisProtocolToLuaType_MultiBulk(lua_State *lua, char *reply, int atype) {
char *p = strchr(reply+1,'\r');
long long mbulklen;
int j = 0;
string2ll(reply+1,p-reply-1,&mbulklen);
p += 2;
if (mbulklen == -1) {
lua_pushboolean(lua,0);
return p;
}
lua_newtable(lua);
for (j = 0; j < mbulklen; j++) {
lua_pushnumber(lua,j+1);
p = redisProtocolToLuaType(lua,p);
lua_settable(lua,-3);
if (server.lua_caller->resp == 2 || atype == '*') {
p += 2;
if (mbulklen == -1) {
lua_pushboolean(lua,0);
return p;
}
lua_newtable(lua);
for (j = 0; j < mbulklen; j++) {
lua_pushnumber(lua,j+1);
p = redisProtocolToLuaType(lua,p);
lua_settable(lua,-3);
}
} else if (server.lua_caller->resp == 3) {
/* Here we handle only Set and Map replies in RESP3 mode, since arrays
* follow the above RESP2 code path. */
p += 2;
lua_newtable(lua);
for (j = 0; j < mbulklen; j++) {
p = redisProtocolToLuaType(lua,p);
if (atype == '%') {
p = redisProtocolToLuaType(lua,p);
} else {
lua_pushboolean(lua,1);
}
lua_settable(lua,-3);
}
}
return p;
}
......@@ -282,7 +300,7 @@ void luaReplyToRedisReply(client *c, lua_State *lua) {
addReplyBulkCBuffer(c,(char*)lua_tostring(lua,-1),lua_strlen(lua,-1));
break;
case LUA_TBOOLEAN:
addReply(c,lua_toboolean(lua,-1) ? shared.cone : shared.nullbulk);
addReply(c,lua_toboolean(lua,-1) ? shared.cone : shared.null[c->resp]);
break;
case LUA_TNUMBER:
addReplyLongLong(c,(long long)lua_tonumber(lua,-1));
......@@ -315,7 +333,7 @@ void luaReplyToRedisReply(client *c, lua_State *lua) {
sdsfree(ok);
lua_pop(lua,1);
} else {
void *replylen = addDeferredMultiBulkLength(c);
void *replylen = addReplyDeferredLen(c);
int j = 1, mbulklen = 0;
lua_pop(lua,1); /* Discard the 'ok' field value we popped */
......@@ -330,11 +348,11 @@ void luaReplyToRedisReply(client *c, lua_State *lua) {
luaReplyToRedisReply(c, lua);
mbulklen++;
}
setDeferredMultiBulkLength(c,replylen,mbulklen);
setDeferredArrayLen(c,replylen,mbulklen);
}
break;
default:
addReply(c,shared.nullbulk);
addReplyNull(c);
}
lua_pop(lua,1);
}
......@@ -1501,7 +1519,7 @@ NULL
} else if (c->argc >= 2 && !strcasecmp(c->argv[1]->ptr,"exists")) {
int j;
addReplyMultiBulkLen(c, c->argc-2);
addReplyArrayLen(c, c->argc-2);
for (j = 2; j < c->argc; j++) {
if (dictFind(server.lua_scripts,c->argv[j]->ptr))
addReply(c,shared.cone);
......
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