Merge branch 'unstable' of https://github.com/antirez/redis into Multithread

deps/hiredis/Makefile

@@ -23,6 +23,15 @@ INSTALL_INCLUDE_PATH= $(DESTDIR)$(PREFIX)/$(INCLUDE_PATH)
 INSTALL_LIBRARY_PATH= $(DESTDIR)$(PREFIX)/$(LIBRARY_PATH)
 INSTALL_PKGCONF_PATH= $(INSTALL_LIBRARY_PATH)/$(PKGCONF_PATH)
 
+# Installation related variables and target
+PREFIX?=/usr/local
+INCLUDE_PATH?=include/hiredis
+LIBRARY_PATH?=lib
+PKGCONF_PATH?=pkgconfig
+INSTALL_INCLUDE_PATH= $(DESTDIR)$(PREFIX)/$(INCLUDE_PATH)
+INSTALL_LIBRARY_PATH= $(DESTDIR)$(PREFIX)/$(LIBRARY_PATH)
+INSTALL_PKGCONF_PATH= $(INSTALL_LIBRARY_PATH)/$(PKGCONF_PATH)
+
 # keydb-server configuration used for testing
 REDIS_PORT=56379
 REDIS_SERVER=keydb-server

redis.conf

@@ -291,6 +291,17 @@ dir ./
 # refuse the replica request.
 #
 # masterauth <master-password>
+#
+# However this is not enough if you are using Redis ACLs (for Redis version
+# 6 or greater), and the default user is not capable of running the PSYNC
+# command and/or other commands needed for replication. In this case it's
+# better to configure a special user to use with replication, and specify the
+# masteruser configuration as such:
+#
+# masteruser <username>
+#
+# When masteruser is specified, the replica will authenticate against its
+# master using the new AUTH form: AUTH <username> <password>.
 
 # When a replica loses its connection with the master, or when the replication
 # is still in progress, the replica can act in two different ways:
@@ -501,6 +512,94 @@ replica-priority 100
 # can be easily a long string from /dev/urandom or whatever, so by using a
 # long and unguessable password no brute force attack will be possible.
 
+# Redis ACL users are defined in the following format:
+#
+#   user <username> ... acl rules ...
+#
+# For example:
+#
+#   user worker +@list +@connection ~jobs:* on >ffa9203c493aa99
+#
+# The special username "default" is used for new connections. If this user
+# has the "nopass" rule, then new connections will be immediately authenticated
+# as the "default" user without the need of any password provided via the
+# AUTH command. Otherwise if the "default" user is not flagged with "nopass"
+# the connections will start in not authenticated state, and will require
+# AUTH (or the HELLO command AUTH option) in order to be authenticated and
+# start to work.
+#
+# The ACL rules that describe what an user can do are the following:
+#
+#  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 are like @admin, @set, @sortedset, ...
+#               and so forth, see the full list in the server.c file where
+#               the Redis command table is described and defined.
+#               The special category @all means all the commands, but currently
+#               present in the server, and that will be loaded in the future
+#               via modules.
+#  +<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 "+".
+#  allcommands  Alias for +@all. Note that it implies the ability to execute
+#               all the future commands loaded via the modules system.
+#  nocommands   Alias for -@all.
+#  ~<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.
+#  allkeys      Alias for ~*
+#  resetkeys    Flush the list of allowed keys 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.
+#  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).
+#  reset        Performs the following actions: resetpass, resetkeys, off,
+#               -@all. The user returns to the same state it has immediately
+#               after its creation.
+#
+# ACL rules can be specified in any order: for instance you can start with
+# passwords, then flags, or key patterns. However note that the additive
+# and subtractive rules will CHANGE MEANING depending on the ordering.
+# For instance see the following example:
+#
+#   user alice on +@all -DEBUG ~* >somepassword
+#
+# This will allow "alice" to use all the commands with the exception of the
+# DEBUG command, since +@all added all the commands to the set of the commands
+# alice can use, and later DEBUG was removed. However if we invert the order
+# of two ACL rules the result will be different:
+#
+#   user alice on -DEBUG +@all ~* >somepassword
+#
+# Now DEBUG was removed when alice had yet no commands in the set of allowed
+# commands, later all the commands are added, so the user will be able to
+# execute everything.
+#
+# Basically ACL rules are processed left-to-right.
+#
+# For more information about ACL configuration please refer to
+# the Redis web site at https://redis.io/topics/acl
+
+# Using an external ACL file
+#
 # Instead of configuring users here in this file, it is possible to use
 # a stand-alone file just listing users. The two methods cannot be mixed:
 # if you configure users here and at the same time you activate the exteranl
@@ -1395,7 +1494,3 @@ rdb-save-incremental-fsync yes
 # the main dictionary scan
 # active-defrag-max-scan-fields 1000
 
-# Path to directory for file backed scratchpad.  The file backed scratchpad
-# reduces memory requirements by storing rarely accessed data on disk 
-# instead of RAM.  A temporary file will be created in this directory.
-# scratch-file-path /tmp/

src/Makefile

@@ -27,7 +27,7 @@ ifneq (,$(findstring FreeBSD,$(uname_S)))
   STD+=-Wno-c11-extensions
 endif
 endif
-WARN=-Wall -Wextra -Werror -W -Wno-missing-field-initializers
+WARN=-Wall -W -Wno-missing-field-initializers
 OPT=$(OPTIMIZATION)
 
 PREFIX?=/usr/local

src/acl.c

@@ -28,6 +28,7 @@
  */
 
 #include "server.h"
+#include <fcntl.h>
 
 /* =============================================================================
  * Global state for ACLs
@@ -90,6 +91,7 @@ struct ACLUserFlag {
 
 void ACLResetSubcommandsForCommand(user *u, unsigned long id);
 void ACLResetSubcommands(user *u);
+void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub);
 
 /* =============================================================================
  * Helper functions for the rest of the ACL implementation
@@ -163,6 +165,11 @@ void ACLListFreeSds(void *item) {
     sdsfree(item);
 }
 
+/* Method to duplicate list elements from ACL users password/ptterns lists. */
+void *ACLListDupSds(void *item) {
+    return sdsdup(item);
+}
+
 /* 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.
@@ -178,13 +185,32 @@ user *ACLCreateUser(const char *name, size_t namelen) {
     u->patterns = listCreate();
     listSetMatchMethod(u->passwords,ACLListMatchSds);
     listSetFreeMethod(u->passwords,ACLListFreeSds);
+    listSetDupMethod(u->passwords,ACLListDupSds);
     listSetMatchMethod(u->patterns,ACLListMatchSds);
     listSetFreeMethod(u->patterns,ACLListFreeSds);
+    listSetDupMethod(u->patterns,ACLListDupSds);
     memset(u->allowed_commands,0,sizeof(u->allowed_commands));
     raxInsert(Users,(unsigned char*)name,namelen,u,NULL);
     return u;
 }
 
+/* This function should be called when we need an unlinked "fake" user
+ * we can use in order to validate ACL rules or for other similar reasons.
+ * The user will not get linked to the Users radix tree. The returned
+ * user should be released with ACLFreeUser() as usually. */
+user *ACLCreateUnlinkedUser(void) {
+    char username[64];
+    for (int j = 0; ; j++) {
+        snprintf(username,sizeof(username),"__fakeuser:%d__",j);
+        user *fakeuser = ACLCreateUser(username,strlen(username));
+        if (fakeuser == NULL) continue;
+        int retval = raxRemove(Users,(unsigned char*) username,
+                               strlen(username),NULL);
+        serverAssert(retval != 0);
+        return fakeuser;
+    }
+}
+
 /* Release the memory used by the user structure. Note that this function
  * will not remove the user from the Users global radix tree. */
 void ACLFreeUser(user *u) {
@@ -195,6 +221,62 @@ void ACLFreeUser(user *u) {
     zfree(u);
 }
 
+/* When a user is deleted we need to cycle the active
+ * connections in order to kill all the pending ones that
+ * are authenticated with such user. */
+void ACLFreeUserAndKillClients(user *u) {
+    listIter li;
+    listNode *ln;
+    listRewind(server.clients,&li);
+    while ((ln = listNext(&li)) != NULL) {
+        client *c = listNodeValue(ln);
+        if (c->puser == u) {
+            /* We'll free the conenction asynchronously, so
+             * in theory to set a different user is not needed.
+             * However if there are bugs in Redis, soon or later
+             * this may result in some security hole: it's much
+             * more defensive to set the default user and put
+             * it in non authenticated mode. */
+            c->puser = DefaultUser;
+            c->authenticated = 0;
+            freeClientAsync(c);
+        }
+    }
+    ACLFreeUser(u);
+}
+
+/* Copy the user ACL rules from the source user 'src' to the destination
+ * user 'dst' so that at the end of the process they'll have exactly the
+ * same rules (but the names will continue to be the original ones). */
+void ACLCopyUser(user *dst, user *src) {
+    listRelease(dst->passwords);
+    listRelease(dst->patterns);
+    dst->passwords = listDup(src->passwords);
+    dst->patterns = listDup(src->patterns);
+    memcpy(dst->allowed_commands,src->allowed_commands,
+           sizeof(dst->allowed_commands));
+    dst->flags = src->flags;
+    ACLResetSubcommands(dst);
+    /* Copy the allowed subcommands array of array of SDS strings. */
+    if (src->allowed_subcommands) {
+        for (int j = 0; j < USER_COMMAND_BITS_COUNT; j++) {
+            if (src->allowed_subcommands[j]) {
+                for (int i = 0; src->allowed_subcommands[j][i]; i++)
+                {
+                    ACLAddAllowedSubcommand(dst, j,
+                        src->allowed_subcommands[j][i]);
+                }
+            }
+        }
+    }
+}
+
+/* Free all the users registered in the radix tree 'users' and free the
+ * radix tree itself. */
+void ACLFreeUsersSet(rax *users) {
+    raxFreeWithCallback(users,(void(*)(void*))ACLFreeUserAndKillClients);
+}
+
 /* Given a command ID, this function set by reference 'word' and 'bit'
  * so that user->allowed_commands[word] will address the right word
  * where the corresponding bit for the provided ID is stored, and
@@ -256,6 +338,7 @@ int ACLSetUserCommandBitsForCategory(user *u, const char *category, int value) {
     dictEntry *de;
     while ((de = dictNext(di)) != NULL) {
         struct redisCommand *cmd = dictGetVal(de);
+        if (cmd->flags & CMD_MODULE) continue; /* Ignore modules commands. */
         if (cmd->flags & cflag) {
             ACLSetUserCommandBit(u,cmd->id,value);
             ACLResetSubcommandsForCommand(u,cmd->id);
@@ -579,6 +662,7 @@ void ACLAddAllowedSubcommand(user *u, unsigned long id, const char *sub) {
  *         fully added.
  * EEXIST: You are adding a key pattern after "*" was already added. This is
  *         almost surely an error on the user side.
+ * ENODEV: The password you are trying to remove from the user does not exist.
  */
 int ACLSetUser(user *u, const char *op, ssize_t oplen) {
     if (oplen == -1) oplen = strlen(op);
@@ -623,8 +707,13 @@ int ACLSetUser(user *u, const char *op, ssize_t oplen) {
     } 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);
+        if (ln) {
+            listDelNode(u->passwords,ln);
+        } else {
+            errno = ENODEV;
+            return C_ERR;
+        }
     } else if (op[0] == '~') {
         if (u->flags & USER_FLAG_ALLKEYS) {
             errno = EEXIST;
@@ -728,6 +817,9 @@ char *ACLSetUserStringError(void) {
                  "'allkeys' flag) is not valid and does not have any "
                  "effect. Try 'resetkeys' to start with an empty "
                  "list of patterns";
+    else if (errno == ENODEV)
+        errmsg = "The password you are trying to remove from the user does "
+                 "not exist";
     return errmsg;
 }
 
@@ -741,10 +833,9 @@ sds ACLDefaultUserFirstPassword(void) {
     return listNodeValue(first);
 }
 
-/* Initialization of the ACL subsystem. */
+/* Initialize the default user, that will always exist for all the process
-void ACLInit(void) {
+ * lifetime. */
-    Users = raxNew();
+void ACLInitDefaultUser(void) {
-    UsersToLoad = listCreate();
     DefaultUser = ACLCreateUser("default",7);
     ACLSetUser(DefaultUser,"+@all",-1);
     ACLSetUser(DefaultUser,"~*",-1);
@@ -752,6 +843,13 @@ void ACLInit(void) {
     ACLSetUser(DefaultUser,"nopass",-1);
 }
 
+/* Initialization of the ACL subsystem. */
+void ACLInit(void) {
+    Users = raxNew();
+    UsersToLoad = listCreate();
+    ACLInitDefaultUser();
+}
+
 /* Check the username and password pair and return C_OK if they are valid,
  * otherwise C_ERR is returned and errno is set to:
  *
@@ -944,11 +1042,7 @@ int ACLAppendUserForLoading(sds *argv, int argc, int *argc_err) {
 
     /* Try to apply the user rules in a fake user to see if they
      * are actually valid. */
-    char *funame = "__fakeuser__";
+    user *fakeuser = ACLCreateUnlinkedUser();
-    user *fakeuser = ACLCreateUser(funame,strlen(funame));
-    serverAssert(fakeuser != NULL);
-    int retval = raxRemove(Users,(unsigned char*) funame,strlen(funame),NULL);
-    serverAssert(retval != 0);
 
     for (int j = 2; j < argc; j++) {
         if (ACLSetUser(fakeuser,argv[j],sdslen(argv[j])) == C_ERR) {
@@ -1009,14 +1103,275 @@ int ACLLoadConfiguredUsers(void) {
     return C_OK;
 }
 
+/* This function loads the ACL from the specified filename: every line
+ * is validated and shold be either empty or in the format used to specify
+ * users in the redis.conf configuration or in the ACL file, that is:
+ *
+ *  user <username> ... rules ...
+ *
+ * Note that this function considers comments starting with '#' as errors
+ * because the ACL file is meant to be rewritten, and comments would be
+ * lost after the rewrite. Yet empty lines are allowed to avoid being too
+ * strict.
+ *
+ * One important part of implementing ACL LOAD, that uses this function, is
+ * to avoid ending with broken rules if the ACL file is invalid for some
+ * reason, so the function will attempt to validate the rules before loading
+ * each user. For every line that will be found broken the function will
+ * collect an error message.
+ *
+ * IMPORTANT: If there is at least a single error, nothing will be loaded
+ * and the rules will remain exactly as they were.
+ *
+ * At the end of the process, if no errors were found in the whole file then
+ * NULL is returned. Otherwise an SDS string describing in a single line
+ * a description of all the issues found is returned. */
+sds ACLLoadFromFile(const char *filename) {
+    FILE *fp;
+    char buf[1024];
+
+    /* Open the ACL file. */
+    if ((fp = fopen(filename,"r")) == NULL) {
+        sds errors = sdscatprintf(sdsempty(),
+            "Error loading ACLs, opening file '%s': %s",
+            filename, strerror(errno));
+        return errors;
+    }
+
+    /* Load the whole file as a single string in memory. */
+    sds acls = sdsempty();
+    while(fgets(buf,sizeof(buf),fp) != NULL)
+        acls = sdscat(acls,buf);
+    fclose(fp);
+
+    /* Split the file into lines and attempt to load each line. */
+    int totlines;
+    sds *lines, errors = sdsempty();
+    lines = sdssplitlen(acls,strlen(acls),"\n",1,&totlines);
+    sdsfree(acls);
+
+    /* We need a fake user to validate the rules before making changes
+     * to the real user mentioned in the ACL line. */
+    user *fakeuser = ACLCreateUnlinkedUser();
+
+    /* We do all the loading in a fresh insteance of the Users radix tree,
+     * so if there are errors loading the ACL file we can rollback to the
+     * old version. */
+    rax *old_users = Users;
+    user *old_default_user = DefaultUser;
+    Users = raxNew();
+    ACLInitDefaultUser();
+
+    /* Load each line of the file. */
+    for (int i = 0; i < totlines; i++) {
+        sds *argv;
+        int argc;
+        int linenum = i+1;
+
+        lines[i] = sdstrim(lines[i]," \t\r\n");
+
+        /* Skip blank lines */
+        if (lines[i][0] == '\0') continue;
+
+        /* Split into arguments */
+        argv = sdssplitargs(lines[i],&argc);
+        if (argv == NULL) {
+            errors = sdscatprintf(errors,
+                     "%s:%d: unbalanced quotes in acl line. ",
+                     server.acl_filename, linenum);
+            continue;
+        }
+
+        /* Skip this line if the resulting command vector is empty. */
+        if (argc == 0) {
+            sdsfreesplitres(argv,argc);
+            continue;
+        }
+
+        /* The line should start with the "user" keyword. */
+        if (strcmp(argv[0],"user") || argc < 2) {
+            errors = sdscatprintf(errors,
+                     "%s:%d should start with user keyword followed "
+                     "by the username. ", server.acl_filename,
+                     linenum);
+            sdsfreesplitres(argv,argc);
+            continue;
+        }
+
+        /* Try to process the line using the fake user to validate iif
+         * the rules are able to apply cleanly. */
+        ACLSetUser(fakeuser,"reset",-1);
+        int j;
+        for (j = 2; j < argc; j++) {
+            if (ACLSetUser(fakeuser,argv[j],sdslen(argv[j])) != C_OK) {
+                char *errmsg = ACLSetUserStringError();
+                errors = sdscatprintf(errors,
+                         "%s:%d: %s. ",
+                         server.acl_filename, linenum, errmsg);
+                continue;
+            }
+        }
+
+        /* Apply the rule to the new users set only if so far there
+         * are no errors, otherwise it's useless since we are going
+         * to discard the new users set anyway. */
+        if (sdslen(errors) != 0) {
+            sdsfreesplitres(argv,argc);
+            continue;
+        }
+
+        /* We can finally lookup the user and apply the rule. If the
+         * user already exists we always reset it to start. */
+        user *u = ACLCreateUser(argv[1],sdslen(argv[1]));
+        if (!u) {
+            u = ACLGetUserByName(argv[1],sdslen(argv[1]));
+            serverAssert(u != NULL);
+            ACLSetUser(u,"reset",-1);
+        }
+
+        /* Note that the same rules already applied to the fake user, so
+         * we just assert that everything goess well: it should. */
+        for (j = 2; j < argc; j++)
+            serverAssert(ACLSetUser(u,argv[j],sdslen(argv[j])) == C_OK);
+
+        sdsfreesplitres(argv,argc);
+    }
+
+    ACLFreeUser(fakeuser);
+    sdsfreesplitres(lines,totlines);
+    DefaultUser = old_default_user; /* This pointer must never change. */
+
+    /* Check if we found errors and react accordingly. */
+    if (sdslen(errors) == 0) {
+        /* The default user pointer is referenced in different places: instead
+         * of replacing such occurrences it is much simpler to copy the new
+         * default user configuration in the old one. */
+        user *new = ACLGetUserByName("default",7);
+        serverAssert(new != NULL);
+        ACLCopyUser(DefaultUser,new);
+        ACLFreeUser(new);
+        raxInsert(Users,(unsigned char*)"default",7,DefaultUser,NULL);
+        raxRemove(old_users,(unsigned char*)"default",7,NULL);
+        ACLFreeUsersSet(old_users);
+        sdsfree(errors);
+        return NULL;
+    } else {
+        ACLFreeUsersSet(Users);
+        Users = old_users;
+        errors = sdscat(errors,"WARNING: ACL errors detected, no change to the previously active ACL rules was performed");
+        return errors;
+    }
+}
+
+/* Generate a copy of the ACLs currently in memory in the specified filename.
+ * Returns C_OK on success or C_ERR if there was an error during the I/O.
+ * When C_ERR is returned a log is produced with hints about the issue. */
+int ACLSaveToFile(const char *filename) {
+    sds acl = sdsempty();
+    int fd;
+
+    /* Let's generate an SDS string containing the new version of the
+     * ACL file. */
+    raxIterator ri;
+    raxStart(&ri,Users);
+    raxSeek(&ri,"^",NULL,0);
+    while(raxNext(&ri)) {
+        user *u = ri.data;
+        /* Return information in the configuration file format. */
+        sds user = sdsnew("user ");
+        user = sdscatsds(user,u->name);
+        user = sdscatlen(user," ",1);
+        sds descr = ACLDescribeUser(u);
+        user = sdscatsds(user,descr);
+        sdsfree(descr);
+        acl = sdscatsds(acl,user);
+        acl = sdscatlen(acl,"\n",1);
+        sdsfree(user);
+    }
+    raxStop(&ri);
+
+    /* Create a temp file with the new content. */
+    sds tmpfilename = sdsnew(filename);
+    tmpfilename = sdscatfmt(tmpfilename,".tmp-%i-%I",
+        (int)getpid(),(int)mstime());
+    if ((fd = open(tmpfilename,O_WRONLY|O_CREAT,0644)) == -1) {
+        serverLog(LL_WARNING,"Opening temp ACL file for ACL SAVE: %s",
+            strerror(errno));
+        sdsfree(tmpfilename);
+        sdsfree(acl);
+        return C_ERR;
+    }
+
+    /* Write it. */
+    if (write(fd,acl,sdslen(acl)) != (ssize_t)sdslen(acl)) {
+        serverLog(LL_WARNING,"Writing ACL file for ACL SAVE: %s",
+            strerror(errno));
+        close(fd);
+        unlink(tmpfilename);
+        sdsfree(tmpfilename);
+        sdsfree(acl);
+        return C_ERR;
+    }
+    close(fd);
+    sdsfree(acl);
+
+    /* Let's replace the new file with the old one. */
+    if (rename(tmpfilename,filename) == -1) {
+        serverLog(LL_WARNING,"Renaming ACL file for ACL SAVE: %s",
+            strerror(errno));
+        unlink(tmpfilename);
+        sdsfree(tmpfilename);
+        return C_ERR;
+    }
+
+    sdsfree(tmpfilename);
+    return C_OK;
+}
+
+/* This function is called once the server is already running, modules are
+ * loaded, and we are ready to start, in order to load the ACLs either from
+ * the pending list of users defined in redis.conf, or from the ACL file.
+ * The function will just exit with an error if the user is trying to mix
+ * both the loading methods. */
+void ACLLoadUsersAtStartup(void) {
+    if (server.acl_filename[0] != '\0' && listLength(UsersToLoad) != 0) {
+        serverLog(LL_WARNING,
+            "Configuring Redis with users defined in redis.conf and at "
+            "the same setting an ACL file path is invalid. This setup "
+            "is very likely to lead to configuration errors and security "
+            "holes, please define either an ACL file or declare users "
+            "directly in your redis.conf, but not both.");
+        exit(1);
+    }
+
+    if (ACLLoadConfiguredUsers() == C_ERR) {
+        serverLog(LL_WARNING,
+            "Critical error while loading ACLs. Exiting.");
+        exit(1);
+    }
+
+    if (server.acl_filename[0] != '\0') {
+        sds errors = ACLLoadFromFile(server.acl_filename);
+        if (errors) {
+            serverLog(LL_WARNING,
+                "Aborting Redis startup because of ACL errors: %s", errors);
+            sdsfree(errors);
+            exit(1);
+        }
+    }
+}
+
 /* =============================================================================
  * ACL related commands
  * ==========================================================================*/
 
 /* ACL -- show and modify the configuration of ACL users.
  * ACL HELP
+ * ACL LOAD
  * ACL LIST
- * ACL SETUSER <username> ... user attribs ...
+ * ACL USERS
+ * ACL CAT [<category>]
+ * ACL SETUSER <username> ... acl rules ...
  * ACL DELUSER <username>
  * ACL GETUSER <username>
  */
@@ -1050,27 +1405,7 @@ void aclCommand(client *c) {
                           sdslen(username),
                           (void**)&u))
             {
-                /* When a user is deleted we need to cycle the active
+                ACLFreeUserAndKillClients(u);
-                 * connections in order to kill all the pending ones that
-                 * are authenticated with such user. */
-                ACLFreeUser(u);
-                listIter li;
-                listNode *ln;
-                listRewind(server.clients,&li);
-                while ((ln = listNext(&li)) != NULL) {
-                    client *c = listNodeValue(ln);
-                    if (c->puser == u) {
-                        /* We'll free the conenction asynchronously, so
-                         * in theory to set a different user is not needed.
-                         * However if there are bugs in Redis, soon or later
-                         * this may result in some security hole: it's much
-                         * more defensive to set the default user and put
-                         * it in non authenticated mode. */
-                        c->puser = DefaultUser;
-                        c->authenticated = 0;
-                        freeClientAsync(c);
-                    }
-                }
                 deleted++;
             }
         }
@@ -1151,19 +1486,69 @@ void aclCommand(client *c) {
             }
         }
         raxStop(&ri);
-    } else if (!strcasecmp(sub,"whoami")) {
+    } else if (!strcasecmp(sub,"whoami") && c->argc == 2) {
         if (c->puser != NULL) {
             addReplyBulkCBuffer(c,c->puser->name,sdslen(c->puser->name));
         } else {
             addReplyNull(c);
         }
+    } else if (server.acl_filename[0] == '\0' &&
+               (!strcasecmp(sub,"load") || !strcasecmp(sub,"save")))
+    {
+        addReplyError(c,"This Redis instance is not configured to use an ACL file. You may want to specify users via the ACL SETUSER command and then issue a CONFIG REWRITE (assuming you have a Redis configuration file set) in order to store users in the Redis configuration.");
+        return;
+    } else if (!strcasecmp(sub,"load") && c->argc == 2) {
+        sds errors = ACLLoadFromFile(server.acl_filename);
+        if (errors == NULL) {
+            addReply(c,shared.ok);
+        } else {
+            addReplyError(c,errors);
+            sdsfree(errors);
+        }
+    } else if (!strcasecmp(sub,"save") && c->argc == 2) {
+        if (ACLSaveToFile(server.acl_filename) == C_OK) {
+            addReply(c,shared.ok);
+        } else {
+            addReplyError(c,"There was an error trying to save the ACLs. "
+                            "Please check the server logs for more "
+                            "information");
+        }
+    } else if (!strcasecmp(sub,"cat") && c->argc == 2) {
+        void *dl = addReplyDeferredLen(c);
+        int j;
+        for (j = 0; ACLCommandCategories[j].flag != 0; j++)
+            addReplyBulkCString(c,ACLCommandCategories[j].name);
+        setDeferredArrayLen(c,dl,j);
+    } else if (!strcasecmp(sub,"cat") && c->argc == 3) {
+        uint64_t cflag = ACLGetCommandCategoryFlagByName(ptrFromObj(c->argv[2]));
+        if (cflag == 0) {
+            addReplyErrorFormat(c, "Unknown category '%s'", (char*)ptrFromObj(c->argv[2]));
+            return;
+        }
+        int arraylen = 0;
+        void *dl = addReplyDeferredLen(c);
+        dictIterator *di = dictGetIterator(server.orig_commands);
+        dictEntry *de;
+        while ((de = dictNext(di)) != NULL) {
+            struct redisCommand *cmd = dictGetVal(de);
+            if (cmd->flags & CMD_MODULE) continue;
+            if (cmd->flags & cflag) {
+                addReplyBulkCString(c,cmd->name);
+                arraylen++;
+            }
+        }
+        dictReleaseIterator(di);
+        setDeferredArrayLen(c,dl,arraylen);
     } else if (!strcasecmp(sub,"help")) {
         const char *help[] = {
+"LOAD                              -- Reload users from the ACL file.",
 "LIST                              -- Show user details in config file format.",
 "USERS                             -- List all the registered usernames.",
 "SETUSER <username> [attribs ...]  -- Create or modify a user.",
 "GETUSER <username>                -- Get the user details.",
 "DELUSER <username>                -- Delete a user.",
+"CAT                               -- List available categories.",
+"CAT <category>                    -- List commands inside category.",
 "WHOAMI                            -- Return the current connection username.",
 NULL
         };
@@ -1172,3 +1557,15 @@ NULL
         addReplySubcommandSyntaxError(c);
     }
 }
+
+void addReplyCommandCategories(client *c, struct redisCommand *cmd) {
+    int flagcount = 0;
+    void *flaglen = addReplyDeferredLen(c);
+    for (int j = 0; ACLCommandCategories[j].flag != 0; j++) {
+        if (cmd->flags & ACLCommandCategories[j].flag) {
+            addReplyStatusFormat(c, "@%s", ACLCommandCategories[j].name);
+            flagcount++;
+        }
+    }
+    setDeferredSetLen(c, flaglen, flagcount);
+}

src/cluster.c

@@ -5460,6 +5460,10 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in
     int i, slot = 0, migrating_slot = 0, importing_slot = 0, missing_keys = 0;
     serverAssert(aeThreadOwnsLock());
 
+    /* Allow any key to be set if a module disabled cluster redirections. */
+    if (server.cluster_module_flags & CLUSTER_MODULE_FLAG_NO_REDIRECTION)
+        return myself;
+
     /* Allow any key to be set if a module disabled cluster redirections. */
     if (server.cluster_module_flags & CLUSTER_MODULE_FLAG_NO_REDIRECTION)
         return myself;

src/config.c

@@ -395,6 +395,9 @@ void loadServerConfigFromString(char *config) {
                 err = "repl-backlog-ttl can't be negative ";
                 goto loaderr;
             }
+        } else if (!strcasecmp(argv[0],"masteruser") && argc == 2) {
+            zfree(server.masteruser);
+            server.masteruser = argv[1][0] ? zstrdup(argv[1]) : NULL;
         } else if (!strcasecmp(argv[0],"masterauth") && argc == 2) {
             zfree(server.masterauth);
             server.masterauth = argv[1][0] ? zstrdup(argv[1]) : NULL;
@@ -953,6 +956,9 @@ void configSetCommand(client *c) {
         sds aclop = sdscatprintf(sdsempty(),">%s",(char*)ptrFromObj(o));
         ACLSetUser(DefaultUser,aclop,sdslen(aclop));
         sdsfree(aclop);
+    } config_set_special_field("masteruser") {
+        zfree(server.masteruser);
+        server.masteruser = ((char*)ptrFromObj(o))[0] ? zstrdup(ptrFromObj(o)) : NULL;
     } config_set_special_field("masterauth") {
         zfree(server.masterauth);
         server.masterauth = ((char*)ptrFromObj(o))[0] ? zstrdup(ptrFromObj(o)) : NULL;
@@ -1368,6 +1374,7 @@ void configGetCommand(client *c) {
 
     /* String values */
     config_get_string_field("dbfilename",server.rdb_filename);
+    config_get_string_field("masteruser",server.masteruser);
     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);
@@ -2246,6 +2253,7 @@ int rewriteConfig(char *path) {
     rewriteConfigDirOption(state);
     rewriteConfigSlaveofOption(state,"replicaof");
     rewriteConfigStringOption(state,"replica-announce-ip",server.slave_announce_ip,CONFIG_DEFAULT_SLAVE_ANNOUNCE_IP);
+    rewriteConfigStringOption(state,"masteruser",server.masteruser,NULL);
     rewriteConfigStringOption(state,"masterauth",server.masterauth,NULL);
     rewriteConfigStringOption(state,"cluster-announce-ip",server.cluster_announce_ip,NULL);
     rewriteConfigYesNoOption(state,"replica-serve-stale-data",server.repl_serve_stale_data,CONFIG_DEFAULT_SLAVE_SERVE_STALE_DATA);

src/debug.c

@@ -803,7 +803,7 @@ static void *getMcontextEip(ucontext_t *uc) {
     #endif
 #elif defined(__linux__)
     /* Linux */
-    #if defined(__i386__)
+    #if defined(__i386__) || defined(__ILP32__)
     return (void*) uc->uc_mcontext.gregs[14]; /* Linux 32 */
     #elif defined(__X86_64__) || defined(__x86_64__)
     return (void*) uc->uc_mcontext.gregs[16]; /* Linux 64 */
@@ -915,7 +915,7 @@ void logRegisters(ucontext_t *uc) {
 /* Linux */
 #elif defined(__linux__)
     /* Linux x86 */
-    #if defined(__i386__)
+    #if defined(__i386__) || defined(__ILP32__)
     serverLog(LL_WARNING,
     "\n"
     "EAX:%08lx EBX:%08lx ECX:%08lx EDX:%08lx\n"

src/dict.c

@@ -739,6 +739,30 @@ unsigned int dictGetSomeKeys(dict *d, dictEntry **des, unsigned int count) {
     return stored;
 }
 
+/* This is like dictGetRandomKey() from the POV of the API, but will do more
+ * work to ensure a better distribution of the returned element.
+ *
+ * This function improves the distribution because the dictGetRandomKey()
+ * problem is that it selects a random bucket, then it selects a random
+ * element from the chain in the bucket. However elements being in different
+ * chain lengths will have different probabilities of being reported. With
+ * this function instead what we do is to consider a "linear" range of the table
+ * that may be constituted of N buckets with chains of different lengths
+ * appearing one after the other. Then we report a random element in the range.
+ * In this way we smooth away the problem of different chain lenghts. */
+#define GETFAIR_NUM_ENTRIES 15
+dictEntry *dictGetFairRandomKey(dict *d) {
+    dictEntry *entries[GETFAIR_NUM_ENTRIES];
+    unsigned int count = dictGetSomeKeys(d,entries,GETFAIR_NUM_ENTRIES);
+    /* Note that dictGetSomeKeys() may return zero elements in an unlucky
+     * run() even if there are actually elements inside the hash table. So
+     * when we get zero, we call the true dictGetRandomKey() that will always
+     * yeld the element if the hash table has at least one. */
+    if (count == 0) return dictGetRandomKey(d);
+    unsigned int idx = rand() % count;
+    return entries[idx];
+}
+
 /* Function to reverse bits. Algorithm from:
  * http://graphics.stanford.edu/~seander/bithacks.html#ReverseParallel */
 static unsigned long rev(unsigned long v) {

src/dict.h

@@ -170,6 +170,7 @@ dictIterator *dictGetSafeIterator(dict *d);
 dictEntry *dictNext(dictIterator *iter);
 void dictReleaseIterator(dictIterator *iter);
 dictEntry *dictGetRandomKey(dict *d);
+dictEntry *dictGetFairRandomKey(dict *d);
 unsigned int dictGetSomeKeys(dict *d, dictEntry **des, unsigned int count);
 void dictGetStats(char *buf, size_t bufsize, dict *d);
 uint64_t dictGenHashFunction(const void *key, int len);

src/redis-benchmark.c

@@ -39,6 +39,7 @@
 #include <sys/time.h>
 #include <signal.h>
 #include <assert.h>
+#include <math.h>
 
 #include <sds.h> /* Use hiredis sds. */
 #include "ae.h"
@@ -49,6 +50,7 @@
 
 #define UNUSED(V) ((void) V)
 #define RANDPTR_INITIAL_SIZE 8
+#define MAX_LATENCY_PRECISION 3
 
 static struct config {
     aeEventLoop *el;
@@ -80,6 +82,7 @@ static struct config {
     sds dbnumstr;
     char *tests;
     char *auth;
+    int precision;
 } config;
 
 typedef struct _client {
@@ -429,8 +432,19 @@ static int compareLatency(const void *a, const void *b) {
     return (*(long long*)a)-(*(long long*)b);
 }
 
+static int ipow(int base, int exp) {
+    int result = 1;
+    while (exp) {
+        if (exp & 1) result *= base;
+        exp /= 2;
+        base *= base;
+    }
+    return result;
+}
+
 static void showLatencyReport(void) {
     int i, curlat = 0;
+    int usbetweenlat = ipow(10, MAX_LATENCY_PRECISION-config.precision);
     float perc, reqpersec;
 
     reqpersec = (float)config.requests_finished/((float)config.totlatency/1000);
@@ -445,10 +459,21 @@ static void showLatencyReport(void) {
 
         qsort(config.latency,config.requests,sizeof(long long),compareLatency);
         for (i = 0; i < config.requests; i++) {
-            if (config.latency[i]/1000 != curlat || i == (config.requests-1)) {
+            if (config.latency[i]/usbetweenlat != curlat ||
-                curlat = config.latency[i]/1000;
+                i == (config.requests-1))
+            {
+                curlat = config.latency[i]/usbetweenlat;
                 perc = ((float)(i+1)*100)/config.requests;
-                printf("%.2f%% <= %d milliseconds\n", perc, curlat);
+                printf("%.2f%% <= %.*f milliseconds\n", perc, config.precision,
+                    curlat/pow(10.0, config.precision));
+
+                /* After the 2 milliseconds latency to have percentages split
+                 * by decimals will just add a lot of noise to the output. */
+                if (config.latency[i] > 2000) {
+                    config.precision = 0;
+                    usbetweenlat = ipow(10,
+                        MAX_LATENCY_PRECISION-config.precision);
+                }
             }
         }
         printf("%.2f requests per second\n\n", reqpersec);
@@ -547,6 +572,11 @@ int parseOptions(int argc, const char **argv) {
             if (lastarg) goto invalid;
             config.dbnum = atoi(argv[++i]);
             config.dbnumstr = sdsfromlonglong(config.dbnum);
+        } else if (!strcmp(argv[i],"--precision")) {
+            if (lastarg) goto invalid;
+            config.precision = atoi(argv[++i]);
+            if (config.precision < 0) config.precision = 0;
+            if (config.precision > MAX_LATENCY_PRECISION) config.precision = MAX_LATENCY_PRECISION;
         } else if (!strcmp(argv[i],"--help")) {
             exit_status = 0;
             goto usage;
@@ -586,6 +616,7 @@ usage:
 " -e                 If server replies with errors, show them on stdout.\n"
 "                    (no more than 1 error per second is displayed)\n"
 " -q                 Quiet. Just show query/sec values\n"
+" --precision        Number of decimal places to display in latency output (default 0)\n"
 " --csv              Output in CSV format\n"
 " -l                 Loop. Run the tests forever\n"
 " -t <tests>         Only run the comma separated list of tests. The test\n"
@@ -682,6 +713,7 @@ int main(int argc, const char **argv) {
     config.tests = NULL;
     config.dbnum = 0;
     config.auth = NULL;
+    config.precision = 1;
 
     i = parseOptions(argc,argv);
     argc -= i;

src/redis-cli.c

@@ -211,6 +211,8 @@ static struct config {
     char *pattern;
     char *rdb_filename;
     int bigkeys;
+    int memkeys;
+    unsigned memkeys_samples;
     int hotkeys;
     int stdinarg; /* get last arg from stdin. (-x option) */
     char *auth;
@@ -1337,6 +1339,12 @@ static int parseOptions(int argc, char **argv) {
             config.pipe_timeout = atoi(argv[++i]);
         } else if (!strcmp(argv[i],"--bigkeys")) {
             config.bigkeys = 1;
+        } else if (!strcmp(argv[i],"--memkeys")) {
+            config.memkeys = 1;
+            config.memkeys_samples = 0; /* use redis default */
+        } else if (!strcmp(argv[i],"--memkeys-samples")) {
+            config.memkeys = 1;
+            config.memkeys_samples = atoi(argv[++i]);
         } else if (!strcmp(argv[i],"--hotkeys")) {
             config.hotkeys = 1;
         } else if (!strcmp(argv[i],"--eval") && !lastarg) {
@@ -1535,7 +1543,10 @@ static void usage(void) {
 "  --pipe-timeout <n> In --pipe mode, abort with error if after sending all data.\n"
 "                     no reply is received within <n> seconds.\n"
 "                     Default timeout: %d. Use 0 to wait forever.\n"
-"  --bigkeys          Sample Redis keys looking for big keys.\n"
+"  --bigkeys          Sample Redis keys looking for keys with many elements (complexity).\n"
+"  --memkeys          Sample Redis keys looking for keys consuming a lot of memory.\n"
+"  --memkeys-samples <n> Sample Redis keys looking for keys consuming a lot of memory.\n"
+"                     And define number of key elements to sample\n"
 "  --hotkeys          Sample Redis keys looking for hot keys.\n"
 "                     only works when maxmemory-policy is *lfu.\n"
 "  --scan             List all keys using the SCAN command.\n"
@@ -6142,9 +6153,31 @@ static void latencyDistMode(void) {
  * Slave mode
  *--------------------------------------------------------------------------- */
 
+#define RDB_EOF_MARK_SIZE 40
+
+void sendReplconf(const char* arg1, const char* arg2) {
+    printf("sending REPLCONF %s %s\n", arg1, arg2);
+    redisReply *reply = redisCommand(context, "REPLCONF %s %s", arg1, arg2);
+
+    /* Handle any error conditions */
+    if(reply == NULL) {
+        fprintf(stderr, "\nI/O error\n");
+        exit(1);
+    } else if(reply->type == REDIS_REPLY_ERROR) {
+        fprintf(stderr, "REPLCONF %s error: %s\n", arg1, reply->str);
+        /* non fatal, old versions may not support it */
+    }
+    freeReplyObject(reply);
+}
+
+void sendCapa() {
+    sendReplconf("capa", "eof");
+}
+
 /* Sends SYNC and reads the number of bytes in the payload. Used both by
- * slaveMode() and getRDB(). */
+ * slaveMode() and getRDB().
-unsigned long long sendSync(int fd) {
+ * returns 0 in case an EOF marker is used. */
+unsigned long long sendSync(int fd, char *out_eof) {
     /* To start we need to send the SYNC command and return the payload.
      * The hiredis client lib does not understand this part of the protocol
      * and we don't want to mess with its buffers, so everything is performed
@@ -6174,17 +6207,33 @@ unsigned long long sendSync(int fd) {
         printf("SYNC with master failed: %s\n", buf);
         exit(1);
     }
+    if (strncmp(buf+1,"EOF:",4) == 0 && strlen(buf+5) >= RDB_EOF_MARK_SIZE) {
+        memcpy(out_eof, buf+5, RDB_EOF_MARK_SIZE);
+        return 0;
+    }
     return strtoull(buf+1,NULL,10);
 }
 
 static void slaveMode(void) {
     int fd = context->fd;
-    unsigned long long payload = sendSync(fd);
+    static char eofmark[RDB_EOF_MARK_SIZE];
+    static char lastbytes[RDB_EOF_MARK_SIZE];
+    static int usemark = 0;
+    unsigned long long payload = sendSync(fd, eofmark);
     char buf[1024];
     int original_output = config.output;
 
+    if (payload == 0) {
+        payload = ULLONG_MAX;
+        memset(lastbytes,0,RDB_EOF_MARK_SIZE);
+        usemark = 1;
+        fprintf(stderr,"SYNC with master, discarding "
+                       "bytes of bulk transfer until EOF marker...\n");
+    } else {
         fprintf(stderr,"SYNC with master, discarding %llu "
                        "bytes of bulk transfer...\n", payload);
+    }
+
 
     /* Discard the payload. */
     while(payload) {
@@ -6196,7 +6245,28 @@ static void slaveMode(void) {
             exit(1);
         }
         payload -= nread;
+
+        if (usemark) {
+            /* Update the last bytes array, and check if it matches our delimiter.*/
+            if (nread >= RDB_EOF_MARK_SIZE) {
+                memcpy(lastbytes,buf+nread-RDB_EOF_MARK_SIZE,RDB_EOF_MARK_SIZE);
+            } else {
+                int rem = RDB_EOF_MARK_SIZE-nread;
+                memmove(lastbytes,lastbytes+nread,rem);
+                memcpy(lastbytes+rem,buf,nread);
             }
+            if (memcmp(lastbytes,eofmark,RDB_EOF_MARK_SIZE) == 0)
+                break;
+        }
+    }
+
+    if (usemark) {
+        unsigned long long offset = ULLONG_MAX - payload;
+        fprintf(stderr,"SYNC done after %llu bytes. Logging commands from master.\n", offset);
+        /* put the slave online */
+        sleep(1);
+        sendReplconf("ACK", "0");
+    } else
         fprintf(stderr,"SYNC done. Logging commands from master.\n");
 
     /* Now we can use hiredis to read the incoming protocol. */
@@ -6214,11 +6284,22 @@ static void slaveMode(void) {
 static void getRDB(void) {
     int s = context->fd;
     int fd;
-    unsigned long long payload = sendSync(s);
+    static char eofmark[RDB_EOF_MARK_SIZE];
+    static char lastbytes[RDB_EOF_MARK_SIZE];
+    static int usemark = 0;
+    unsigned long long payload = sendSync(s, eofmark);
     char buf[4096];
 
+    if (payload == 0) {
+        payload = ULLONG_MAX;
+        memset(lastbytes,0,RDB_EOF_MARK_SIZE);
+        usemark = 1;
+        fprintf(stderr,"SYNC sent to master, writing bytes of bulk transfer until EOF marker to '%s'\n",
+            config.rdb_filename);
+    } else {
         fprintf(stderr,"SYNC sent to master, writing %llu bytes to '%s'\n",
             payload, config.rdb_filename);
+    }
 
     /* Write to file. */
     if (!strcmp(config.rdb_filename,"-")) {
@@ -6247,11 +6328,31 @@ static void getRDB(void) {
             exit(1);
         }
         payload -= nread;
+
+        if (usemark) {
+            /* Update the last bytes array, and check if it matches our delimiter.*/
+            if (nread >= RDB_EOF_MARK_SIZE) {
+                memcpy(lastbytes,buf+nread-RDB_EOF_MARK_SIZE,RDB_EOF_MARK_SIZE);
+            } else {
+                int rem = RDB_EOF_MARK_SIZE-nread;
+                memmove(lastbytes,lastbytes+nread,rem);
+                memcpy(lastbytes+rem,buf,nread);
+            }
+            if (memcmp(lastbytes,eofmark,RDB_EOF_MARK_SIZE) == 0)
+                break;
+        }
+    }
+    if (usemark) {
+        payload = ULLONG_MAX - payload - RDB_EOF_MARK_SIZE;
+        if (ftruncate(fd, payload) == -1)
+            fprintf(stderr,"ftruncate failed: %s.\n", strerror(errno));
+        fprintf(stderr,"Transfer finished with success after %llu bytes\n", payload);
+    } else {
+        fprintf(stderr,"Transfer finished with success.\n");
     }
     close(s); /* Close the file descriptor ASAP as fsync() may take time. */
     fsync(fd);
     close(fd);
-    fprintf(stderr,"Transfer finished with success.\n");
     exit(0);
 }
 
@@ -6419,15 +6520,6 @@ static void pipeMode(void) {
  * Find big keys
  *--------------------------------------------------------------------------- */
 
-#define TYPE_STRING 0
-#define TYPE_LIST   1
-#define TYPE_SET    2
-#define TYPE_HASH   3
-#define TYPE_ZSET   4
-#define TYPE_STREAM 5
-#define TYPE_NONE   6
-#define TYPE_COUNT  7
-
 static redisReply *sendScan(unsigned long long *it) {
     redisReply *reply = redisCommand(context, "SCAN %llu", *it);
 
@@ -6474,28 +6566,51 @@ static int getDbSize(void) {
     return size;
 }
 
-static int toIntType(char *key, char *type) {
+typedef struct {
-    if(!strcmp(type, "string")) {
+    char *name;
-        return TYPE_STRING;
+    char *sizecmd;
-    } else if(!strcmp(type, "list")) {
+    char *sizeunit;
-        return TYPE_LIST;
+    unsigned long long biggest;
-    } else if(!strcmp(type, "set")) {
+    unsigned long long count;
-        return TYPE_SET;
+    unsigned long long totalsize;
-    } else if(!strcmp(type, "hash")) {
+    sds biggest_key;
-        return TYPE_HASH;
+} typeinfo;
-    } else if(!strcmp(type, "zset")) {
+
-        return TYPE_ZSET;
+typeinfo type_string = { "string", "STRLEN", "bytes" };
-    } else if(!strcmp(type, "stream")) {
+typeinfo type_list = { "list", "LLEN", "items" };
-        return TYPE_STREAM;
+typeinfo type_set = { "set", "SCARD", "members" };
-    } else if(!strcmp(type, "none")) {
+typeinfo type_hash = { "hash", "HLEN", "fields" };
-        return TYPE_NONE;
+typeinfo type_zset = { "zset", "ZCARD", "members" };
-    } else {
+typeinfo type_stream = { "stream", "XLEN", "entries" };
-        fprintf(stderr, "Unknown type '%s' for key '%s'\n", type, key);
+typeinfo type_other = { "other", NULL, "?" };
-        exit(1);
+
-    }
+static typeinfo* typeinfo_add(dict *types, char* name, typeinfo* type_template) {
+    typeinfo *info = zmalloc(sizeof(typeinfo), MALLOC_LOCAL);
+    *info = *type_template;
+    info->name = sdsnew(name);
+    dictAdd(types, info->name, info);
+    return info;
 }
 
-static void getKeyTypes(redisReply *keys, int *types) {
+void type_free(void* priv_data, void* val) {
+    typeinfo *info = val;
+    UNUSED(priv_data);
+    if (info->biggest_key)
+        sdsfree(info->biggest_key);
+    sdsfree(info->name);
+    zfree(info);
+}
+
+static dictType typeinfoDictType = {
+    dictSdsHash,               /* hash function */
+    NULL,                      /* key dup */
+    NULL,                      /* val dup */
+    dictSdsKeyCompare,         /* key compare */
+    NULL,                      /* key destructor (owned by the value)*/
+    type_free                  /* val destructor */
+};
+
+static void getKeyTypes(dict *types_dict, redisReply *keys, typeinfo **types) {
     redisReply *reply;
     unsigned int i;
 
@@ -6521,32 +6636,47 @@ static void getKeyTypes(redisReply *keys, int *types) {
             exit(1);
         }
 
-        types[i] = toIntType(keys->element[i]->str, reply->str);
+        sds typereply = sdsnew(reply->str);
+        dictEntry *de = dictFind(types_dict, typereply);
+        sdsfree(typereply);
+        typeinfo *type = NULL;
+        if (de)
+            type = dictGetVal(de);
+        else if (strcmp(reply->str, "none")) /* create new types for modules, (but not for deleted keys) */
+            type = typeinfo_add(types_dict, reply->str, &type_other);
+        types[i] = type;
         freeReplyObject(reply);
     }
 }
 
-static void getKeySizes(redisReply *keys, int *types,
+static void getKeySizes(redisReply *keys, typeinfo **types,
-                        unsigned long long *sizes)
+                        unsigned long long *sizes, int memkeys,
+                        unsigned memkeys_samples)
 {
     redisReply *reply;
-    char *sizecmds[] = {"STRLEN","LLEN","SCARD","HLEN","ZCARD"};
     unsigned int i;
 
     /* Pipeline size commands */
     for(i=0;i<keys->elements;i++) {
-        /* Skip keys that were deleted */
+        /* Skip keys that disappeared between SCAN and TYPE (or unknown types when not in memkeys mode) */
-        if(types[i]==TYPE_NONE)
+        if(!types[i] || (!types[i]->sizecmd && !memkeys))
             continue;
 
-        redisAppendCommand(context, "%s %s", sizecmds[types[i]],
+        if (!memkeys)
-            keys->element[i]->str);
+            redisAppendCommand(context, "%s %s",
+                types[i]->sizecmd, keys->element[i]->str);
+        else if (memkeys_samples==0)
+            redisAppendCommand(context, "%s %s %s",
+                "MEMORY", "USAGE", keys->element[i]->str);
+        else
+            redisAppendCommand(context, "%s %s %s SAMPLES %u",
+                "MEMORY", "USAGE", keys->element[i]->str, memkeys_samples);
     }
 
     /* Retrieve sizes */
     for(i=0;i<keys->elements;i++) {
-        /* Skip keys that disappeared between SCAN and TYPE */
+        /* Skip keys that disappeared between SCAN and TYPE (or unknown types when not in memkeys mode) */
-        if(types[i] == TYPE_NONE) {
+        if(!types[i] || (!types[i]->sizecmd && !memkeys)) {
             sizes[i] = 0;
             continue;
         }
@@ -6561,7 +6691,8 @@ static void getKeySizes(redisReply *keys, int *types,
              * added as a different type between TYPE and SIZE */
             fprintf(stderr,
                 "Warning:  %s on '%s' failed (may have changed type)\n",
-                 sizecmds[types[i]], keys->element[i]->str);
+                !memkeys? types[i]->sizecmd: "MEMORY USAGE",
+                keys->element[i]->str);
             sizes[i] = 0;
         } else {
             sizes[i] = reply->integer;
@@ -6571,17 +6702,23 @@ static void getKeySizes(redisReply *keys, int *types,
     }
 }
 
-static void findBigKeys(void) {
+static void findBigKeys(int memkeys, unsigned memkeys_samples) {
-    unsigned long long biggest[TYPE_COUNT] = {0}, counts[TYPE_COUNT] = {0}, totalsize[TYPE_COUNT] = {0};
     unsigned long long sampled = 0, total_keys, totlen=0, *sizes=NULL, it=0;
-    sds maxkeys[TYPE_COUNT] = {0};
-    char *typename[] = {"string","list","set","hash","zset","stream","none"};
-    char *typeunit[] = {"bytes","items","members","fields","members","entries",""};
     redisReply *reply, *keys;
     unsigned int arrsize=0, i;
-    int type, *types=NULL;
+    dictIterator *di;
+    dictEntry *de;
+    typeinfo **types = NULL;
     double pct;
 
+    dict *types_dict = dictCreate(&typeinfoDictType, NULL);
+    typeinfo_add(types_dict, "string", &type_string);
+    typeinfo_add(types_dict, "list", &type_list);
+    typeinfo_add(types_dict, "set", &type_set);
+    typeinfo_add(types_dict, "hash", &type_hash);
+    typeinfo_add(types_dict, "zset", &type_zset);
+    typeinfo_add(types_dict, "stream", &type_stream);
+
     /* Total keys pre scanning */
     total_keys = getDbSize();
 
@@ -6590,15 +6727,6 @@ static void findBigKeys(void) {
     printf("# average sizes per key type.  You can use -i 0.1 to sleep 0.1 sec\n");
     printf("# per 100 SCAN commands (not usually needed).\n\n");
 
-    /* New up sds strings to keep track of overall biggest per type */
-    for(i=0;i<TYPE_NONE; i++) {
-        maxkeys[i] = sdsempty();
-        if(!maxkeys[i]) {
-            fprintf(stderr, "Failed to allocate memory for largest key names!\n");
-            exit(1);
-        }
-    }
-
     /* SCAN loop */
     do {
         /* Calculate approximate percentage completion */
@@ -6622,34 +6750,38 @@ static void findBigKeys(void) {
         }
 
         /* Retrieve types and then sizes */
-        getKeyTypes(keys, types);
+        getKeyTypes(types_dict, keys, types);
-        getKeySizes(keys, types, sizes);
+        getKeySizes(keys, types, sizes, memkeys, memkeys_samples);
 
         /* Now update our stats */
         for(i=0;i<keys->elements;i++) {
-            if((type = types[i]) == TYPE_NONE)
+            typeinfo *type = types[i];
+            /* Skip keys that disappeared between SCAN and TYPE */
+            if(!type)
                 continue;
 
-            totalsize[type] += sizes[i];
+            type->totalsize += sizes[i];
-            counts[type]++;
+            type->count++;
             totlen += keys->element[i]->len;
             sampled++;
 
-            if(biggest[type]<sizes[i]) {
+            if(type->biggest<sizes[i]) {
                 printf(
                    "[%05.2f%%] Biggest %-6s found so far '%s' with %llu %s\n",
-                   pct, typename[type], keys->element[i]->str, sizes[i],
+                   pct, type->name, keys->element[i]->str, sizes[i],
-                   typeunit[type]);
+                   !memkeys? type->sizeunit: "bytes");
 
                 /* Keep track of biggest key name for this type */
-                maxkeys[type] = sdscpy(maxkeys[type], keys->element[i]->str);
+                if (type->biggest_key)
-                if(!maxkeys[type]) {
+                    sdsfree(type->biggest_key);
+                type->biggest_key = sdsnew(keys->element[i]->str);
+                if(!type->biggest_key) {
                     fprintf(stderr, "Failed to allocate memory for key!\n");
                     exit(1);
                 }
 
                 /* Keep track of the biggest size for this type */
-                biggest[type] = sizes[i];
+                type->biggest = sizes[i];
             }
 
             /* Update overall progress */
@@ -6677,26 +6809,29 @@ static void findBigKeys(void) {
        totlen, totlen ? (double)totlen/sampled : 0);
 
     /* Output the biggest keys we found, for types we did find */
-    for(i=0;i<TYPE_NONE;i++) {
+    di = dictGetIterator(types_dict);
-        if(sdslen(maxkeys[i])>0) {
+    while ((de = dictNext(di))) {
-            printf("Biggest %6s found '%s' has %llu %s\n", typename[i], maxkeys[i],
+        typeinfo *type = dictGetVal(de);
-               biggest[i], typeunit[i]);
+        if(type->biggest_key) {
+            printf("Biggest %6s found '%s' has %llu %s\n", type->name, type->biggest_key,
+               type->biggest, !memkeys? type->sizeunit: "bytes");
         }
     }
+    dictReleaseIterator(di);
 
     printf("\n");
 
-    for(i=0;i<TYPE_NONE;i++) {
+    di = dictGetIterator(types_dict);
+    while ((de = dictNext(di))) {
+        typeinfo *type = dictGetVal(de);
         printf("%llu %ss with %llu %s (%05.2f%% of keys, avg size %.2f)\n",
-           counts[i], typename[i], totalsize[i], typeunit[i],
+           type->count, type->name, type->totalsize, !memkeys? type->sizeunit: "bytes",
-           sampled ? 100 * (double)counts[i]/sampled : 0,
+           sampled ? 100 * (double)type->count/sampled : 0,
-           counts[i] ? (double)totalsize[i]/counts[i] : 0);
+           type->count ? (double)type->totalsize/type->count : 0);
     }
+    dictReleaseIterator(di);
 
-    /* Free sds strings containing max keys */
+    dictRelease(types_dict);
-    for(i=0;i<TYPE_NONE;i++) {
-        sdsfree(maxkeys[i]);
-    }
 
     /* Success! */
     exit(0);
@@ -7271,12 +7406,14 @@ int main(int argc, char **argv) {
     /* Slave mode */
     if (config.slave_mode) {
         if (cliConnect(0) == REDIS_ERR) exit(1);
+        sendCapa();
         slaveMode();
     }
 
     /* Get RDB mode. */
     if (config.getrdb_mode) {
         if (cliConnect(0) == REDIS_ERR) exit(1);
+        sendCapa();
         getRDB();
     }
 
@@ -7289,7 +7426,19 @@ int main(int argc, char **argv) {
     /* Find big keys */
     if (config.bigkeys) {
         if (cliConnect(0) == REDIS_ERR) exit(1);
-        findBigKeys();
+        findBigKeys(0, 0);
+    }
+
+    /* Find large keys */
+    if (config.memkeys) {
+        if (cliConnect(0) == REDIS_ERR) exit(1);
+        findBigKeys(1, config.memkeys_samples);
+    }
+
+    /* Find hot keys */
+    if (config.hotkeys) {
+        if (cliConnect(0) == REDIS_ERR) exit(1);
+        findHotKeys();
     }
 
     /* Find hot keys */

src/replication.cpp

@@ -186,6 +186,13 @@ void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
      * master replication history and has the same backlog and offsets). */
     if (server.masterhost != NULL) return;
 
+    /* If the instance is not a top level master, return ASAP: we'll just proxy
+     * the stream of data we receive from our master instead, in order to
+     * propagate *identical* replication stream. In this way this slave can
+     * advertise the same replication ID as the master (since it shares the
+     * master replication history and has the same backlog and offsets). */
+    if (server.masterhost != NULL) return;
+
     /* If there aren't slaves, and there is no backlog buffer to populate,
      * we can return ASAP. */
     if (server.repl_backlog == NULL && listLength(slaves) == 0) return;
@@ -1695,7 +1702,13 @@ void syncWithMaster(aeEventLoop *el, int fd, void *privdata, int mask) {
 
     /* AUTH with the master if required. */
     if (server.repl_state == REPL_STATE_SEND_AUTH) {
-        if (server.masterauth) {
+        if (server.masteruser && server.masterauth) {
+            err = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"AUTH",
+                                         server.masteruser,server.masterauth,NULL);
+            if (err) goto write_error;
+            server.repl_state = REPL_STATE_RECEIVE_AUTH;
+            return;
+        } else if (server.masterauth) {
             err = sendSynchronousCommand(SYNC_CMD_WRITE,fd,"AUTH",server.masterauth,NULL);
             if (err) goto write_error;
             server.repl_state = REPL_STATE_RECEIVE_AUTH;
@@ -2045,8 +2058,6 @@ void replicationHandleMasterDisconnection(void) {
 }
 
 void replicaofCommand(client *c) {
-    // Changing the master needs to be done on the main thread.
-
     /* SLAVEOF is not allowed in cluster mode as replication is automatically
      * configured using the current address of the master node. */
     if (server.cluster_enabled) {

src/server.c

@@ -661,7 +661,7 @@ struct redisCommand redisCommandTable[] = {
      0,NULL,0,0,0,0,0,0},
 
     {"lastsave",lastsaveCommand,1,
-     "read-only random fast @admin",
+     "read-only random fast @admin @dangerous",
      0,NULL,0,0,0,0,0,0},
 
     {"type",typeCommand,2,
@@ -2908,6 +2908,16 @@ void initServer(void) {
         exit(1);
     }
 
+    /* Register a readable event for the pipe used to awake the event loop
+     * when a blocked client in a module needs attention. */
+    if (aeCreateFileEvent(server.rgthreadvar[IDX_EVENT_LOOP_MAIN].el, server.module_blocked_pipe[0], AE_READABLE,
+        moduleBlockedClientPipeReadable,NULL) == AE_ERR) {
+            serverPanic(
+                "Error registering the readable event for the module "
+                "blocked clients subsystem.");
+    }
+
+
     /* Register a readable event for the pipe used to awake the event loop
      * when a blocked client in a module needs attention. */
     if (aeCreateFileEvent(server.rgthreadvar[IDX_EVENT_LOOP_MAIN].el, server.module_blocked_pipe[0], AE_READABLE,
@@ -2998,10 +3008,10 @@ int populateCommandTableParseFlags(struct redisCommand *c, char *strflags) {
                 return C_ERR;
             }
         }
-
+    }
     /* If it's not @fast is @slow in this binary world. */
     if (!(c->flags & CMD_CATEGORY_FAST)) c->flags |= CMD_CATEGORY_SLOW;
-    }
+
     sdsfreesplitres(argv,argc);
     return C_OK;
 }
@@ -3390,15 +3400,18 @@ int processCommand(client *c) {
         return C_OK;
     }
 
-    /* Check if the user is authenticated */
+    /* Check if the user is authenticated. This check is skipped in case
-    if (!(DefaultUser->flags & USER_FLAG_NOPASS) &&
+     * the default user is flagged as "nopass" and is active. */
-        !c->authenticated &&
+    int auth_required = !(DefaultUser->flags & USER_FLAG_NOPASS) &&
-        (c->cmd->proc != authCommand || c->cmd->proc == helloCommand))
+                        !c->authenticated;
-    {
+    if (auth_required || DefaultUser->flags & USER_FLAG_DISABLED) {
+        /* AUTH and HELLO are valid even in non authenticated state. */
+        if (c->cmd->proc != authCommand || c->cmd->proc == helloCommand) {
             flagTransaction(c);
             addReply(c,shared.noautherr);
             return C_OK;
         }
+    }
 
     /* Check if the user can run this command according to the current
      * ACLs. */
@@ -3791,8 +3804,8 @@ void addReplyCommand(client *c, struct redisCommand *cmd) {
     if (!cmd) {
         addReplyNull(c);
     } else {
-        /* We are adding: command name, arg count, flags, first, last, offset */
+        /* We are adding: command name, arg count, flags, first, last, offset, categories */
-        addReplyArrayLen(c, 6);
+        addReplyArrayLen(c, 7);
         addReplyBulkCString(c, cmd->name);
         addReplyLongLong(c, cmd->arity);
 
@@ -3822,6 +3835,8 @@ void addReplyCommand(client *c, struct redisCommand *cmd) {
         addReplyLongLong(c, cmd->firstkey);
         addReplyLongLong(c, cmd->lastkey);
         addReplyLongLong(c, cmd->keystep);
+
+        addReplyCommandCategories(c,cmd);
     }
 }
 
@@ -5034,11 +5049,7 @@ int main(int argc, char **argv) {
         linuxMemoryWarnings();
     #endif
         moduleLoadFromQueue();
-        if (ACLLoadConfiguredUsers() == C_ERR) {
+        ACLLoadUsersAtStartup();
-            serverLog(LL_WARNING,
-                "Critical error while loading ACLs. Exiting.");
-            exit(1);
-        }
         loadDataFromDisk();
         if (server.cluster_enabled) {
             if (verifyClusterConfigWithData() == C_ERR) {

src/server.h

@@ -1270,6 +1270,7 @@ struct redisServer {
     int repl_diskless_sync;         /* Send RDB to slaves sockets directly. */
     int repl_diskless_sync_delay;   /* Delay to start a diskless repl BGSAVE. */
     /* Replication (slave) */
+    char *masteruser;               /* AUTH with this user and masterauth with master */
     char *masterauth;               /* AUTH with this password with master */
     char *masterhost;               /* Hostname of master */
     int masterport;                 /* Port of master */
@@ -1820,6 +1821,8 @@ int ACLAppendUserForLoading(sds *argv, int argc, int *argc_err);
 char *ACLSetUserStringError(void);
 int ACLLoadConfiguredUsers(void);
 sds ACLDescribeUser(user *u);
+void ACLLoadUsersAtStartup(void);
+void addReplyCommandCategories(client *c, struct redisCommand *cmd);
 
 /* Sorted sets data type */
 

src/t_list.c

@@ -630,6 +630,7 @@ int serveClientBlockedOnList(client *receiver, robj *key, robj *dstkey, redisDb
     robj *argv[3];
 
     if (dstkey == NULL) {
+        fastlock_lock(&receiver->lock);
         /* Propagate the [LR]POP operation. */
         argv[0] = (where == LIST_HEAD) ? shared.lpop :
                                           shared.rpop;
@@ -646,7 +647,9 @@ int serveClientBlockedOnList(client *receiver, robj *key, robj *dstkey, redisDb
         /* Notify event. */
         char *event = (where == LIST_HEAD) ? "lpop" : "rpop";
         notifyKeyspaceEvent(NOTIFY_LIST,event,key,receiver->db->id);
+        fastlock_unlock(&receiver->lock);
     } else {
+        fastlock_lock(&receiver->lock);
         /* BRPOPLPUSH */
         robj *dstobj =
             lookupKeyWrite(receiver->db,dstkey);
@@ -673,6 +676,7 @@ int serveClientBlockedOnList(client *receiver, robj *key, robj *dstkey, redisDb
 
             /* Notify event ("lpush" was notified by rpoplpushHandlePush). */
             notifyKeyspaceEvent(NOTIFY_LIST,"rpop",key,receiver->db->id);
+            fastlock_unlock(&receiver->lock);
         } else {
             /* BRPOPLPUSH failed because of wrong
              * destination type. */

src/t_set.c

@@ -207,7 +207,7 @@ sds setTypeNextObject(setTypeIterator *si) {
  * used field with values which are easy to trap if misused. */
 int setTypeRandomElement(robj *setobj, sds *sdsele, int64_t *llele) {
     if (setobj->encoding == OBJ_ENCODING_HT) {
-        dictEntry *de = dictGetRandomKey(setobj->m_ptr);
+        dictEntry *de = dictGetFairRandomKey(setobj->m_ptr);
         *sdsele = dictGetKey(de);
         *llele = -123456789; /* Not needed. Defensive. */
     } else if (setobj->encoding == OBJ_ENCODING_INTSET) {

tests/unit/slowlog.tcl

@@ -82,7 +82,6 @@ start_server {tags {"slowlog"} overrides {slowlog-log-slower-than 1000000}} {
     test {SLOWLOG - can be disabled} {
         r config set slowlog-log-slower-than 1
         r slowlog reset
-        r debug sleep 0.2
         assert_equal [r slowlog len] 1 
         r config set slowlog-log-slower-than -1
         r slowlog reset

utils/srandmember/README.md

@@ -0,0 +1,14 @@
+The utilities in this directory plot the distribution of SRANDMEMBER to
+evaluate how fair it is.
+
+See http://theshfl.com/redis_sets for more information on the topic that lead
+to such investigation fix.
+
+showdist.rb -- shows the distribution of the frequency elements are returned.
+               The x axis is the number of times elements were returned, and
+               the y axis is how many elements were returned with such
+               frequency.
+
+showfreq.rb -- shows the frequency each element was returned.
+               The x axis is the element number.
+               The y axis is the times it was returned.

utils/srandmember/showdist.rb

@@ -0,0 +1,33 @@
+require 'redis'
+
+r = Redis.new
+r.select(9)
+r.del("myset");
+r.sadd("myset",(0..999).to_a)
+freq = {}
+100.times {
+    res = r.pipelined {
+        1000.times {
+            r.srandmember("myset")
+        }
+    }
+    res.each{|ele|
+        freq[ele] = 0 if freq[ele] == nil
+        freq[ele] += 1
+    }
+}
+
+# Convert into frequency distribution
+dist = {}
+freq.each{|item,count|
+    dist[count] = 0 if dist[count] == nil
+    dist[count] += 1
+}
+
+min = dist.keys.min
+max = dist.keys.max
+(min..max).each{|x|
+    count = dist[x]
+    count = 0 if count == nil
+    puts "#{x} -> #{"*"*count}"
+}

utils/srandmember/showfreq.rb

@@ -0,0 +1,23 @@
+require 'redis'
+
+r = Redis.new
+r.select(9)
+r.del("myset");
+r.sadd("myset",(0..999).to_a)
+freq = {}
+500.times {
+    res = r.pipelined {
+        1000.times {
+            r.srandmember("myset")
+        }
+    }
+    res.each{|ele|
+        freq[ele] = 0 if freq[ele] == nil
+        freq[ele] += 1
+    }
+}
+
+# Print the frequency each element was yeld to process it with gnuplot
+freq.each{|item,count|
+    puts "#{item} #{count}"
+}