Redis 5.0 RC4.

Very useful with --stop in order to catch heisenbugs.

redis.conf

@@ -1205,6 +1205,22 @@ client-output-buffer-limit pubsub 32mb 8mb 60
 # 100 only in environments where very low latency is required.
 hz 10
 
+# Normally it is useful to have an HZ value which is proportional to the
+# number of clients connected. This is useful in order, for instance, to
+# avoid too many clients are processed for each background task invocation
+# in order to avoid latency spikes.
+#
+# Since the default HZ value by default is conservatively set to 10, Redis
+# offers, and enables by default, the ability to use an adaptive HZ value
+# which will temporary raise when there are many connected clients.
+#
+# When dynamic HZ is enabled, the actual configured HZ will be used as
+# as a baseline, but multiples of the configured HZ value will be actually
+# used as needed once more clients are connected. In this way an idle
+# instance will use very little CPU time while a busy instance will be
+# more responsive.
+dynamic-hz yes
+
 # When a child rewrites the AOF file, if the following option is enabled
 # the file will be fsync-ed every 32 MB of data generated. This is useful
 # in order to commit the file to the disk more incrementally and avoid

runtest

@@ -11,4 +11,4 @@ then
     echo "You need tcl 8.5 or newer in order to run the Redis test"
     exit 1
 fi
-$TCLSH tests/test_helper.tcl $*
+$TCLSH tests/test_helper.tcl "${@}"

src/cluster.c

@@ -3321,14 +3321,18 @@ void clusterCron(void) {
         int changed = 0;
 
         if (prev_ip == NULL && curr_ip != NULL) changed = 1;
-        if (prev_ip != NULL && curr_ip == NULL) changed = 1;
-        if (prev_ip && curr_ip && strcmp(prev_ip,curr_ip)) changed = 1;
+        else if (prev_ip != NULL && curr_ip == NULL) changed = 1;
+        else if (prev_ip && curr_ip && strcmp(prev_ip,curr_ip)) changed = 1;
 
         if (changed) {
+            if (prev_ip) zfree(prev_ip);
             prev_ip = curr_ip;
-            if (prev_ip) prev_ip = zstrdup(prev_ip);
 
             if (curr_ip) {
+                /* We always take a copy of the previous IP address, by
+                 * duplicating the string. This way later we can check if
+                 * the address really changed. */
+                prev_ip = zstrdup(prev_ip);
                 strncpy(myself->ip,server.cluster_announce_ip,NET_IP_STR_LEN);
                 myself->ip[NET_IP_STR_LEN-1] = '\0';
             } else {
@@ -5146,6 +5150,11 @@ try_again:
         serverAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,dbid));
     }
 
+    int expired = 0; /* Number of keys that we'll find already expired.
+                        Note that serializing large keys may take some time
+                        so certain keys that were found non expired by the
+                        lookupKey() function, may be expired later. */
+
     /* Create RESTORE payload and generate the protocol to call the command. */
     for (j = 0; j < num_keys; j++) {
         long long ttl = 0;
@@ -5153,6 +5162,10 @@ try_again:
 
         if (expireat != -1) {
             ttl = expireat-mstime();
+            if (ttl < 0) {
+                expired++;
+                continue;
+            }
             if (ttl < 1) ttl = 1;
         }
         serverAssertWithInfo(c,NULL,
@@ -5217,9 +5230,13 @@ try_again:
     int socket_error = 0;
     int del_idx = 1; /* Index of the key argument for the replicated DEL op. */
 
+    /* Allocate the new argument vector that will replace the current command,
+     * to propagate the MIGRATE as a DEL command (if no COPY option was given).
+     * We allocate num_keys+1 because the additional argument is for "DEL"
+     * command name itself. */
     if (!copy) newargv = zmalloc(sizeof(robj*)*(num_keys+1));
 
-    for (j = 0; j < num_keys; j++) {
+    for (j = 0; j < num_keys-expired; j++) {
         if (syncReadLine(cs->fd, buf2, sizeof(buf2), timeout) <= 0) {
             socket_error = 1;
             break;

src/config.c

@@ -440,10 +440,14 @@ void loadServerConfigFromString(char *config) {
             if ((server.daemonize = yesnotoi(argv[1])) == -1) {
                 err = "argument must be 'yes' or 'no'"; goto loaderr;
             }
+        } else if (!strcasecmp(argv[0],"dynamic-hz") && argc == 2) {
+            if ((server.dynamic_hz = yesnotoi(argv[1])) == -1) {
+                err = "argument must be 'yes' or 'no'"; goto loaderr;
+            }
         } else if (!strcasecmp(argv[0],"hz") && argc == 2) {
-            server.hz = atoi(argv[1]);
-            if (server.hz < CONFIG_MIN_HZ) server.hz = CONFIG_MIN_HZ;
-            if (server.hz > CONFIG_MAX_HZ) server.hz = CONFIG_MAX_HZ;
+            server.config_hz = atoi(argv[1]);
+            if (server.config_hz < CONFIG_MIN_HZ) server.config_hz = CONFIG_MIN_HZ;
+            if (server.config_hz > CONFIG_MAX_HZ) server.config_hz = CONFIG_MAX_HZ;
         } else if (!strcasecmp(argv[0],"appendonly") && argc == 2) {
             int yes;
 
@@ -1072,6 +1076,8 @@ void configSetCommand(client *c) {
       "slave-lazy-flush",server.repl_slave_lazy_flush) {
     } config_set_bool_field(
       "no-appendfsync-on-rewrite",server.aof_no_fsync_on_rewrite) {
+    } config_set_bool_field(
+      "dynamic-hz",server.dynamic_hz) {
 
     /* Numerical fields.
      * config_set_numerical_field(name,var,min,max) */
@@ -1158,11 +1164,11 @@ void configSetCommand(client *c) {
     } config_set_numerical_field(
       "cluster-slave-validity-factor",server.cluster_slave_validity_factor,0,INT_MAX) {
     } config_set_numerical_field(
-      "hz",server.hz,0,INT_MAX) {
+      "hz",server.config_hz,0,INT_MAX) {
         /* Hz is more an hint from the user, so we accept values out of range
          * but cap them to reasonable values. */
-        if (server.hz < CONFIG_MIN_HZ) server.hz = CONFIG_MIN_HZ;
-        if (server.hz > CONFIG_MAX_HZ) server.hz = CONFIG_MAX_HZ;
+        if (server.config_hz < CONFIG_MIN_HZ) server.config_hz = CONFIG_MIN_HZ;
+        if (server.config_hz > CONFIG_MAX_HZ) server.config_hz = CONFIG_MAX_HZ;
     } config_set_numerical_field(
       "watchdog-period",ll,0,INT_MAX) {
         if (ll)
@@ -1329,7 +1335,7 @@ void configGetCommand(client *c) {
     config_get_numerical_field("slave-announce-port",server.slave_announce_port);
     config_get_numerical_field("min-slaves-to-write",server.repl_min_slaves_to_write);
     config_get_numerical_field("min-slaves-max-lag",server.repl_min_slaves_max_lag);
-    config_get_numerical_field("hz",server.hz);
+    config_get_numerical_field("hz",server.config_hz);
     config_get_numerical_field("cluster-node-timeout",server.cluster_node_timeout);
     config_get_numerical_field("cluster-migration-barrier",server.cluster_migration_barrier);
     config_get_numerical_field("cluster-slave-validity-factor",server.cluster_slave_validity_factor);
@@ -1375,6 +1381,8 @@ void configGetCommand(client *c) {
             server.lazyfree_lazy_server_del);
     config_get_bool_field("slave-lazy-flush",
             server.repl_slave_lazy_flush);
+    config_get_bool_field("dynamic-hz",
+            server.dynamic_hz);
 
     /* Enum values */
     config_get_enum_field("maxmemory-policy",
@@ -2098,7 +2106,7 @@ int rewriteConfig(char *path) {
     rewriteConfigYesNoOption(state,"activedefrag",server.active_defrag_enabled,CONFIG_DEFAULT_ACTIVE_DEFRAG);
     rewriteConfigYesNoOption(state,"protected-mode",server.protected_mode,CONFIG_DEFAULT_PROTECTED_MODE);
     rewriteConfigClientoutputbufferlimitOption(state);
-    rewriteConfigNumericalOption(state,"hz",server.hz,CONFIG_DEFAULT_HZ);
+    rewriteConfigNumericalOption(state,"hz",server.config_hz,CONFIG_DEFAULT_HZ);
     rewriteConfigYesNoOption(state,"aof-rewrite-incremental-fsync",server.aof_rewrite_incremental_fsync,CONFIG_DEFAULT_AOF_REWRITE_INCREMENTAL_FSYNC);
     rewriteConfigYesNoOption(state,"rdb-save-incremental-fsync",server.rdb_save_incremental_fsync,CONFIG_DEFAULT_RDB_SAVE_INCREMENTAL_FSYNC);
     rewriteConfigYesNoOption(state,"aof-load-truncated",server.aof_load_truncated,CONFIG_DEFAULT_AOF_LOAD_TRUNCATED);
@@ -2108,6 +2116,7 @@ int rewriteConfig(char *path) {
     rewriteConfigYesNoOption(state,"lazyfree-lazy-expire",server.lazyfree_lazy_expire,CONFIG_DEFAULT_LAZYFREE_LAZY_EXPIRE);
     rewriteConfigYesNoOption(state,"lazyfree-lazy-server-del",server.lazyfree_lazy_server_del,CONFIG_DEFAULT_LAZYFREE_LAZY_SERVER_DEL);
     rewriteConfigYesNoOption(state,"slave-lazy-flush",server.repl_slave_lazy_flush,CONFIG_DEFAULT_SLAVE_LAZY_FLUSH);
+    rewriteConfigYesNoOption(state,"dynamic-hz",server.dynamic_hz,CONFIG_DEFAULT_DYNAMIC_HZ);
 
     /* Rewrite Sentinel config if in Sentinel mode. */
     if (server.sentinel_mode) rewriteConfigSentinelOption(state);

src/db.c

@@ -184,14 +184,19 @@ void dbOverwrite(redisDb *db, robj *key, robj *val) {
     dictEntry *de = dictFind(db->dict,key->ptr);
 
     serverAssertWithInfo(NULL,key,de != NULL);
+    dictEntry auxentry = *de;
+    robj *old = dictGetVal(de);
     if (server.maxmemory_policy & MAXMEMORY_FLAG_LFU) {
-        robj *old = dictGetVal(de);
-        int saved_lru = old->lru;
-        dictReplace(db->dict, key->ptr, val);
-        val->lru = saved_lru;
-    } else {
-        dictReplace(db->dict, key->ptr, val);
+        val->lru = old->lru;
     }
+    dictSetVal(db->dict, de, val);
+
+    if (server.lazyfree_lazy_server_del) {
+        freeObjAsync(old);
+        dictSetVal(db->dict, &auxentry, NULL);
+    }
+
+    dictFreeVal(db->dict, &auxentry);
 }
 
 /* High level Set operation. This function can be used in order to set
@@ -329,7 +334,7 @@ robj *dbUnshareStringValue(redisDb *db, robj *key, robj *o) {
  * database(s). Otherwise -1 is returned in the specific case the
  * DB number is out of range, and errno is set to EINVAL. */
 long long emptyDb(int dbnum, int flags, void(callback)(void*)) {
-    int j, async = (flags & EMPTYDB_ASYNC);
+    int async = (flags & EMPTYDB_ASYNC);
     long long removed = 0;
 
     if (dbnum < -1 || dbnum >= server.dbnum) {
@@ -337,8 +342,15 @@ long long emptyDb(int dbnum, int flags, void(callback)(void*)) {
         return -1;
     }
 
-    for (j = 0; j < server.dbnum; j++) {
-        if (dbnum != -1 && dbnum != j) continue;
+    int startdb, enddb;
+    if (dbnum == -1) {
+        startdb = 0;
+        enddb = server.dbnum-1;
+    } else {
+        startdb = enddb = dbnum;
+    }
+
+    for (int j = startdb; j <= enddb; j++) {
         removed += dictSize(server.db[j].dict);
         if (async) {
             emptyDbAsync(&server.db[j]);

src/debug.c

@@ -290,6 +290,7 @@ void debugCommand(client *c) {
 "CRASH-AND-RECOVER <milliseconds> -- Hard crash and restart after <milliseconds> delay.",
 "DIGEST -- Output a hex signature representing the current DB content.",
 "ERROR <string> -- Return a Redis protocol error with <string> as message. Useful for clients unit tests to simulate Redis errors.",
+"LOG <message> -- write message to the server log.",
 "HTSTATS <dbid> -- Return hash table statistics of the specified Redis database.",
 "HTSTATS-KEY <key> -- Like htstats but for the hash table stored as key's value.",
 "LOADAOF -- Flush the AOF buffers on disk and reload the AOF in memory.",
@@ -333,6 +334,9 @@ NULL
     } else if (!strcasecmp(c->argv[1]->ptr,"assert")) {
         if (c->argc >= 3) c->argv[2] = tryObjectEncoding(c->argv[2]);
         serverAssertWithInfo(c,c->argv[0],1 == 2);
+    } else if (!strcasecmp(c->argv[1]->ptr,"log") && c->argc == 3) {
+        serverLog(LL_WARNING, "DEBUG LOG: %s", (char*)c->argv[2]->ptr);
+        addReply(c,shared.ok);
     } else if (!strcasecmp(c->argv[1]->ptr,"reload")) {
         rdbSaveInfo rsi, *rsiptr;
         rsiptr = rdbPopulateSaveInfo(&rsi);

src/lazyfree.c

@@ -90,6 +90,17 @@ int dbAsyncDelete(redisDb *db, robj *key) {
     }
 }
 
+/* Free an object, if the object is huge enough, free it in async way. */
+void freeObjAsync(robj *o) {
+    size_t free_effort = lazyfreeGetFreeEffort(o);
+    if (free_effort > LAZYFREE_THRESHOLD && o->refcount == 1) {
+        atomicIncr(lazyfree_objects,1);
+        bioCreateBackgroundJob(BIO_LAZY_FREE,o,NULL,NULL);
+    } else {
+        decrRefCount(o);
+    }
+}
+
 /* Empty a Redis DB asynchronously. What the function does actually is to
  * create a new empty set of hash tables and scheduling the old ones for
  * lazy freeing. */

src/module.c

@@ -548,7 +548,7 @@ void RM_KeyAtPos(RedisModuleCtx *ctx, int pos) {
     ctx->keys_pos[ctx->keys_count++] = pos;
 }
 
-/* Helper for RM_CreateCommand(). Truns a string representing command
+/* Helper for RM_CreateCommand(). Turns a string representing command
  * flags into the command flags used by the Redis core.
  *
  * It returns the set of flags, or -1 if unknown flags are found. */
@@ -595,7 +595,7 @@ int commandFlagsFromString(char *s) {
  * And is supposed to always return REDISMODULE_OK.
  *
  * The set of flags 'strflags' specify the behavior of the command, and should
- * be passed as a C string compoesd of space separated words, like for
+ * be passed as a C string composed of space separated words, like for
  * example "write deny-oom". The set of flags are:
  *
  * * **"write"**:     The command may modify the data set (it may also read
@@ -616,7 +616,7 @@ int commandFlagsFromString(char *s) {
  * * **"allow-stale"**: The command is allowed to run on slaves that don't
  *                      serve stale data. Don't use if you don't know what
  *                      this means.
- * * **"no-monitor"**: Don't propoagate the command on monitor. Use this if
+ * * **"no-monitor"**: Don't propagate the command on monitor. Use this if
  *                     the command has sensible data among the arguments.
  * * **"fast"**:      The command time complexity is not greater
  *                    than O(log(N)) where N is the size of the collection or
@@ -956,9 +956,9 @@ RedisModuleString *moduleAssertUnsharedString(RedisModuleString *str) {
     return str;
 }
 
-/* Append the specified buffere to the string 'str'. The string must be a
+/* Append the specified buffer to the string 'str'. The string must be a
  * string created by the user that is referenced only a single time, otherwise
- * REDISMODULE_ERR is returend and the operation is not performed. */
+ * REDISMODULE_ERR is returned and the operation is not performed. */
 int RM_StringAppendBuffer(RedisModuleCtx *ctx, RedisModuleString *str, const char *buf, size_t len) {
     UNUSED(ctx);
     str = moduleAssertUnsharedString(str);
@@ -1118,7 +1118,7 @@ int RM_ReplyWithArray(RedisModuleCtx *ctx, long len) {
  *
  * Note that in the above example there is no reason to postpone the array
  * length, since we produce a fixed number of elements, but in the practice
- * the code may use an interator or other ways of creating the output so
+ * the code may use an iterator or other ways of creating the output so
  * that is not easy to calculate in advance the number of elements.
  */
 void RM_ReplySetArrayLength(RedisModuleCtx *ctx, long len) {
@@ -1410,7 +1410,7 @@ int RM_SelectDb(RedisModuleCtx *ctx, int newid) {
  * to call other APIs with the key handle as argument to perform
  * operations on the key.
  *
- * The return value is the handle repesenting the key, that must be
+ * The return value is the handle representing the key, that must be
  * closed with RM_CloseKey().
  *
  * If the key does not exist and WRITE mode is requested, the handle
@@ -1664,7 +1664,7 @@ int RM_StringTruncate(RedisModuleKey *key, size_t newlen) {
  * Key API for List type
  * -------------------------------------------------------------------------- */
 
-/* Push an element into a list, on head or tail depending on 'where' argumnet.
+/* Push an element into a list, on head or tail depending on 'where' argument.
  * If the key pointer is about an empty key opened for writing, the key
  * is created. On error (key opened for read-only operations or of the wrong
  * type) REDISMODULE_ERR is returned, otherwise REDISMODULE_OK is returned. */
@@ -1769,7 +1769,7 @@ int RM_ZsetAdd(RedisModuleKey *key, double score, RedisModuleString *ele, int *f
  * The input and output flags, and the return value, have the same exact
  * meaning, with the only difference that this function will return
  * REDISMODULE_ERR even when 'score' is a valid double number, but adding it
- * to the existing score resuts into a NaN (not a number) condition.
+ * to the existing score results into a NaN (not a number) condition.
  *
  * This function has an additional field 'newscore', if not NULL is filled
  * with the new score of the element after the increment, if no error
@@ -3793,11 +3793,11 @@ void moduleReleaseGIL(void) {
  * -------------------------------------------------------------------------- */
 
 /* Subscribe to keyspace notifications. This is a low-level version of the
- * keyspace-notifications API. A module cand register callbacks to be notified
+ * keyspace-notifications API. A module can register callbacks to be notified
  * when keyspce events occur.
  *
  * Notification events are filtered by their type (string events, set events,
- * etc), and the subsriber callback receives only events that match a specific
+ * etc), and the subscriber callback receives only events that match a specific
  * mask of event types.
  *
  * When subscribing to notifications with RedisModule_SubscribeToKeyspaceEvents 
@@ -3832,7 +3832,7 @@ void moduleReleaseGIL(void) {
  * used to send anything to the client, and has the db number where the event
  * occurred as its selected db number.
  *
- * Notice that it is not necessary to enable norifications in redis.conf for
+ * Notice that it is not necessary to enable notifications in redis.conf for
  * module notifications to work.
  *
  * Warning: the notification callbacks are performed in a synchronous manner,
@@ -3876,7 +3876,7 @@ void moduleNotifyKeyspaceEvent(int type, const char *event, robj *key, int dbid)
             ctx.client = moduleKeyspaceSubscribersClient;
             selectDb(ctx.client, dbid);
 
-            /* mark the handler as activer to avoid reentrant loops.
+            /* mark the handler as active to avoid reentrant loops.
              * If the subscriber performs an action triggering itself,
              * it will not be notified about it. */
             sub->active = 1;

src/object.c

@@ -1017,6 +1017,18 @@ struct redisMemOverhead *getMemoryOverheadData(void) {
     mh->aof_buffer = mem;
     mem_total+=mem;
 
+    mem = 0;
+    mem += dictSize(server.lua_scripts) * sizeof(dictEntry) +
+        dictSlots(server.lua_scripts) * sizeof(dictEntry*);
+    mem += dictSize(server.repl_scriptcache_dict) * sizeof(dictEntry) +
+        dictSlots(server.repl_scriptcache_dict) * sizeof(dictEntry*);
+    if (listLength(server.repl_scriptcache_fifo) > 0) {
+        mem += listLength(server.repl_scriptcache_fifo) * (sizeof(listNode) + 
+            sdsZmallocSize(listNodeValue(listFirst(server.repl_scriptcache_fifo))));
+    }
+    mh->lua_caches = mem;
+    mem_total+=mem;
+
     for (j = 0; j < server.dbnum; j++) {
         redisDb *db = server.db+j;
         long long keyscount = dictSize(db->dict);
@@ -1074,6 +1086,7 @@ sds getMemoryDoctorReport(void) {
     int high_alloc_rss = 0; /* High rss overhead. */
     int big_slave_buf = 0;  /* Slave buffers are too big. */
     int big_client_buf = 0; /* Client buffers are too big. */
+    int many_scripts = 0;   /* Script cache has too many scripts. */
     int num_reports = 0;
     struct redisMemOverhead *mh = getMemoryOverheadData();
 
@@ -1124,6 +1137,12 @@ sds getMemoryDoctorReport(void) {
             big_slave_buf = 1;
             num_reports++;
         }
+
+        /* Too many scripts are cached? */
+        if (dictSize(server.lua_scripts) > 1000) {
+            many_scripts = 1;
+            num_reports++;
+        }
     }
 
     sds s;
@@ -1153,7 +1172,7 @@ sds getMemoryDoctorReport(void) {
             s = sdscatprintf(s," * High allocator RSS overhead: This instance has an RSS memory overhead is greater than 1.1 (this means that the Resident Set Size of the allocator is much larger than the sum what the allocator actually holds). This problem is usually due to a large peak memory (check if there is a peak memory entry above in the report), you can try the MEMORY PURGE command to reclaim it.\n\n");
         }
         if (high_proc_rss) {
-            s = sdscatprintf(s," * High process RSS overhead: This instance has non-allocator RSS memory overhead is greater than 1.1 (this means that the Resident Set Size of the Redis process is much larger than the RSS the allocator holds). This problem may be due to LUA scripts or Modules.\n\n");
+            s = sdscatprintf(s," * High process RSS overhead: This instance has non-allocator RSS memory overhead is greater than 1.1 (this means that the Resident Set Size of the Redis process is much larger than the RSS the allocator holds). This problem may be due to Lua scripts or Modules.\n\n");
         }
         if (big_slave_buf) {
             s = sdscat(s," * Big slave buffers: The slave output buffers in this instance are greater than 10MB for each slave (on average). This likely means that there is some slave instance that is struggling receiving data, either because it is too slow or because of networking issues. As a result, data piles on the master output buffers. Please try to identify what slave is not receiving data correctly and why. You can use the INFO output in order to check the slaves delays and the CLIENT LIST command to check the output buffers of each slave.\n\n");
@@ -1161,6 +1180,9 @@ sds getMemoryDoctorReport(void) {
         if (big_client_buf) {
             s = sdscat(s," * Big client buffers: The clients output buffers in this instance are greater than 200K per client (on average). This may result from different causes, like Pub/Sub clients subscribed to channels bot not receiving data fast enough, so that data piles on the Redis instance output buffer, or clients sending commands with large replies or very large sequences of commands in the same pipeline. Please use the CLIENT LIST command in order to investigate the issue if it causes problems in your instance, or to understand better why certain clients are using a big amount of memory.\n\n");
         }
+        if (many_scripts) {
+            s = sdscat(s," * Many scripts: There seem to be many cached scripts in this instance (more than 1000). This may be because scripts are generated and `EVAL`ed, instead of being parameterized (with KEYS and ARGV), `SCRIPT LOAD`ed and `EVALSHA`ed. Unless `SCRIPT FLUSH` is called periodically, the scripts' caches may end up consuming most of your memory.\n\n");
+        }
         s = sdscat(s,"I'm here to keep you safe, Sam. I want to help you.\n");
     }
     freeMemoryOverheadData(mh);
@@ -1293,7 +1315,7 @@ void memoryCommand(client *c) {
     } else if (!strcasecmp(c->argv[1]->ptr,"stats") && c->argc == 2) {
         struct redisMemOverhead *mh = getMemoryOverheadData();
 
-        addReplyMultiBulkLen(c,(24+mh->num_dbs)*2);
+        addReplyMultiBulkLen(c,(25+mh->num_dbs)*2);
 
         addReplyBulkCString(c,"peak.allocated");
         addReplyLongLong(c,mh->peak_allocated);
@@ -1316,6 +1338,9 @@ void memoryCommand(client *c) {
         addReplyBulkCString(c,"aof.buffer");
         addReplyLongLong(c,mh->aof_buffer);
 
+        addReplyBulkCString(c,"lua.caches");
+        addReplyLongLong(c,mh->lua_caches);
+
         for (size_t j = 0; j < mh->num_dbs; j++) {
             char dbname[32];
             snprintf(dbname,sizeof(dbname),"db.%zd",mh->db[j].dbid);

src/scripting.c

@@ -483,6 +483,7 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) {
      * command marked as non-deterministic was already called in the context
      * of this script. */
     if (cmd->flags & CMD_WRITE) {
+        int deny_write_type = writeCommandsDeniedByDiskError();
         if (server.lua_random_dirty && !server.lua_replicate_commands) {
             luaPushError(lua,
                 "Write commands not allowed after non deterministic commands. Call redis.replicate_commands() at the start of your script in order to switch to single commands replication mode.");
@@ -493,11 +494,16 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) {
         {
             luaPushError(lua, shared.roslaveerr->ptr);
             goto cleanup;
-        } else if (server.stop_writes_on_bgsave_err &&
-                   server.saveparamslen > 0 &&
-                   server.lastbgsave_status == C_ERR)
-        {
-            luaPushError(lua, shared.bgsaveerr->ptr);
+        } else if (deny_write_type != DISK_ERROR_TYPE_NONE) {
+            if (deny_write_type == DISK_ERROR_TYPE_RDB) {
+                luaPushError(lua, shared.bgsaveerr->ptr);
+            } else {
+                sds aof_write_err = sdscatfmt(sdsempty(),
+                    "-MISCONF Errors writing to the AOF file: %s\r\n",
+                    strerror(server.aof_last_write_errno));
+                luaPushError(lua, aof_write_err);
+                sdsfree(aof_write_err);
+            }
             goto cleanup;
         }
     }
@@ -919,6 +925,7 @@ void scriptingInit(int setup) {
      * This is useful for replication, as we need to replicate EVALSHA
      * as EVAL, so we need to remember the associated script. */
     server.lua_scripts = dictCreate(&shaScriptObjectDictType,NULL);
+    server.lua_scripts_mem = 0;
 
     /* Register the redis commands table and fields */
     lua_newtable(lua);
@@ -1073,6 +1080,7 @@ void scriptingInit(int setup) {
  * This function is used in order to reset the scripting environment. */
 void scriptingRelease(void) {
     dictRelease(server.lua_scripts);
+    server.lua_scripts_mem = 0;
     lua_close(server.lua);
 }
 
@@ -1207,6 +1215,7 @@ sds luaCreateFunction(client *c, lua_State *lua, robj *body) {
      * EVALSHA commands as EVAL using the original script. */
     int retval = dictAdd(server.lua_scripts,sha,body);
     serverAssertWithInfo(c ? c : server.lua_client,NULL,retval == DICT_OK);
+    server.lua_scripts_mem += sdsZmallocSize(sha) + sdsZmallocSize(body->ptr);
     incrRefCount(body);
     return sha;
 }

src/server.c

@@ -357,7 +357,7 @@ void serverLogRaw(int level, const char *msg) {
         gettimeofday(&tv,NULL);
         struct tm tm;
         nolocks_localtime(&tm,tv.tv_sec,server.timezone,server.daylight_active);
-        off = strftime(buf,sizeof(buf),"%d %b %H:%M:%S.",&tm);
+        off = strftime(buf,sizeof(buf),"%d %b %Y %H:%M:%S.",&tm);
         snprintf(buf+off,sizeof(buf)-off,"%03d",(int)tv.tv_usec/1000);
         if (server.sentinel_mode) {
             role_char = 'X'; /* Sentinel. */
@@ -1096,6 +1096,21 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
     /* Update the time cache. */
     updateCachedTime();
 
+    server.hz = server.config_hz;
+    /* Adapt the server.hz value to the number of configured clients. If we have
+     * many clients, we want to call serverCron() with an higher frequency. */
+    if (server.dynamic_hz) {
+        while (listLength(server.clients) / server.hz >
+               MAX_CLIENTS_PER_CLOCK_TICK)
+        {
+            server.hz *= 2;
+            if (server.hz > CONFIG_MAX_HZ) {
+                server.hz = CONFIG_MAX_HZ;
+                break;
+            }
+        }
+    }
+
     run_with_period(100) {
         trackInstantaneousMetric(STATS_METRIC_COMMAND,server.stat_numcommands);
         trackInstantaneousMetric(STATS_METRIC_NET_INPUT,
@@ -1512,7 +1527,8 @@ void initServerConfig(void) {
     server.timezone = timezone; /* Initialized by tzset(). */
     server.configfile = NULL;
     server.executable = NULL;
-    server.hz = CONFIG_DEFAULT_HZ;
+    server.config_hz = CONFIG_DEFAULT_HZ;
+    server.dynamic_hz = CONFIG_DEFAULT_DYNAMIC_HZ;
     server.arch_bits = (sizeof(long) == 8) ? 64 : 32;
     server.port = CONFIG_DEFAULT_SERVER_PORT;
     server.tcp_backlog = CONFIG_DEFAULT_TCP_BACKLOG;
@@ -1991,6 +2007,7 @@ void initServer(void) {
             server.syslog_facility);
     }
 
+    server.hz = server.config_hz;
     server.pid = getpid();
     server.current_client = NULL;
     server.clients = listCreate();
@@ -2592,17 +2609,14 @@ int processCommand(client *c) {
 
     /* Don't accept write commands if there are problems persisting on disk
      * and if this is a master instance. */
-    if (((server.stop_writes_on_bgsave_err &&
-          server.saveparamslen > 0 &&
-          server.lastbgsave_status == C_ERR) ||
-          (server.aof_state != AOF_OFF &&
-           server.aof_last_write_status == C_ERR)) &&
+    int deny_write_type = writeCommandsDeniedByDiskError();
+    if (deny_write_type != DISK_ERROR_TYPE_NONE &&
         server.masterhost == NULL &&
         (c->cmd->flags & CMD_WRITE ||
          c->cmd->proc == pingCommand))
     {
         flagTransaction(c);
-        if (server.aof_last_write_status == C_OK)
+        if (deny_write_type == DISK_ERROR_TYPE_RDB)
             addReply(c, shared.bgsaveerr);
         else
             addReplySds(c,
@@ -2788,6 +2802,32 @@ int prepareForShutdown(int flags) {
 
 /*================================== Commands =============================== */
 
+/* Sometimes Redis cannot accept write commands because there is a perstence
+ * error with the RDB or AOF file, and Redis is configured in order to stop
+ * accepting writes in such situation. This function returns if such a
+ * condition is active, and the type of the condition.
+ *
+ * Function return values:
+ *
+ * DISK_ERROR_TYPE_NONE:    No problems, we can accept writes.
+ * DISK_ERROR_TYPE_AOF:     Don't accept writes: AOF errors.
+ * DISK_ERROR_TYPE_RDB:     Don't accept writes: RDB errors.
+ */
+int writeCommandsDeniedByDiskError(void) {
+    if (server.stop_writes_on_bgsave_err &&
+        server.saveparamslen > 0 &&
+        server.lastbgsave_status == C_ERR)
+    {
+        return DISK_ERROR_TYPE_RDB;
+    } else if (server.aof_state != AOF_OFF &&
+               server.aof_last_write_status == C_ERR)
+    {
+        return DISK_ERROR_TYPE_AOF;
+    } else {
+        return DISK_ERROR_TYPE_NONE;
+    }
+}
+
 /* 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
@@ -3074,6 +3114,7 @@ sds genRedisInfoString(char *section) {
             "uptime_in_seconds:%jd\r\n"
             "uptime_in_days:%jd\r\n"
             "hz:%d\r\n"
+            "configured_hz:%d\r\n"
             "lru_clock:%ld\r\n"
             "executable:%s\r\n"
             "config_file:%s\r\n",
@@ -3097,6 +3138,7 @@ sds genRedisInfoString(char *section) {
             (intmax_t)uptime,
             (intmax_t)(uptime/(3600*24)),
             server.hz,
+            server.config_hz,
             (unsigned long) lruclock,
             server.executable ? server.executable : "",
             server.configfile ? server.configfile : "");
@@ -3124,6 +3166,7 @@ sds genRedisInfoString(char *section) {
         char peak_hmem[64];
         char total_system_hmem[64];
         char used_memory_lua_hmem[64];
+        char used_memory_scripts_hmem[64];
         char used_memory_rss_hmem[64];
         char maxmemory_hmem[64];
         size_t zmalloc_used = zmalloc_used_memory();
@@ -3143,6 +3186,7 @@ sds genRedisInfoString(char *section) {
         bytesToHuman(peak_hmem,server.stat_peak_memory);
         bytesToHuman(total_system_hmem,total_system_mem);
         bytesToHuman(used_memory_lua_hmem,memory_lua);
+        bytesToHuman(used_memory_scripts_hmem,server.lua_scripts_mem);
         bytesToHuman(used_memory_rss_hmem,server.cron_malloc_stats.process_rss);
         bytesToHuman(maxmemory_hmem,server.maxmemory);
 
@@ -3167,6 +3211,9 @@ sds genRedisInfoString(char *section) {
             "total_system_memory_human:%s\r\n"
             "used_memory_lua:%lld\r\n"
             "used_memory_lua_human:%s\r\n"
+            "used_memory_scripts:%lld\r\n"
+            "used_memory_scripts_human:%s\r\n"
+            "number_of_cached_scripts:%lu\r\n"
             "maxmemory:%lld\r\n"
             "maxmemory_human:%s\r\n"
             "maxmemory_policy:%s\r\n"
@@ -3204,6 +3251,9 @@ sds genRedisInfoString(char *section) {
             total_system_hmem,
             memory_lua,
             used_memory_lua_hmem,
+            server.lua_scripts_mem,
+            used_memory_scripts_hmem,
+            dictSize(server.lua_scripts),
             server.maxmemory,
             maxmemory_hmem,
             evict_policy,
@@ -3509,14 +3559,14 @@ sds genRedisInfoString(char *section) {
         if (sections++) info = sdscat(info,"\r\n");
         info = sdscatprintf(info,
         "# CPU\r\n"
-        "used_cpu_sys:%.2f\r\n"
-        "used_cpu_user:%.2f\r\n"
-        "used_cpu_sys_children:%.2f\r\n"
-        "used_cpu_user_children:%.2f\r\n",
-        (float)self_ru.ru_stime.tv_sec+(float)self_ru.ru_stime.tv_usec/1000000,
-        (float)self_ru.ru_utime.tv_sec+(float)self_ru.ru_utime.tv_usec/1000000,
-        (float)c_ru.ru_stime.tv_sec+(float)c_ru.ru_stime.tv_usec/1000000,
-        (float)c_ru.ru_utime.tv_sec+(float)c_ru.ru_utime.tv_usec/1000000);
+        "used_cpu_sys:%ld.%06ld\r\n"
+        "used_cpu_user:%ld.%06ld\r\n"
+        "used_cpu_sys_children:%ld.%06ld\r\n"
+        "used_cpu_user_children:%ld.%06ld\r\n",
+        (long)self_ru.ru_stime.tv_sec, (long)self_ru.ru_stime.tv_usec,
+        (long)self_ru.ru_utime.tv_sec, (long)self_ru.ru_utime.tv_usec,
+        (long)c_ru.ru_stime.tv_sec, (long)c_ru.ru_stime.tv_usec,
+        (long)c_ru.ru_utime.tv_sec, (long)c_ru.ru_utime.tv_usec);
     }
 
     /* Command statistics */

src/server.h

@@ -78,12 +78,14 @@ typedef long long mstime_t; /* millisecond time type. */
 #define C_ERR                   -1
 
 /* Static server configuration */
-#define CONFIG_DEFAULT_HZ        10      /* Time interrupt calls/sec. */
+#define CONFIG_DEFAULT_DYNAMIC_HZ 1             /* Adapt hz to # of clients.*/
+#define CONFIG_DEFAULT_HZ        10             /* Time interrupt calls/sec. */
 #define CONFIG_MIN_HZ            1
 #define CONFIG_MAX_HZ            500
-#define CONFIG_DEFAULT_SERVER_PORT        6379    /* TCP port */
-#define CONFIG_DEFAULT_TCP_BACKLOG       511     /* TCP listen backlog */
-#define CONFIG_DEFAULT_CLIENT_TIMEOUT       0       /* default client timeout: infinite */
+#define MAX_CLIENTS_PER_CLOCK_TICK 200          /* HZ is adapted based on that. */
+#define CONFIG_DEFAULT_SERVER_PORT        6379  /* TCP port. */
+#define CONFIG_DEFAULT_TCP_BACKLOG       511    /* TCP listen backlog. */
+#define CONFIG_DEFAULT_CLIENT_TIMEOUT       0   /* Default client timeout: infinite */
 #define CONFIG_DEFAULT_DBNUM     16
 #define CONFIG_MAX_LINE    1024
 #define CRON_DBS_PER_CALL 16
@@ -91,7 +93,7 @@ typedef long long mstime_t; /* millisecond time type. */
 #define PROTO_SHARED_SELECT_CMDS 10
 #define OBJ_SHARED_INTEGERS 10000
 #define OBJ_SHARED_BULKHDR_LEN 32
-#define LOG_MAX_LEN    1024 /* Default maximum length of syslog messages */
+#define LOG_MAX_LEN    1024 /* Default maximum length of syslog messages.*/
 #define AOF_REWRITE_PERC  100
 #define AOF_REWRITE_MIN_SIZE (64*1024*1024)
 #define AOF_REWRITE_ITEMS_PER_CMD 64
@@ -851,6 +853,7 @@ struct redisMemOverhead {
     size_t clients_slaves;
     size_t clients_normal;
     size_t aof_buffer;
+    size_t lua_caches;
     size_t overhead_total;
     size_t dataset;
     size_t total_keys;
@@ -923,6 +926,10 @@ struct redisServer {
     char *configfile;           /* Absolute config file path, or NULL */
     char *executable;           /* Absolute executable file path. */
     char **exec_argv;           /* Executable argv vector (copy). */
+    int dynamic_hz;             /* Change hz value depending on # of clients. */
+    int config_hz;              /* Configured HZ value. May be different than
+                                   the actual 'hz' field value if dynamic-hz
+                                   is enabled. */
     int hz;                     /* serverCron() calls frequency in hertz */
     redisDb *db;
     dict *commands;             /* Command table */
@@ -1227,6 +1234,7 @@ struct redisServer {
     client *lua_client;   /* The "fake client" to query Redis from Lua */
     client *lua_caller;   /* The client running EVAL right now, or NULL */
     dict *lua_scripts;         /* A dictionary of SHA1 -> Lua scripts */
+    unsigned long long lua_scripts_mem;  /* Cached scripts' memory + oh */
     mstime_t lua_time_limit;  /* Script timeout in milliseconds */
     mstime_t lua_time_start;  /* Start time of script, milliseconds time */
     int lua_write_dirty;  /* True if a write command was called during the
@@ -1583,6 +1591,11 @@ void startLoading(FILE *fp);
 void loadingProgress(off_t pos);
 void stopLoading(void);
 
+#define DISK_ERROR_TYPE_AOF 1       /* Don't accept writes: AOF errors. */
+#define DISK_ERROR_TYPE_RDB 2       /* Don't accept writes: RDB errors. */
+#define DISK_ERROR_TYPE_NONE 0      /* No problems, we can accept writes. */
+int writeCommandsDeniedByDiskError(void);
+
 /* RDB persistence */
 #include "rdb.h"
 int rdbSaveRio(rio *rdb, int *error, int flags, rdbSaveInfo *rsi);
@@ -1822,6 +1835,7 @@ int dbAsyncDelete(redisDb *db, robj *key);
 void emptyDbAsync(redisDb *db);
 void slotToKeyFlushAsync(void);
 size_t lazyfreeGetPendingObjectsCount(void);
+void freeObjAsync(robj *o);
 
 /* API to get key arguments from commands */
 int *getKeysFromCommand(struct redisCommand *cmd, robj **argv, int argc, int *numkeys);

src/t_set.c

@@ -516,11 +516,7 @@ void spopWithCountCommand(client *c) {
             sdsfree(sdsele);
         }
 
-        /* Assign the new set as the key value. */
-        incrRefCount(set); /* Protect the old set value. */
-        dbOverwrite(c->db,c->argv[1],newset);
-
-        /* Tranfer the old set to the client and release it. */
+        /* Transfer the old set to the client. */
         setTypeIterator *si;
         si = setTypeInitIterator(set);
         while((encoding = setTypeNext(si,&sdsele,&llele)) != -1) {
@@ -539,7 +535,9 @@ void spopWithCountCommand(client *c) {
             decrRefCount(objele);
         }
         setTypeReleaseIterator(si);
-        decrRefCount(set);
+
+        /* Assign the new set as the key value. */
+        dbOverwrite(c->db,c->argv[1],newset);
     }
 
     /* Don't propagate the command itself even if we incremented the

src/t_stream.c

@@ -548,13 +548,20 @@ int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
             si->master_fields_count = lpGetInteger(si->lp_ele);
             si->lp_ele = lpNext(si->lp,si->lp_ele); /* Seek first field. */
             si->master_fields_start = si->lp_ele;
-            /* Skip master fileds to seek the first entry. */
-            for (uint64_t i = 0; i < si->master_fields_count; i++)
-                si->lp_ele = lpNext(si->lp,si->lp_ele);
-            /* We are now pointing the zero term of the master entry. If
-             * we are iterating in reverse order, we need to seek the
-             * end of the listpack. */
-            if (si->rev) si->lp_ele = lpLast(si->lp);
+            /* We are now pointing to the first field of the master entry.
+             * We need to seek either the first or the last entry depending
+             * on the direction of the iteration. */
+            if (!si->rev) {
+                /* If we are iterating in normal order, skip the master fields
+                 * to seek the first actual entry. */
+                for (uint64_t i = 0; i < si->master_fields_count; i++)
+                    si->lp_ele = lpNext(si->lp,si->lp_ele);
+            } else {
+                /* If we are iterating in reverse direction, just seek the
+                 * last part of the last entry in the listpack (that is, the
+                 * fields count). */
+                si->lp_ele = lpLast(si->lp);
+            }
         } else if (si->rev) {
             /* If we are itereating in the reverse order, and this is not
              * the first entry emitted for this listpack, then we already
@@ -757,6 +764,7 @@ int streamDeleteItem(stream *s, streamID *id) {
         streamIteratorRemoveEntry(&si,&myid);
         deleted = 1;
     }
+    streamIteratorStop(&si);
     return deleted;
 }
 
@@ -1051,15 +1059,20 @@ int string2ull(const char *s, unsigned long long *value) {
  * to the client, otherwise C_OK is returned. The ID may be in incomplete
  * form, just stating the milliseconds time part of the stream. In such a case
  * the missing part is set according to the value of 'missing_seq' parameter.
+ *
  * The IDs "-" and "+" specify respectively the minimum and maximum IDs
- * that can be represented.
+ * that can be represented. If 'strict' is set to 1, "-" and "+" will be
+ * treated as an invalid ID.
  *
  * If 'c' is set to NULL, no reply is sent to the client. */
-int streamParseIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq) {
+int streamGenericParseIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq, int strict) {
     char buf[128];
     if (sdslen(o->ptr) > sizeof(buf)-1) goto invalid;
     memcpy(buf,o->ptr,sdslen(o->ptr)+1);
 
+    if (strict && (buf[0] == '-' || buf[0] == '+') && buf[1] == '\0')
+        goto invalid;
+
     /* Handle the "-" and "+" special cases. */
     if (buf[0] == '-' && buf[1] == '\0') {
         id->ms = 0;
@@ -1088,6 +1101,20 @@ invalid:
     return C_ERR;
 }
 
+/* Wrapper for streamGenericParseIDOrReply() with 'strict' argument set to
+ * 0, to be used when - and + are accetable IDs. */
+int streamParseIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq) {
+    return streamGenericParseIDOrReply(c,o,id,missing_seq,0);
+}
+
+/* Wrapper for streamGenericParseIDOrReply() with 'strict' argument set to
+ * 1, to be used when we want to return an error if the special IDs + or -
+ * are provided. */
+int streamParseStrictIDOrReply(client *c, robj *o, streamID *id, uint64_t missing_seq) {
+    return streamGenericParseIDOrReply(c,o,id,missing_seq,1);
+}
+
+
 /* XADD key [MAXLEN <count>] <ID or *> [field value] [field value] ... */
 void xaddCommand(client *c) {
     streamID id;
@@ -1125,7 +1152,7 @@ void xaddCommand(client *c) {
             maxlen_arg_idx = i;
         } else {
             /* If we are here is a syntax error or a valid ID. */
-            if (streamParseIDOrReply(c,c->argv[i],&id,0) != C_OK) return;
+            if (streamParseStrictIDOrReply(c,c->argv[i],&id,0) != C_OK) return;
             id_given = 1;
             break;
         }
@@ -1381,7 +1408,7 @@ void xreadCommand(client *c) {
             ids[id_idx].seq = UINT64_MAX;
             continue;
         }
-        if (streamParseIDOrReply(c,c->argv[i],ids+id_idx,0) != C_OK)
+        if (streamParseStrictIDOrReply(c,c->argv[i],ids+id_idx,0) != C_OK)
             goto cleanup;
     }
 
@@ -1655,7 +1682,7 @@ NULL
         streamID id;
         if (!strcmp(c->argv[4]->ptr,"$")) {
             id = s->last_id;
-        } else if (streamParseIDOrReply(c,c->argv[4],&id,0) != C_OK) {
+        } else if (streamParseStrictIDOrReply(c,c->argv[4],&id,0) != C_OK) {
             return;
         }
         streamCG *cg = streamCreateCG(s,grpname,sdslen(grpname),&id);
@@ -1732,7 +1759,7 @@ void xackCommand(client *c) {
     for (int j = 3; j < c->argc; j++) {
         streamID id;
         unsigned char buf[sizeof(streamID)];
-        if (streamParseIDOrReply(c,c->argv[j],&id,0) != C_OK) return;
+        if (streamParseStrictIDOrReply(c,c->argv[j],&id,0) != C_OK) return;
         streamEncodeID(buf,&id);
 
         /* Lookup the ID in the group PEL: it will have a reference to the
@@ -1990,7 +2017,7 @@ void xclaimCommand(client *c) {
     int j;
     for (j = 4; j < c->argc; j++) {
         streamID id;
-        if (streamParseIDOrReply(NULL,c->argv[j],&id,0) != C_OK) break;
+        if (streamParseStrictIDOrReply(NULL,c->argv[j],&id,0) != C_OK) break;
     }
     int last_id_arg = j-1; /* Next time we iterate the IDs we now the range. */
 
@@ -2049,7 +2076,7 @@ void xclaimCommand(client *c) {
     for (int j = 5; j <= last_id_arg; j++) {
         streamID id;
         unsigned char buf[sizeof(streamID)];
-        if (streamParseIDOrReply(c,c->argv[j],&id,0) != C_OK) return;
+        if (streamParseStrictIDOrReply(c,c->argv[j],&id,0) != C_OK) return;
         streamEncodeID(buf,&id);
 
         /* Lookup the ID in the group PEL. */
@@ -2132,13 +2159,13 @@ void xdelCommand(client *c) {
      * executed because at some point an invalid ID is parsed. */
     streamID id;
     for (int j = 2; j < c->argc; j++) {
-        if (streamParseIDOrReply(c,c->argv[j],&id,0) != C_OK) return;
+        if (streamParseStrictIDOrReply(c,c->argv[j],&id,0) != C_OK) return;
     }
 
     /* Actually apply the command. */
     int deleted = 0;
     for (int j = 2; j < c->argc; j++) {
-        streamParseIDOrReply(c,c->argv[j],&id,0); /* Retval already checked. */
+        streamParseStrictIDOrReply(c,c->argv[j],&id,0); /* Retval already checked. */
         deleted += streamDeleteItem(s,&id);
     }
 

src/t_zset.c

@@ -244,6 +244,61 @@ int zslDelete(zskiplist *zsl, double score, sds ele, zskiplistNode **node) {
     return 0; /* not found */
 }
 
+/* Update the score of an elmenent inside the sorted set skiplist.
+ * Note that the element must exist and must match 'score'.
+ * This function does not update the score in the hash table side, the
+ * caller should take care of it.
+ *
+ * Note that this function attempts to just update the node, in case after
+ * the score update, the node would be exactly at the same position.
+ * Otherwise the skiplist is modified by removing and re-adding a new
+ * element, which is more costly.
+ *
+ * The function returns the updated element skiplist node pointer. */
+zskiplistNode *zslUpdateScore(zskiplist *zsl, double curscore, sds ele, double newscore) {
+    zskiplistNode *update[ZSKIPLIST_MAXLEVEL], *x;
+    int i;
+
+    /* We need to seek to element to update to start: this is useful anyway,
+     * we'll have to update or remove it. */
+    x = zsl->header;
+    for (i = zsl->level-1; i >= 0; i--) {
+        while (x->level[i].forward &&
+                (x->level[i].forward->score < curscore ||
+                    (x->level[i].forward->score == curscore &&
+                     sdscmp(x->level[i].forward->ele,ele) < 0)))
+        {
+            x = x->level[i].forward;
+        }
+        update[i] = x;
+    }
+
+    /* Jump to our element: note that this function assumes that the
+     * element with the matching score exists. */
+    x = x->level[0].forward;
+    serverAssert(x && curscore == x->score && sdscmp(x->ele,ele) == 0);
+
+    /* If the node, after the score update, would be still exactly
+     * at the same position, we can just update the score without
+     * actually removing and re-inserting the element in the skiplist. */
+    if ((x->backward == NULL || x->backward->score < newscore) &&
+        (x->level[0].forward == NULL || x->level[0].forward->score > newscore))
+    {
+        x->score = newscore;
+        return x;
+    }
+
+    /* No way to reuse the old node: we need to remove and insert a new
+     * one at a different place. */
+    zslDeleteNode(zsl, x, update);
+    zskiplistNode *newnode = zslInsert(zsl,newscore,x->ele);
+    /* We reused the old node x->ele SDS string, free the node now
+     * since zslInsert created a new one. */
+    x->ele = NULL;
+    zslFreeNode(x);
+    return newnode;
+}
+
 int zslValueGteMin(double value, zrangespec *spec) {
     return spec->minex ? (value > spec->min) : (value >= spec->min);
 }
@@ -1341,13 +1396,7 @@ int zsetAdd(robj *zobj, double score, sds ele, int *flags, double *newscore) {
 
             /* Remove and re-insert when score changes. */
             if (score != curscore) {
-                zskiplistNode *node;
-                serverAssert(zslDelete(zs->zsl,curscore,ele,&node));
-                znode = zslInsert(zs->zsl,score,node->ele);
-                /* We reused the node->ele SDS string, free the node now
-                 * since zslInsert created a new one. */
-                node->ele = NULL;
-                zslFreeNode(node);
+                znode = zslUpdateScore(zs->zsl,curscore,ele,score);
                 /* Note that we did not removed the original element from
                  * the hash table representing the sorted set, so we just
                  * update the score. */

src/util.c

@@ -346,6 +346,7 @@ int string2ll(const char *s, size_t slen, long long *value) {
     int negative = 0;
     unsigned long long v;
 
+    /* A zero length string is not a valid number. */
     if (plen == slen)
         return 0;
 
@@ -355,6 +356,8 @@ int string2ll(const char *s, size_t slen, long long *value) {
         return 1;
     }
 
+    /* Handle negative numbers: just set a flag and continue like if it
+     * was a positive number. Later convert into negative. */
     if (p[0] == '-') {
         negative = 1;
         p++; plen++;
@@ -368,13 +371,11 @@ int string2ll(const char *s, size_t slen, long long *value) {
     if (p[0] >= '1' && p[0] <= '9') {
         v = p[0]-'0';
         p++; plen++;
-    } else if (p[0] == '0' && slen == 1) {
-        *value = 0;
-        return 1;
     } else {
         return 0;
     }
 
+    /* Parse all the other digits, checking for overflow at every step. */
     while (plen < slen && p[0] >= '0' && p[0] <= '9') {
         if (v > (ULLONG_MAX / 10)) /* Overflow. */
             return 0;
@@ -391,6 +392,8 @@ int string2ll(const char *s, size_t slen, long long *value) {
     if (plen < slen)
         return 0;
 
+    /* Convert to negative if needed, and do the final overflow check when
+     * converting from unsigned long long to long long. */
     if (negative) {
         if (v > ((unsigned long long)(-(LLONG_MIN+1))+1)) /* Overflow. */
             return 0;

src/version.h

@@ -1 +1 @@
-#define REDIS_VERSION "999.999.999"
+#define REDIS_VERSION "4.9.104"

src/zmalloc.c

@@ -183,7 +183,7 @@ size_t zmalloc_size(void *ptr) {
     return size+PREFIX_SIZE;
 }
 size_t zmalloc_usable(void *ptr) {
-    return zmalloc_usable(ptr)-PREFIX_SIZE;
+    return zmalloc_size(ptr)-PREFIX_SIZE;
 }
 #endif
 

tests/integration/psync2.tcl

@@ -33,9 +33,8 @@ start_server {} {
 
     set cycle 1
     while {([clock seconds]-$start_time) < $duration} {
-        test "PSYNC2: --- CYCLE $cycle ---" {
-            incr cycle
-        }
+        test "PSYNC2: --- CYCLE $cycle ---" {}
+        incr cycle
 
         # Create a random replication layout.
         # Start with switching master (this simulates a failover).
@@ -139,6 +138,11 @@ start_server {} {
             }
             assert {$sum == 4}
         }
+
+        # Limit anyway the maximum number of cycles. This is useful when the
+        # test is skipped via --only option of the test suite. In that case
+        # we don't want to see many seconds of this test being just skipped.
+        if {$cycle > 50} break
     }
 
     test "PSYNC2: Bring the master back again for next test" {

tests/support/test.tcl

@@ -1,6 +1,8 @@
 set ::num_tests 0
 set ::num_passed 0
 set ::num_failed 0
+set ::num_skipped 0
+set ::num_aborted 0
 set ::tests_failed {}
 
 proc fail {msg} {
@@ -68,10 +70,26 @@ proc test {name code {okpattern undefined}} {
     # abort if tagged with a tag to deny
     foreach tag $::denytags {
         if {[lsearch $::tags $tag] >= 0} {
+            incr ::num_aborted
+            send_data_packet $::test_server_fd ignore $name
             return
         }
     }
 
+    # abort if test name in skiptests
+    if {[lsearch $::skiptests $name] >= 0} {
+        incr ::num_skipped
+        send_data_packet $::test_server_fd skip $name
+        return
+    }
+
+    # abort if test name in skiptests
+    if {[llength $::only_tests] > 0 && [lsearch $::only_tests $name] < 0} {
+        incr ::num_skipped
+        send_data_packet $::test_server_fd skip $name
+        return
+    }
+
     # check if tagged with at least 1 tag to allow when there *is* a list
     # of tags to allow, because default policy is to run everything
     if {[llength $::allowtags] > 0} {
@@ -82,6 +100,8 @@ proc test {name code {okpattern undefined}} {
             }
         }
         if {$matched < 1} {
+            incr ::num_aborted
+            send_data_packet $::test_server_fd ignore $name
             return
         }
     }

tests/test_helper.tcl

@@ -74,7 +74,11 @@ set ::stack_logging 0
 set ::verbose 0
 set ::quiet 0
 set ::denytags {}
+set ::skiptests {}
 set ::allowtags {}
+set ::only_tests {}
+set ::single_tests {}
+set ::skip_till ""
 set ::external 0; # If "1" this means, we are running against external instance
 set ::file ""; # If set, runs only the tests in this comma separated list
 set ::curfile ""; # Hold the filename of the current suite
@@ -85,6 +89,8 @@ set ::last_progress [clock seconds]
 set ::active_servers {} ; # Pids of active Redis instances.
 set ::dont_clean 0
 set ::wait_server 0
+set ::stop_on_failure 0
+set ::loop 0
 
 # Set to 1 when we are running in client mode. The Redis test uses a
 # server-client model to run tests simultaneously. The server instance
@@ -255,6 +261,8 @@ proc accept_test_clients {fd addr port} {
 # testing: just used to signal that a given test started.
 # ok: a test was executed with success.
 # err: a test was executed with an error.
+# skip: a test was skipped by skipfile or individual test options.
+# ignore: a test was skipped by a group tag.
 # exception: there was a runtime exception while executing the test.
 # done: all the specified test file was processed, this test client is
 #       ready to accept a new task.
@@ -283,11 +291,24 @@ proc read_from_test_client fd {
             puts "\[[colorstr green $status]\]: $data"
         }
         set ::active_clients_task($fd) "(OK) $data"
+    } elseif {$status eq {skip}} {
+        if {!$::quiet} {
+            puts "\[[colorstr yellow $status]\]: $data"
+        }
+    } elseif {$status eq {ignore}} {
+        if {!$::quiet} {
+            puts "\[[colorstr cyan $status]\]: $data"
+        }
     } elseif {$status eq {err}} {
         set err "\[[colorstr red $status]\]: $data"
         puts $err
         lappend ::failed_tests $err
         set ::active_clients_task($fd) "(ERR) $data"
+            if {$::stop_on_failure} {
+            puts -nonewline "(Test stopped, press enter to continue)"
+            flush stdout
+            gets stdin
+        }
     } elseif {$status eq {exception}} {
         puts "\[[colorstr red $status]\]: $data"
         kill_clients
@@ -350,6 +371,9 @@ proc signal_idle_client fd {
         send_data_packet $fd run [lindex $::all_tests $::next_test]
         lappend ::active_clients $fd
         incr ::next_test
+        if {$::loop && $::next_test == [llength $::all_tests]} {
+            set ::next_test 0
+        }
     } else {
         lappend ::idle_clients $fd
         if {[llength $::active_clients] == 0} {
@@ -412,13 +436,19 @@ proc print_help_screen {} {
         "--stack-logging    Enable OSX leaks/malloc stack logging."
         "--accurate         Run slow randomized tests for more iterations."
         "--quiet            Don't show individual tests."
-        "--single <unit>    Just execute the specified unit (see next option)."
+        "--single <unit>    Just execute the specified unit (see next option). this option can be repeated."
         "--list-tests       List all the available test units."
+        "--only <test>      Just execute the specified test by test name. this option can be repeated."
+        "--skip-till <unit> Skip all units until (and including) the specified one."
         "--clients <num>    Number of test clients (default 16)."
         "--timeout <sec>    Test timeout in seconds (default 10 min)."
         "--force-failure    Force the execution of a test that always fails."
-        "--dont-clean       don't delete redis log files after the run"
-        "--wait-server      wait after server is started (so that you can attach a debugger)"
+        "--config <k> <v>   Extra config file argument."
+        "--skipfile <file>  Name of a file containing test names that should be skipped (one per line)."
+        "--dont-clean       Don't delete redis log files after the run."
+        "--stop             Blocks once the first test fails."
+        "--loop             Execute the specified set of tests forever."
+        "--wait-server      Wait after server is started (so that you can attach a debugger)."
         "--help             Print this help screen."
     } "\n"]
 }
@@ -436,6 +466,17 @@ for {set j 0} {$j < [llength $argv]} {incr j} {
             }
         }
         incr j
+    } elseif {$opt eq {--config}} {
+        set arg2 [lindex $argv [expr $j+2]]
+        lappend ::global_overrides $arg
+        lappend ::global_overrides $arg2
+        incr j 2
+    } elseif {$opt eq {--skipfile}} {
+        incr j
+        set fp [open $arg r]
+        set file_data [read $fp]
+        close $fp
+        set ::skiptests [split $file_data "\n"]
     } elseif {$opt eq {--valgrind}} {
         set ::valgrind 1
     } elseif {$opt eq {--stack-logging}} {
@@ -456,7 +497,13 @@ for {set j 0} {$j < [llength $argv]} {incr j} {
     } elseif {$opt eq {--force-failure}} {
         set ::force_failure 1
     } elseif {$opt eq {--single}} {
-        set ::all_tests $arg
+        lappend ::single_tests $arg
+        incr j
+    } elseif {$opt eq {--only}} {
+        lappend ::only_tests $arg
+        incr j
+    } elseif {$opt eq {--skiptill}} {
+        set ::skip_till $arg
         incr j
     } elseif {$opt eq {--list-tests}} {
         foreach t $::all_tests {
@@ -476,6 +523,10 @@ for {set j 0} {$j < [llength $argv]} {incr j} {
         set ::dont_clean 1
     } elseif {$opt eq {--wait-server}} {
         set ::wait_server 1
+    } elseif {$opt eq {--stop}} {
+        set ::stop_on_failure 1
+    } elseif {$opt eq {--loop}} {
+        set ::loop 1
     } elseif {$opt eq {--timeout}} {
         set ::timeout $arg
         incr j
@@ -488,6 +539,30 @@ for {set j 0} {$j < [llength $argv]} {incr j} {
     }
 }
 
+# If --skil-till option was given, we populate the list of single tests
+# to run with everything *after* the specified unit.
+if {$::skip_till != ""} {
+    set skipping 1
+    foreach t $::all_tests {
+        if {$skipping == 0} {
+            lappend ::single_tests $t
+        }
+        if {$t == $::skip_till} {
+            set skipping 0
+        }
+    }
+    if {$skipping} {
+        puts "test $::skip_till not found"
+        exit 0
+    }
+}
+
+# Override the list of tests with the specific tests we want to run
+# in case there was some filter, that is --single or --skip-till options.
+if {[llength $::single_tests] > 0} {
+    set ::all_tests $::single_tests
+}
+
 proc attach_to_replication_stream {} {
     set s [socket [srv 0 "host"] [srv 0 "port"]]
     fconfigure $s -translation binary

tests/unit/maxmemory.tcl

@@ -182,7 +182,7 @@ proc test_slave_buffers {cmd_count payload_len limit_memory pipeline} {
 
             # put the slave to sleep
             set rd_slave [redis_deferring_client]
-            $rd_slave debug sleep 60
+            $rd_slave debug sleep 300
 
             # send some 10mb woth of commands that don't increase the memory usage
             if {$pipeline == 1} {

tests/unit/type/zset.tcl

@@ -1185,4 +1185,30 @@ start_server {tags {"zset"}} {
         stressers ziplist
         stressers skiplist
     }
+
+    test {ZSET skiplist order consistency when elements are moved} {
+        set original_max [lindex [r config get zset-max-ziplist-entries] 1]
+        r config set zset-max-ziplist-entries 0
+        for {set times 0} {$times < 10} {incr times} {
+            r del zset
+            for {set j 0} {$j < 1000} {incr j} {
+                r zadd zset [randomInt 50] ele-[randomInt 10]
+            }
+
+            # Make sure that element ordering is correct
+            set prev_element {}
+            set prev_score -1
+            foreach {element score} [r zrange zset 0 -1 WITHSCORES] {
+                # Assert that elements are in increasing ordering
+                assert {
+                    $prev_score < $score ||
+                    ($prev_score == $score &&
+                     [string compare $prev_element $element] == -1)
+                }
+                set prev_element $element
+                set prev_score $score
+            }
+        }
+        r config set zset-max-ziplist-entries $original_max
+    }
 }