Redis 5.0 RC5.

Document that effects replication is Redis 5 default.
Fix scripting tests now that we default to commands repl.
2018-09-06 13:04:23 +02:00 · 2018-09-05 19:58:25 +02:00 · 2018-09-05 19:55:55 +02:00 · 2018-09-05 19:55:52 +02:00 · 2018-09-05 19:55:49 +02:00 · 2018-09-05 19:55:45 +02:00
42 changed files with 2294 additions and 277 deletions
@@ -602,6 +602,26 @@ slave-priority 100
 #
 # maxmemory-samples 5

+# Starting from Redis 5, by default a slave will ignore its maxmemory setting
+# (unless it is promoted to master after a failover or manually). It means
+# that the eviction of keys will be just handled by the master, sending the
+# DEL commands to the slave as keys evict in the master side.
+#
+# This behavior ensures that masters and slaves stay consistent, and is usually
+# what you want, however if your slave is writable, or you want the slave to have
+# a different memory setting, and you are sure all the writes performed to the
+# slave are idempotent, then you may change this default (but be sure to understand
+# what you are doing).
+#
+# Note that since the slave by default does not evict, it may end using more
+# memory than the one set via maxmemory (there are certain buffers that may
+# be larger on the slave, or data structures may sometimes take more memory and so
+# forth). So make sure you monitor your slaves and make sure they have enough
+# memory to never hit a real out-of-memory condition before the master hits
+# the configured maxmemory setting.
+#
+# slave-ingore-maxmemory yes
+
 ############################# LAZY FREEING ####################################

 # Redis has two primitives to delete keys. One is called DEL and is a blocking
@@ -1205,6 +1225,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
@@ -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 "${@}"
@@ -20,6 +20,21 @@
 # The port that this sentinel instance will run on
 port 26379

+# By default Redis Sentinel does not run as a daemon. Use 'yes' if you need it.
+# Note that Redis will write a pid file in /var/run/redis-sentinel.pid when
+# daemonized.
+daemonize no
+
+# When running daemonized, Redis Sentinel writes a pid file in
+# /var/run/redis-sentinel.pid by default. You can specify a custom pid file
+# location here.
+pidfile /var/run/redis-sentinel.pid
+
+# Specify the log file name. Also the empty string can be used to force
+# Sentinel to log on the standard output. Note that if you use standard
+# output for logging but daemonize, logs will be sent to /dev/null
+logfile ""
+
 # sentinel announce-ip <ip>
 # sentinel announce-port <port>
 #
@@ -798,7 +798,9 @@ int loadAppendOnlyFile(char *filename) {
    }

    /* This point can only be reached when EOF is reached without errors.
-     * If the client is in the middle of a MULTI/EXEC, log error and quit. */
+     * If the client is in the middle of a MULTI/EXEC, handle it as it was
+     * a short read, even if technically the protocol is correct: we want
+     * to remove the unprocessed tail and continue. */
    if (fakeClient->flags & CLIENT_MULTI) goto uxeof;

 loaded_ok: /* DB loaded, cleanup and return C_OK to the caller. */
@@ -126,12 +126,37 @@ void processUnblockedClients(void) {
         * the code is conceptually more correct this way. */
        if (!(c->flags & CLIENT_BLOCKED)) {
            if (c->querybuf && sdslen(c->querybuf) > 0) {
-                processInputBuffer(c);
+                processInputBufferAndReplicate(c);
            }
        }
    }
 }

+/* This function will schedule the client for reprocessing at a safe time.
+ *
+ * This is useful when a client was blocked for some reason (blocking opeation,
+ * CLIENT PAUSE, or whatever), because it may end with some accumulated query
+ * buffer that needs to be processed ASAP:
+ *
+ * 1. When a client is blocked, its readable handler is still active.
+ * 2. However in this case it only gets data into the query buffer, but the
+ *    query is not parsed or executed once there is enough to proceed as
+ *    usually (because the client is blocked... so we can't execute commands).
+ * 3. When the client is unblocked, without this function, the client would
+ *    have to write some query in order for the readable handler to finally
+ *    call processQueryBuffer*() on it.
+ * 4. With this function instead we can put the client in a queue that will
+ *    process it for queries ready to be executed at a safe time.
+ */
+void queueClientForReprocessing(client *c) {
+    /* The client may already be into the unblocked list because of a previous
+     * blocking operation, don't add back it into the list multiple times. */
+    if (!(c->flags & CLIENT_UNBLOCKED)) {
+        c->flags |= CLIENT_UNBLOCKED;
+        listAddNodeTail(server.unblocked_clients,c);
+    }
+}
+
 /* Unblock a client calling the right function depending on the kind
 * of operation the client is blocking for. */
 void unblockClient(client *c) {
@@ -152,12 +177,7 @@ void unblockClient(client *c) {
    server.blocked_clients_by_type[c->btype]--;
    c->flags &= ~CLIENT_BLOCKED;
    c->btype = BLOCKED_NONE;
-    /* The client may already be into the unblocked list because of a previous
-     * blocking operation, don't add back it into the list multiple times. */
-    if (!(c->flags & CLIENT_UNBLOCKED)) {
-        c->flags |= CLIENT_UNBLOCKED;
-        listAddNodeTail(server.unblocked_clients,c);
-    }
+    queueClientForReprocessing(c);
 }

 /* This function gets called when a blocked client timed out in order to
@@ -269,7 +289,7 @@ void handleClientsBlockedOnKeys(void) {
                        robj *dstkey = receiver->bpop.target;
                        int where = (receiver->lastcmd &&
                                     receiver->lastcmd->proc == blpopCommand) ?
-                                    LIST_HEAD : LIST_TAIL;
+                                     LIST_HEAD : LIST_TAIL;
                        robj *value = listTypePop(o,where);

                        if (value) {
@@ -285,7 +305,7 @@ void handleClientsBlockedOnKeys(void) {
                            {
                                /* If we failed serving the client we need
                                 * to also undo the POP operation. */
-                                    listTypePush(o,value,where);
+                                listTypePush(o,value,where);
                            }

                            if (dstkey) decrRefCount(dstkey);
@@ -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;
@@ -399,6 +399,10 @@ void loadServerConfigFromString(char *config) {
            if ((server.repl_slave_ro = yesnotoi(argv[1])) == -1) {
                err = "argument must be 'yes' or 'no'"; goto loaderr;
            }
+        } else if (!strcasecmp(argv[0],"slave-ignore-maxmemory") && argc == 2) {
+            if ((server.repl_slave_ignore_maxmemory = yesnotoi(argv[1])) == -1) {
+                err = "argument must be 'yes' or 'no'"; goto loaderr;
+            }
        } else if (!strcasecmp(argv[0],"rdbcompression") && argc == 2) {
            if ((server.rdb_compression = yesnotoi(argv[1])) == -1) {
                err = "argument must be 'yes' or 'no'"; goto loaderr;
@@ -440,10 +444,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;

@@ -669,6 +677,8 @@ void loadServerConfigFromString(char *config) {
            }
        } else if (!strcasecmp(argv[0],"lua-time-limit") && argc == 2) {
            server.lua_time_limit = strtoll(argv[1],NULL,10);
+        } else if (!strcasecmp(argv[0],"lua-replicate-commands") && argc == 2) {
+            server.lua_always_replicate_commands = yesnotoi(argv[1]);
        } else if (!strcasecmp(argv[0],"slowlog-log-slower-than") &&
                   argc == 2)
        {
@@ -1043,6 +1053,8 @@ void configSetCommand(client *c) {
      "slave-serve-stale-data",server.repl_serve_stale_data) {
    } config_set_bool_field(
      "slave-read-only",server.repl_slave_ro) {
+    } config_set_bool_field(
+      "slave-ignore-maxmemory",server.repl_slave_ignore_maxmemory) {
    } config_set_bool_field(
      "activerehashing",server.activerehashing) {
    } config_set_bool_field(
@@ -1072,6 +1084,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 +1172,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)
@@ -1175,7 +1189,7 @@ void configSetCommand(client *c) {
    } config_set_memory_field("maxmemory",server.maxmemory) {
        if (server.maxmemory) {
            if (server.maxmemory < zmalloc_used_memory()) {
-                serverLog(LL_WARNING,"WARNING: the new maxmemory value set via CONFIG SET is smaller than the current memory usage. This will result in keys eviction and/or inability to accept new write commands depending on the maxmemory-policy.");
+                serverLog(LL_WARNING,"WARNING: the new maxmemory value set via CONFIG SET is smaller than the current memory usage. This will result in key eviction and/or the inability to accept new write commands depending on the maxmemory-policy.");
            }
            freeMemoryIfNeeded();
        }
@@ -1329,7 +1343,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);
@@ -1347,6 +1361,8 @@ void configGetCommand(client *c) {
            server.repl_serve_stale_data);
    config_get_bool_field("slave-read-only",
            server.repl_slave_ro);
+    config_get_bool_field("slave-ignore-maxmemory",
+            server.repl_slave_ignore_maxmemory);
    config_get_bool_field("stop-writes-on-bgsave-error",
            server.stop_writes_on_bgsave_err);
    config_get_bool_field("daemonize", server.daemonize);
@@ -1375,6 +1391,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",
@@ -2041,6 +2059,7 @@ int rewriteConfig(char *path) {
    rewriteConfigStringOption(state,"cluster-announce-ip",server.cluster_announce_ip,NULL);
    rewriteConfigYesNoOption(state,"slave-serve-stale-data",server.repl_serve_stale_data,CONFIG_DEFAULT_SLAVE_SERVE_STALE_DATA);
    rewriteConfigYesNoOption(state,"slave-read-only",server.repl_slave_ro,CONFIG_DEFAULT_SLAVE_READ_ONLY);
+    rewriteConfigYesNoOption(state,"slave-ignore-maxmemory",server.repl_slave_ignore_maxmemory,CONFIG_DEFAULT_SLAVE_IGNORE_MAXMEMORY);
    rewriteConfigNumericalOption(state,"repl-ping-slave-period",server.repl_ping_slave_period,CONFIG_DEFAULT_REPL_PING_SLAVE_PERIOD);
    rewriteConfigNumericalOption(state,"repl-timeout",server.repl_timeout,CONFIG_DEFAULT_REPL_TIMEOUT);
    rewriteConfigBytesOption(state,"repl-backlog-size",server.repl_backlog_size,CONFIG_DEFAULT_REPL_BACKLOG_SIZE);
@@ -2098,7 +2117,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 +2127,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);
@@ -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]);
@@ -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);
@@ -444,6 +444,10 @@ int getMaxmemoryState(size_t *total, size_t *logical, size_t *tofree, float *lev
 * Otehrwise if we are over the memory limit, but not enough memory
 * was freed to return back under the limit, the function returns C_ERR. */
 int freeMemoryIfNeeded(void) {
+    /* By default slaves should ignore maxmemory and just be masters excat
+     * copies. */
+    if (server.masterhost && server.repl_slave_ignore_maxmemory) return C_OK;
+
    size_t mem_reported, mem_tofree, mem_freed;
    mstime_t latency, eviction_latency;
    long long delta;
@@ -560,10 +560,11 @@ sds latencyCommandGenSparkeline(char *event, struct latencyTimeSeries *ts) {

 /* LATENCY command implementations.
 *
- * LATENCY SAMPLES: return time-latency samples for the specified event.
+ * LATENCY HISTORY: return time-latency samples for the specified event.
 * LATENCY LATEST: return the latest latency for all the events classes.
 * LATENCY DOCTOR: returns an human readable analysis of instance latency.
 * LATENCY GRAPH: provide an ASCII graph of the latency of the specified event.
+ * LATENCY RESET: reset data of a specified event or all the data if no event provided.
 */
 void latencyCommand(client *c) {
    struct latencyTimeSeries *ts;
@@ -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. */
@@ -2,6 +2,9 @@
 GIT_SHA1=`(git show-ref --head --hash=8 2> /dev/null || echo 00000000) | head -n1`
 GIT_DIRTY=`git diff --no-ext-diff 2> /dev/null | wc -l`
 BUILD_ID=`uname -n`"-"`date +%s`
+if [ -n "$SOURCE_DATE_EPOCH" ]; then
+  BUILD_ID=$(date -u -d "@$SOURCE_DATE_EPOCH" +%s 2>/dev/null || date -u -r "$SOURCE_DATE_EPOCH" +%s 2>/dev/null || date -u %s)
+fi
 test -f release.h || touch release.h
 (cat release.h | grep SHA1 | grep $GIT_SHA1) && \
 (cat release.h | grep DIRTY | grep $GIT_DIRTY) && exit 0 # Already up-to-date
@@ -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;
@@ -33,7 +33,7 @@
 #include <math.h>
 #include <ctype.h>

-static void setProtocolError(const char *errstr, client *c, long pos);
+static void setProtocolError(const char *errstr, client *c);

 /* Return the size consumed from the allocator, for the specified SDS string,
 * including internal fragmentation. This function is used in order to compute
@@ -110,6 +110,7 @@ client *createClient(int fd) {
    c->fd = fd;
    c->name = NULL;
    c->bufpos = 0;
+    c->qb_pos = 0;
    c->querybuf = sdsempty();
    c->pending_querybuf = sdsempty();
    c->querybuf_peak = 0;
@@ -826,8 +827,7 @@ void freeClient(client *c) {
        serverLog(LL_WARNING,"Connection with master lost.");
        if (!(c->flags & (CLIENT_CLOSE_AFTER_REPLY|
                          CLIENT_CLOSE_ASAP|
-                          CLIENT_BLOCKED|
-                          CLIENT_UNBLOCKED)))
+                          CLIENT_BLOCKED)))
        {
            replicationCacheMaster(c);
            return;
@@ -1119,29 +1119,29 @@ int processInlineBuffer(client *c) {
    size_t querylen;

    /* Search for end of line */
-    newline = strchr(c->querybuf,'\n');
+    newline = strchr(c->querybuf+c->qb_pos,'\n');

    /* Nothing to do without a \r\n */
    if (newline == NULL) {
-        if (sdslen(c->querybuf) > PROTO_INLINE_MAX_SIZE) {
+        if (sdslen(c->querybuf)-c->qb_pos > PROTO_INLINE_MAX_SIZE) {
            addReplyError(c,"Protocol error: too big inline request");
-            setProtocolError("too big inline request",c,0);
+            setProtocolError("too big inline request",c);
        }
        return C_ERR;
    }

    /* Handle the \r\n case. */
-    if (newline && newline != c->querybuf && *(newline-1) == '\r')
+    if (newline && newline != c->querybuf+c->qb_pos && *(newline-1) == '\r')
        newline--, linefeed_chars++;

    /* Split the input buffer up to the \r\n */
-    querylen = newline-(c->querybuf);
-    aux = sdsnewlen(c->querybuf,querylen);
+    querylen = newline-(c->querybuf+c->qb_pos);
+    aux = sdsnewlen(c->querybuf+c->qb_pos,querylen);
    argv = sdssplitargs(aux,&argc);
    sdsfree(aux);
    if (argv == NULL) {
        addReplyError(c,"Protocol error: unbalanced quotes in request");
-        setProtocolError("unbalanced quotes in inline request",c,0);
+        setProtocolError("unbalanced quotes in inline request",c);
        return C_ERR;
    }

@@ -1151,8 +1151,8 @@ int processInlineBuffer(client *c) {
    if (querylen == 0 && c->flags & CLIENT_SLAVE)
        c->repl_ack_time = server.unixtime;

-    /* Leave data after the first line of the query in the buffer */
-    sdsrange(c->querybuf,querylen+linefeed_chars,-1);
+    /* Move querybuffer position to the next query in the buffer. */
+    c->qb_pos += querylen+linefeed_chars;

    /* Setup argv array on client structure */
    if (argc) {
@@ -1173,19 +1173,19 @@ int processInlineBuffer(client *c) {
    return C_OK;
 }

-/* Helper function. Trims query buffer to make the function that processes
- * multi bulk requests idempotent. */
+/* Helper function. Record protocol erro details in server log,
+ * and set the client as CLIENT_CLOSE_AFTER_REPLY. */
 #define PROTO_DUMP_LEN 128
-static void setProtocolError(const char *errstr, client *c, long pos) {
+static void setProtocolError(const char *errstr, client *c) {
    if (server.verbosity <= LL_VERBOSE) {
        sds client = catClientInfoString(sdsempty(),c);

        /* Sample some protocol to given an idea about what was inside. */
        char buf[256];
-        if (sdslen(c->querybuf) < PROTO_DUMP_LEN) {
-            snprintf(buf,sizeof(buf),"Query buffer during protocol error: '%s'", c->querybuf);
+        if (sdslen(c->querybuf)-c->qb_pos < PROTO_DUMP_LEN) {
+            snprintf(buf,sizeof(buf),"Query buffer during protocol error: '%s'", c->querybuf+c->qb_pos);
        } else {
-            snprintf(buf,sizeof(buf),"Query buffer during protocol error: '%.*s' (... more %zu bytes ...) '%.*s'", PROTO_DUMP_LEN/2, c->querybuf, sdslen(c->querybuf)-PROTO_DUMP_LEN, PROTO_DUMP_LEN/2, c->querybuf+sdslen(c->querybuf)-PROTO_DUMP_LEN/2);
+            snprintf(buf,sizeof(buf),"Query buffer during protocol error: '%.*s' (... more %zu bytes ...) '%.*s'", PROTO_DUMP_LEN/2, c->querybuf+c->qb_pos, sdslen(c->querybuf)-c->qb_pos-PROTO_DUMP_LEN, PROTO_DUMP_LEN/2, c->querybuf+sdslen(c->querybuf)-PROTO_DUMP_LEN/2);
        }

        /* Remove non printable chars. */
@@ -1201,7 +1201,6 @@ static void setProtocolError(const char *errstr, client *c, long pos) {
        sdsfree(client);
    }
    c->flags |= CLIENT_CLOSE_AFTER_REPLY;
-    sdsrange(c->querybuf,pos,-1);
 }

 /* Process the query buffer for client 'c', setting up the client argument
@@ -1217,7 +1216,6 @@ static void setProtocolError(const char *errstr, client *c, long pos) {
 * to be '*'. Otherwise for inline commands processInlineBuffer() is called. */
 int processMultibulkBuffer(client *c) {
    char *newline = NULL;
-    long pos = 0;
    int ok;
    long long ll;

@@ -1226,34 +1224,32 @@ int processMultibulkBuffer(client *c) {
        serverAssertWithInfo(c,NULL,c->argc == 0);

        /* Multi bulk length cannot be read without a \r\n */
-        newline = strchr(c->querybuf,'\r');
+        newline = strchr(c->querybuf+c->qb_pos,'\r');
        if (newline == NULL) {
-            if (sdslen(c->querybuf) > PROTO_INLINE_MAX_SIZE) {
+            if (sdslen(c->querybuf)-c->qb_pos > PROTO_INLINE_MAX_SIZE) {
                addReplyError(c,"Protocol error: too big mbulk count string");
-                setProtocolError("too big mbulk count string",c,0);
+                setProtocolError("too big mbulk count string",c);
            }
            return C_ERR;
        }

        /* Buffer should also contain \n */
-        if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2))
+        if (newline-(c->querybuf+c->qb_pos) > (ssize_t)(sdslen(c->querybuf)-c->qb_pos-2))
            return C_ERR;

        /* We know for sure there is a whole line since newline != NULL,
         * so go ahead and find out the multi bulk length. */
-        serverAssertWithInfo(c,NULL,c->querybuf[0] == '*');
-        ok = string2ll(c->querybuf+1,newline-(c->querybuf+1),&ll);
+        serverAssertWithInfo(c,NULL,c->querybuf[c->qb_pos] == '*');
+        ok = string2ll(c->querybuf+1+c->qb_pos,newline-(c->querybuf+1+c->qb_pos),&ll);
        if (!ok || ll > 1024*1024) {
            addReplyError(c,"Protocol error: invalid multibulk length");
-            setProtocolError("invalid mbulk count",c,pos);
+            setProtocolError("invalid mbulk count",c);
            return C_ERR;
        }

-        pos = (newline-c->querybuf)+2;
-        if (ll <= 0) {
-            sdsrange(c->querybuf,pos,-1);
-            return C_OK;
-        }
+        c->qb_pos = (newline-c->querybuf)+2;
+
+        if (ll <= 0) return C_OK;

        c->multibulklen = ll;

@@ -1266,64 +1262,67 @@ int processMultibulkBuffer(client *c) {
    while(c->multibulklen) {
        /* Read bulk length if unknown */
        if (c->bulklen == -1) {
-            newline = strchr(c->querybuf+pos,'\r');
+            newline = strchr(c->querybuf+c->qb_pos,'\r');
            if (newline == NULL) {
-                if (sdslen(c->querybuf) > PROTO_INLINE_MAX_SIZE) {
+                if (sdslen(c->querybuf)-c->qb_pos > PROTO_INLINE_MAX_SIZE) {
                    addReplyError(c,
                        "Protocol error: too big bulk count string");
-                    setProtocolError("too big bulk count string",c,0);
+                    setProtocolError("too big bulk count string",c);
                    return C_ERR;
                }
                break;
            }

            /* Buffer should also contain \n */
-            if (newline-(c->querybuf) > ((signed)sdslen(c->querybuf)-2))
+            if (newline-(c->querybuf+c->qb_pos) > (ssize_t)(sdslen(c->querybuf)-c->qb_pos-2))
                break;

-            if (c->querybuf[pos] != '$') {
+            if (c->querybuf[c->qb_pos] != '$') {
                addReplyErrorFormat(c,
                    "Protocol error: expected '$', got '%c'",
-                    c->querybuf[pos]);
-                setProtocolError("expected $ but got something else",c,pos);
+                    c->querybuf[c->qb_pos]);
+                setProtocolError("expected $ but got something else",c);
                return C_ERR;
            }

-            ok = string2ll(c->querybuf+pos+1,newline-(c->querybuf+pos+1),&ll);
+            ok = string2ll(c->querybuf+c->qb_pos+1,newline-(c->querybuf+c->qb_pos+1),&ll);
            if (!ok || ll < 0 || ll > server.proto_max_bulk_len) {
                addReplyError(c,"Protocol error: invalid bulk length");
-                setProtocolError("invalid bulk length",c,pos);
+                setProtocolError("invalid bulk length",c);
                return C_ERR;
            }

-            pos += newline-(c->querybuf+pos)+2;
+            c->qb_pos = newline-c->querybuf+2;
            if (ll >= PROTO_MBULK_BIG_ARG) {
-                size_t qblen;
-
                /* If we are going to read a large object from network
                 * try to make it likely that it will start at c->querybuf
                 * boundary so that we can optimize object creation
-                 * avoiding a large copy of data. */
-                sdsrange(c->querybuf,pos,-1);
-                pos = 0;
-                qblen = sdslen(c->querybuf);
-                /* Hint the sds library about the amount of bytes this string is
-                 * going to contain. */
-                if (qblen < (size_t)ll+2)
-                    c->querybuf = sdsMakeRoomFor(c->querybuf,ll+2-qblen);
+                 * avoiding a large copy of data.
+                 *
+                 * But only when the data we have not parsed is less than
+                 * or equal to ll+2. If the data length is greater than
+                 * ll+2, trimming querybuf is just a waste of time, because
+                 * at this time the querybuf contains not only our bulk. */
+                if (sdslen(c->querybuf)-c->qb_pos <= (size_t)ll+2) {
+                    sdsrange(c->querybuf,c->qb_pos,-1);
+                    c->qb_pos = 0;
+                    /* Hint the sds library about the amount of bytes this string is
+                     * going to contain. */
+                    c->querybuf = sdsMakeRoomFor(c->querybuf,ll+2);
+                }
            }
            c->bulklen = ll;
        }

        /* Read bulk argument */
-        if (sdslen(c->querybuf)-pos < (size_t)(c->bulklen+2)) {
+        if (sdslen(c->querybuf)-c->qb_pos < (size_t)(c->bulklen+2)) {
            /* Not enough data (+2 == trailing \r\n) */
            break;
        } else {
            /* Optimization: if the buffer contains JUST our bulk element
             * instead of creating a new object by *copying* the sds we
             * just use the current sds string. */
-            if (pos == 0 &&
+            if (c->qb_pos == 0 &&
                c->bulklen >= PROTO_MBULK_BIG_ARG &&
                sdslen(c->querybuf) == (size_t)(c->bulklen+2))
            {
@@ -1333,20 +1332,16 @@ int processMultibulkBuffer(client *c) {
                 * likely... */
                c->querybuf = sdsnewlen(SDS_NOINIT,c->bulklen+2);
                sdsclear(c->querybuf);
-                pos = 0;
            } else {
                c->argv[c->argc++] =
-                    createStringObject(c->querybuf+pos,c->bulklen);
-                pos += c->bulklen+2;
+                    createStringObject(c->querybuf+c->qb_pos,c->bulklen);
+                c->qb_pos += c->bulklen+2;
            }
            c->bulklen = -1;
            c->multibulklen--;
        }
    }

-    /* Trim to pos */
-    if (pos) sdsrange(c->querybuf,pos,-1);
-
    /* We're done when c->multibulk == 0 */
    if (c->multibulklen == 0) return C_OK;

@@ -1360,14 +1355,21 @@ int processMultibulkBuffer(client *c) {
 * pending query buffer, already representing a full command, to process. */
 void processInputBuffer(client *c) {
    server.current_client = c;
+
    /* Keep processing while there is something in the input buffer */
-    while(sdslen(c->querybuf)) {
+    while(c->qb_pos < sdslen(c->querybuf)) {
        /* Return if clients are paused. */
        if (!(c->flags & CLIENT_SLAVE) && clientsArePaused()) break;

        /* Immediately abort if the client is in the middle of something. */
        if (c->flags & CLIENT_BLOCKED) break;

+        /* Don't process input from the master while there is a busy script
+         * condition on the slave. We want just to accumulate the replication
+         * stream (instead of replying -BUSY like we do with other clients) and
+         * later resume the processing. */
+        if (server.lua_timedout && c->flags & CLIENT_MASTER) break;
+
        /* CLIENT_CLOSE_AFTER_REPLY closes the connection once the reply is
         * written to the client. Make sure to not let the reply grow after
         * this flag has been set (i.e. don't process more commands).
@@ -1377,7 +1379,7 @@ void processInputBuffer(client *c) {

        /* Determine request type when unknown. */
        if (!c->reqtype) {
-            if (c->querybuf[0] == '*') {
+            if (c->querybuf[c->qb_pos] == '*') {
                c->reqtype = PROTO_REQ_MULTIBULK;
            } else {
                c->reqtype = PROTO_REQ_INLINE;
@@ -1400,7 +1402,7 @@ void processInputBuffer(client *c) {
            if (processCommand(c) == C_OK) {
                if (c->flags & CLIENT_MASTER && !(c->flags & CLIENT_MULTI)) {
                    /* Update the applied replication offset of our master. */
-                    c->reploff = c->read_reploff - sdslen(c->querybuf);
+                    c->reploff = c->read_reploff - sdslen(c->querybuf) + c->qb_pos;
                }

                /* Don't reset the client structure for clients blocked in a
@@ -1416,9 +1418,35 @@ void processInputBuffer(client *c) {
            if (server.current_client == NULL) break;
        }
    }
+
+    /* Trim to pos */
+    if (c->qb_pos) {
+        sdsrange(c->querybuf,c->qb_pos,-1);
+        c->qb_pos = 0;
+    }
+
    server.current_client = NULL;
 }

+/* This is a wrapper for processInputBuffer that also cares about handling
+ * the replication forwarding to the sub-slaves, in case the client 'c'
+ * is flagged as master. Usually you want to call this instead of the
+ * raw processInputBuffer(). */
+void processInputBufferAndReplicate(client *c) {
+    if (!(c->flags & CLIENT_MASTER)) {
+        processInputBuffer(c);
+    } else {
+        size_t prev_offset = c->reploff;
+        processInputBuffer(c);
+        size_t applied = c->reploff - prev_offset;
+        if (applied) {
+            replicationFeedSlavesFromMasterStream(server.slaves,
+                    c->pending_querybuf, applied);
+            sdsrange(c->pending_querybuf,applied,-1);
+        }
+    }
+}
+
 void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
    client *c = (client*) privdata;
    int nread, readlen;
@@ -1438,7 +1466,9 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
    {
        ssize_t remaining = (size_t)(c->bulklen+2)-sdslen(c->querybuf);

-        if (remaining < readlen) readlen = remaining;
+        /* Note that the 'remaining' variable may be zero in some edge case,
+         * for example once we resume a blocked client after CLIENT PAUSE. */
+        if (remaining > 0 && remaining < readlen) readlen = remaining;
    }

    qblen = sdslen(c->querybuf);
@@ -1486,18 +1516,7 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
     * was actually applied to the master state: this quantity, and its
     * corresponding part of the replication stream, will be propagated to
     * the sub-slaves and to the replication backlog. */
-    if (!(c->flags & CLIENT_MASTER)) {
-        processInputBuffer(c);
-    } else {
-        size_t prev_offset = c->reploff;
-        processInputBuffer(c);
-        size_t applied = c->reploff - prev_offset;
-        if (applied) {
-            replicationFeedSlavesFromMasterStream(server.slaves,
-                    c->pending_querybuf, applied);
-            sdsrange(c->pending_querybuf,applied,-1);
-        }
-    }
+    processInputBufferAndReplicate(c);
 }

 void getClientsMaxBuffers(unsigned long *longest_output_list,
@@ -2120,11 +2139,10 @@ int clientsArePaused(void) {
        while ((ln = listNext(&li)) != NULL) {
            c = listNodeValue(ln);

-            /* Don't touch slaves and blocked clients. The latter pending
-             * requests be processed when unblocked. */
+            /* Don't touch slaves and blocked clients.
+             * The latter pending requests will be processed when unblocked. */
            if (c->flags & (CLIENT_SLAVE|CLIENT_BLOCKED)) continue;
-            c->flags |= CLIENT_UNBLOCKED;
-            listAddNodeTail(server.unblocked_clients,c);
+            queueClientForReprocessing(c);
        }
    }
    return server.clients_paused;
@@ -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);
@@ -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;
        }
    }
@@ -506,10 +512,13 @@ int luaRedisGenericCommand(lua_State *lua, int raise_error) {
     * could enlarge the memory usage are not allowed, but only if this is the
     * first write in the context of this script, otherwise we can't stop
     * in the middle. */
-    if (server.maxmemory && server.lua_write_dirty == 0 &&
+    if (server.maxmemory &&             /* Maxmemory is actually enabled. */
+        !server.loading &&              /* Don't care about mem if loading. */
+        !server.masterhost &&           /* Slave must execute the script. */
+        server.lua_write_dirty == 0 &&  /* Script had no side effects so far. */
        (cmd->flags & CMD_DENYOOM))
    {
-        if (freeMemoryIfNeeded() == C_ERR) {
+        if (getMaxmemoryState(NULL,NULL,NULL,NULL) != C_OK) {
            luaPushError(lua, shared.oomerr->ptr);
            goto cleanup;
        }
@@ -908,7 +917,6 @@ void scriptingInit(int setup) {
        server.lua_client = NULL;
        server.lua_caller = NULL;
        server.lua_timedout = 0;
-        server.lua_always_replicate_commands = 0; /* Only DEBUG can change it.*/
        ldbInit();
    }

@@ -919,6 +927,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 +1082,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,17 +1217,19 @@ 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;
 }

 /* This is the Lua script "count" hook that we use to detect scripts timeout. */
 void luaMaskCountHook(lua_State *lua, lua_Debug *ar) {
-    long long elapsed;
+    long long elapsed = mstime() - server.lua_time_start;
    UNUSED(ar);
    UNUSED(lua);

-    elapsed = mstime() - server.lua_time_start;
+    /* Set the timeout condition if not already set and the maximum
+     * execution time was reached. */
    if (elapsed >= server.lua_time_limit && server.lua_timedout == 0) {
        serverLog(LL_WARNING,"Lua slow script detected: still in execution after %lld milliseconds. You can try killing the script using the SCRIPT KILL command.",elapsed);
        server.lua_timedout = 1;
@@ -1240,6 +1252,7 @@ void evalGenericCommand(client *c, int evalsha) {
    lua_State *lua = server.lua;
    char funcname[43];
    long long numkeys;
+    long long initial_server_dirty = server.dirty;
    int delhook = 0, err;

    /* When we replicate whole scripts, we want the same PRNG sequence at
@@ -1336,9 +1349,7 @@ void evalGenericCommand(client *c, int evalsha) {
    server.lua_caller = c;
    server.lua_time_start = mstime();
    server.lua_kill = 0;
-    if (server.lua_time_limit > 0 && server.masterhost == NULL &&
-        ldb.active == 0)
-    {
+    if (server.lua_time_limit > 0 && ldb.active == 0) {
        lua_sethook(lua,luaMaskCountHook,LUA_MASKCOUNT,100000);
        delhook = 1;
    } else if (ldb.active) {
@@ -1359,6 +1370,8 @@ void evalGenericCommand(client *c, int evalsha) {
         * script timeout was detected. */
        aeCreateFileEvent(server.el,c->fd,AE_READABLE,
                          readQueryFromClient,c);
+        if (server.masterhost && server.master)
+            queueClientForReprocessing(server.master);
    }
    server.lua_caller = NULL;

@@ -1422,9 +1435,21 @@ void evalGenericCommand(client *c, int evalsha) {

            replicationScriptCacheAdd(c->argv[1]->ptr);
            serverAssertWithInfo(c,NULL,script != NULL);
-            rewriteClientCommandArgument(c,0,
-                resetRefCount(createStringObject("EVAL",4)));
-            rewriteClientCommandArgument(c,1,script);
+
+            /* If the script did not produce any changes in the dataset we want
+             * just to replicate it as SCRIPT LOAD, otherwise we risk running
+             * an aborted script on slaves (that may then produce results there)
+             * or just running a CPU costly read-only script on the slaves. */
+            if (server.dirty == initial_server_dirty) {
+                rewriteClientCommandVector(c,3,
+                    resetRefCount(createStringObject("SCRIPT",6)),
+                    resetRefCount(createStringObject("LOAD",4)),
+                    script);
+            } else {
+                rewriteClientCommandArgument(c,0,
+                    resetRefCount(createStringObject("EVAL",4)));
+                rewriteClientCommandArgument(c,1,script);
+            }
            forceCommandPropagation(c,PROPAGATE_REPL|PROPAGATE_AOF);
        }
    }
@@ -1488,6 +1513,8 @@ NULL
    } else if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"kill")) {
        if (server.lua_caller == NULL) {
            addReplySds(c,sdsnew("-NOTBUSY No scripts in execution right now.\r\n"));
+        } else if (server.lua_caller->flags & CLIENT_MASTER) {
+            addReplySds(c,sdsnew("-UNKILLABLE The busy script was sent by a master instance in the context of replication and cannot be killed.\r\n"));
        } else if (server.lua_write_dirty) {
            addReplySds(c,sdsnew("-UNKILLABLE Sorry the script already executed write commands against the dataset. You can either wait the script termination or kill the server in a hard way using the SHUTDOWN NOSAVE command.\r\n"));
        } else {
@@ -1716,7 +1743,7 @@ int ldbRemoveChild(pid_t pid) {
    return 0;
 }

-/* Return the number of children we still did not received termination
+/* Return the number of children we still did not receive termination
 * acknowledge via wait() in the parent process. */
 int ldbPendingChildren(void) {
    return listLength(ldb.children);
@@ -2628,7 +2628,7 @@ int sentinelSendPing(sentinelRedisInstance *ri) {
        ri->link->last_ping_time = mstime();
        /* We update the active ping time only if we received the pong for
         * the previous ping, otherwise we are technically waiting since the
-         * first ping that did not received a reply. */
+         * first ping that did not receive a reply. */
        if (ri->link->act_ping_time == 0)
            ri->link->act_ping_time = ri->link->last_ping_time;
        return 1;
@@ -3569,7 +3569,7 @@ void sentinelCheckSubjectivelyDown(sentinelRedisInstance *ri) {
        (mstime() - ri->link->cc_conn_time) >
        SENTINEL_MIN_LINK_RECONNECT_PERIOD &&
        ri->link->act_ping_time != 0 && /* There is a pending ping... */
-        /* The pending ping is delayed, and we did not received
+        /* The pending ping is delayed, and we did not receive
         * error replies as well. */
        (mstime() - ri->link->act_ping_time) > (ri->down_after_period/2) &&
        (mstime() - ri->link->last_pong_time) > (ri->down_after_period/2))
@@ -3725,7 +3725,7 @@ void sentinelAskMasterStateToOtherSentinels(sentinelRedisInstance *master, int f
         *
         * 1) We believe it is down, or there is a failover in progress.
         * 2) Sentinel is connected.
-         * 3) We did not received the info within SENTINEL_ASK_PERIOD ms. */
+         * 3) We did not receive the info within SENTINEL_ASK_PERIOD ms. */
        if ((master->flags & SRI_S_DOWN) == 0) continue;
        if (ri->link->disconnected) continue;
        if (!(flags & SENTINEL_ASK_FORCED) &&
@@ -214,7 +214,7 @@ struct redisCommand redisCommandTable[] = {
    {"hstrlen",hstrlenCommand,3,"rF",0,NULL,1,1,1,0,0},
    {"hkeys",hkeysCommand,2,"rS",0,NULL,1,1,1,0,0},
    {"hvals",hvalsCommand,2,"rS",0,NULL,1,1,1,0,0},
-    {"hgetall",hgetallCommand,2,"r",0,NULL,1,1,1,0,0},
+    {"hgetall",hgetallCommand,2,"rR",0,NULL,1,1,1,0,0},
    {"hexists",hexistsCommand,3,"rF",0,NULL,1,1,1,0,0},
    {"hscan",hscanCommand,-3,"rR",0,NULL,1,1,1,0,0},
    {"incrby",incrbyCommand,3,"wmF",0,NULL,1,1,1,0,0},
@@ -240,9 +240,9 @@ struct redisCommand redisCommandTable[] = {
    {"ping",pingCommand,-1,"tF",0,NULL,0,0,0,0,0},
    {"echo",echoCommand,2,"F",0,NULL,0,0,0,0,0},
    {"save",saveCommand,1,"as",0,NULL,0,0,0,0,0},
-    {"bgsave",bgsaveCommand,-1,"a",0,NULL,0,0,0,0,0},
-    {"bgrewriteaof",bgrewriteaofCommand,1,"a",0,NULL,0,0,0,0,0},
-    {"shutdown",shutdownCommand,-1,"alt",0,NULL,0,0,0,0,0},
+    {"bgsave",bgsaveCommand,-1,"as",0,NULL,0,0,0,0,0},
+    {"bgrewriteaof",bgrewriteaofCommand,1,"as",0,NULL,0,0,0,0,0},
+    {"shutdown",shutdownCommand,-1,"aslt",0,NULL,0,0,0,0,0},
    {"lastsave",lastsaveCommand,1,"RF",0,NULL,0,0,0,0,0},
    {"type",typeCommand,2,"rF",0,NULL,1,1,1,0,0},
    {"multi",multiCommand,1,"sF",0,NULL,0,0,0,0,0},
@@ -254,16 +254,16 @@ struct redisCommand redisCommandTable[] = {
    {"flushdb",flushdbCommand,-1,"w",0,NULL,0,0,0,0,0},
    {"flushall",flushallCommand,-1,"w",0,NULL,0,0,0,0,0},
    {"sort",sortCommand,-2,"wm",0,sortGetKeys,1,1,1,0,0},
-    {"info",infoCommand,-1,"lt",0,NULL,0,0,0,0,0},
+    {"info",infoCommand,-1,"ltR",0,NULL,0,0,0,0,0},
    {"monitor",monitorCommand,1,"as",0,NULL,0,0,0,0,0},
-    {"ttl",ttlCommand,2,"rF",0,NULL,1,1,1,0,0},
+    {"ttl",ttlCommand,2,"rFR",0,NULL,1,1,1,0,0},
    {"touch",touchCommand,-2,"rF",0,NULL,1,1,1,0,0},
-    {"pttl",pttlCommand,2,"rF",0,NULL,1,1,1,0,0},
+    {"pttl",pttlCommand,2,"rFR",0,NULL,1,1,1,0,0},
    {"persist",persistCommand,2,"wF",0,NULL,1,1,1,0,0},
    {"slaveof",slaveofCommand,3,"ast",0,NULL,0,0,0,0,0},
    {"role",roleCommand,1,"lst",0,NULL,0,0,0,0,0},
    {"debug",debugCommand,-2,"as",0,NULL,0,0,0,0,0},
-    {"config",configCommand,-2,"lat",0,NULL,0,0,0,0,0},
+    {"config",configCommand,-2,"last",0,NULL,0,0,0,0,0},
    {"subscribe",subscribeCommand,-2,"pslt",0,NULL,0,0,0,0,0},
    {"unsubscribe",unsubscribeCommand,-1,"pslt",0,NULL,0,0,0,0,0},
    {"psubscribe",psubscribeCommand,-2,"pslt",0,NULL,0,0,0,0,0},
@@ -275,24 +275,24 @@ struct redisCommand redisCommandTable[] = {
    {"cluster",clusterCommand,-2,"a",0,NULL,0,0,0,0,0},
    {"restore",restoreCommand,-4,"wm",0,NULL,1,1,1,0,0},
    {"restore-asking",restoreCommand,-4,"wmk",0,NULL,1,1,1,0,0},
-    {"migrate",migrateCommand,-6,"w",0,migrateGetKeys,0,0,0,0,0},
+    {"migrate",migrateCommand,-6,"wR",0,migrateGetKeys,0,0,0,0,0},
    {"asking",askingCommand,1,"F",0,NULL,0,0,0,0,0},
    {"readonly",readonlyCommand,1,"F",0,NULL,0,0,0,0,0},
    {"readwrite",readwriteCommand,1,"F",0,NULL,0,0,0,0,0},
-    {"dump",dumpCommand,2,"r",0,NULL,1,1,1,0,0},
-    {"object",objectCommand,-2,"r",0,NULL,2,2,1,0,0},
-    {"memory",memoryCommand,-2,"r",0,NULL,0,0,0,0,0},
+    {"dump",dumpCommand,2,"rR",0,NULL,1,1,1,0,0},
+    {"object",objectCommand,-2,"rR",0,NULL,2,2,1,0,0},
+    {"memory",memoryCommand,-2,"rR",0,NULL,0,0,0,0,0},
    {"client",clientCommand,-2,"as",0,NULL,0,0,0,0,0},
    {"eval",evalCommand,-3,"s",0,evalGetKeys,0,0,0,0,0},
    {"evalsha",evalShaCommand,-3,"s",0,evalGetKeys,0,0,0,0,0},
-    {"slowlog",slowlogCommand,-2,"a",0,NULL,0,0,0,0,0},
+    {"slowlog",slowlogCommand,-2,"aR",0,NULL,0,0,0,0,0},
    {"script",scriptCommand,-2,"s",0,NULL,0,0,0,0,0},
    {"time",timeCommand,1,"RF",0,NULL,0,0,0,0,0},
    {"bitop",bitopCommand,-4,"wm",0,NULL,2,-1,1,0,0},
    {"bitcount",bitcountCommand,-2,"r",0,NULL,1,1,1,0,0},
    {"bitpos",bitposCommand,-3,"r",0,NULL,1,1,1,0,0},
    {"wait",waitCommand,3,"s",0,NULL,0,0,0,0,0},
-    {"command",commandCommand,0,"lt",0,NULL,0,0,0,0,0},
+    {"command",commandCommand,0,"ltR",0,NULL,0,0,0,0,0},
    {"geoadd",geoaddCommand,-5,"wm",0,NULL,1,1,1,0,0},
    {"georadius",georadiusCommand,-6,"w",0,georadiusGetKeys,1,1,1,0,0},
    {"georadius_ro",georadiusroCommand,-6,"r",0,georadiusGetKeys,1,1,1,0,0},
@@ -310,14 +310,14 @@ struct redisCommand redisCommandTable[] = {
    {"xrange",xrangeCommand,-4,"r",0,NULL,1,1,1,0,0},
    {"xrevrange",xrevrangeCommand,-4,"r",0,NULL,1,1,1,0,0},
    {"xlen",xlenCommand,2,"rF",0,NULL,1,1,1,0,0},
-    {"xread",xreadCommand,-3,"rs",0,xreadGetKeys,1,1,1,0,0},
-    {"xreadgroup",xreadCommand,-3,"ws",0,xreadGetKeys,1,1,1,0,0},
+    {"xread",xreadCommand,-4,"rs",0,xreadGetKeys,1,1,1,0,0},
+    {"xreadgroup",xreadCommand,-7,"ws",0,xreadGetKeys,1,1,1,0,0},
    {"xgroup",xgroupCommand,-2,"wm",0,NULL,2,2,1,0,0},
-    {"xack",xackCommand,-3,"wF",0,NULL,1,1,1,0,0},
+    {"xack",xackCommand,-4,"wF",0,NULL,1,1,1,0,0},
    {"xpending",xpendingCommand,-3,"r",0,NULL,1,1,1,0,0},
-    {"xclaim",xclaimCommand,-5,"wF",0,NULL,1,1,1,0,0},
+    {"xclaim",xclaimCommand,-6,"wF",0,NULL,1,1,1,0,0},
    {"xinfo",xinfoCommand,-2,"r",0,NULL,2,2,1,0,0},
-    {"xdel",xdelCommand,-2,"wF",0,NULL,1,1,1,0,0},
+    {"xdel",xdelCommand,-3,"wF",0,NULL,1,1,1,0,0},
    {"xtrim",xtrimCommand,-2,"wF",0,NULL,1,1,1,0,0},
    {"post",securityWarningCommand,-1,"lt",0,NULL,0,0,0,0,0},
    {"host:",securityWarningCommand,-1,"lt",0,NULL,0,0,0,0,0},
@@ -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,
@@ -1476,6 +1491,7 @@ void createSharedObjects(void) {
    shared.rpop = createStringObject("RPOP",4);
    shared.lpop = createStringObject("LPOP",4);
    shared.lpush = createStringObject("LPUSH",5);
+    shared.rpoplpush = createStringObject("RPOPLPUSH",9);
    shared.zpopmin = createStringObject("ZPOPMIN",7);
    shared.zpopmax = createStringObject("ZPOPMAX",7);
    for (j = 0; j < OBJ_SHARED_INTEGERS; j++) {
@@ -1512,7 +1528,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;
@@ -1632,6 +1649,7 @@ void initServerConfig(void) {
    server.repl_syncio_timeout = CONFIG_REPL_SYNCIO_TIMEOUT;
    server.repl_serve_stale_data = CONFIG_DEFAULT_SLAVE_SERVE_STALE_DATA;
    server.repl_slave_ro = CONFIG_DEFAULT_SLAVE_READ_ONLY;
+    server.repl_slave_ignore_maxmemory = CONFIG_DEFAULT_SLAVE_IGNORE_MAXMEMORY;
    server.repl_slave_lazy_flush = CONFIG_DEFAULT_SLAVE_LAZY_FLUSH;
    server.repl_down_since = 0; /* Never connected, repl is down since EVER. */
    server.repl_disable_tcp_nodelay = CONFIG_DEFAULT_REPL_DISABLE_TCP_NODELAY;
@@ -1697,6 +1715,12 @@ void initServerConfig(void) {
    server.assert_line = 0;
    server.bug_report_start = 0;
    server.watchdog_period = 0;
+
+    /* By default we want scripts to be always replicated by effects
+     * (single commands executed by the script), and not by sending the
+     * script to the slave / AOF. This is the new way starting from
+     * Redis 5. However it is possible to revert it via redis.conf. */
+    server.lua_always_replicate_commands = 1;
 }

 extern char **environ;
@@ -1991,6 +2015,7 @@ void initServer(void) {
            server.syslog_facility);
    }

+    server.hz = server.config_hz;
    server.pid = getpid();
    server.current_client = NULL;
    server.clients = listCreate();
@@ -2574,16 +2599,21 @@ int processCommand(client *c) {
     *
     * First we try to free some memory if possible (if there are volatile
     * keys in the dataset). If there are not the only thing we can do
-     * is returning an error. */
-    if (server.maxmemory) {
-        int retval = freeMemoryIfNeeded();
+     * is returning an error.
+     *
+     * Note that we do not want to reclaim memory if we are here re-entering
+     * the event loop since there is a busy Lua script running in timeout
+     * condition, to avoid mixing the propagation of scripts with the propagation
+     * of DELs due to eviction. */
+    if (server.maxmemory && !server.lua_timedout) {
+        int out_of_memory = freeMemoryIfNeeded() == C_ERR;
        /* freeMemoryIfNeeded may flush slave output buffers. This may result
         * into a slave, that may be the active client, to be freed. */
        if (server.current_client == NULL) return C_ERR;

        /* It was impossible to free enough memory, and the command the client
         * is trying to execute is denied during OOM conditions? Error. */
-        if ((c->cmd->flags & CMD_DENYOOM) && retval == C_ERR) {
+        if ((c->cmd->flags & CMD_DENYOOM) && out_of_memory) {
            flagTransaction(c);
            addReply(c, shared.oomerr);
            return C_OK;
@@ -2592,17 +2622,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 +2815,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 +3127,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 +3151,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 +3179,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 +3199,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 +3224,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 +3264,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 +3572,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 */
@@ -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
@@ -129,6 +131,7 @@ typedef long long mstime_t; /* millisecond time type. */
 #define CONFIG_DEFAULT_REPL_DISKLESS_SYNC_DELAY 5
 #define CONFIG_DEFAULT_SLAVE_SERVE_STALE_DATA 1
 #define CONFIG_DEFAULT_SLAVE_READ_ONLY 1
+#define CONFIG_DEFAULT_SLAVE_IGNORE_MAXMEMORY 1
 #define CONFIG_DEFAULT_SLAVE_ANNOUNCE_IP NULL
 #define CONFIG_DEFAULT_SLAVE_ANNOUNCE_PORT 0
 #define CONFIG_DEFAULT_REPL_DISABLE_TCP_NODELAY 0
@@ -708,6 +711,7 @@ typedef struct client {
    redisDb *db;            /* Pointer to currently SELECTed DB. */
    robj *name;             /* As set by CLIENT SETNAME. */
    sds querybuf;           /* Buffer we use to accumulate client queries. */
+    size_t qb_pos;          /* The position we have read in querybuf. */
    sds pending_querybuf;   /* If this client is flagged as master, this buffer
                               represents the yet not applied portion of the
                               replication stream that we are receiving from
@@ -779,7 +783,7 @@ struct sharedObjectsStruct {
    *masterdownerr, *roslaveerr, *execaborterr, *noautherr, *noreplicaserr,
    *busykeyerr, *oomerr, *plus, *messagebulk, *pmessagebulk, *subscribebulk,
    *unsubscribebulk, *psubscribebulk, *punsubscribebulk, *del, *unlink,
-    *rpop, *lpop, *lpush, *zpopmin, *zpopmax, *emptyscan,
+    *rpop, *lpop, *lpush, *rpoplpush, *zpopmin, *zpopmax, *emptyscan,
    *select[PROTO_SHARED_SELECT_CMDS],
    *integers[OBJ_SHARED_INTEGERS],
    *mbulkhdr[OBJ_SHARED_BULKHDR_LEN], /* "*<value>\r\n" */
@@ -851,6 +855,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 +928,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 */
@@ -1149,6 +1158,7 @@ struct redisServer {
    time_t repl_transfer_lastio; /* Unix time of the latest read, for timeout */
    int repl_serve_stale_data; /* Serve stale data when link is down? */
    int repl_slave_ro;          /* Slave is read only? */
+    int repl_slave_ignore_maxmemory;    /* If true slaves do not evict. */
    time_t repl_down_since; /* Unix time at which link with master went down */
    int repl_disable_tcp_nodelay;   /* Disable TCP_NODELAY after SYNC? */
    int slave_priority;             /* Reported in INFO and used by Sentinel. */
@@ -1227,6 +1237,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
@@ -1407,6 +1418,7 @@ void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask);
 void *addDeferredMultiBulkLength(client *c);
 void setDeferredMultiBulkLength(client *c, void *node, long length);
 void processInputBuffer(client *c);
+void processInputBufferAndReplicate(client *c);
 void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask);
 void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask);
 void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask);
@@ -1583,6 +1595,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 +1839,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);
@@ -1866,6 +1884,7 @@ sds luaCreateFunction(client *c, lua_State *lua, robj *body);
 void processUnblockedClients(void);
 void blockClient(client *c, int btype);
 void unblockClient(client *c);
+void queueClientForReprocessing(client *c);
 void replyToBlockedClientTimedOut(client *c);
 int getTimeoutFromObjectOrReply(client *c, robj *object, mstime_t *timeout, int unit);
 void disconnectAllBlockedClients(void);
@@ -596,6 +596,9 @@ void rpoplpushCommand(client *c) {
        signalModifiedKey(c->db,touchedkey);
        decrRefCount(touchedkey);
        server.dirty++;
+        if (c->lastcmd->proc == brpoplpushCommand) {
+            rewriteClientCommandVector(c,3,shared.rpoplpush,c->argv[1],c->argv[2]);
+        }
    }
 }

@@ -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
@@ -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;
    }

@@ -1409,7 +1436,7 @@ void xreadCommand(client *c) {
                 * synchronously in case the group top item delivered is smaller
                 * than what the stream has inside. */
                streamID *last = &groups[i]->last_id;
-                if (streamCompareID(&s->last_id, last) > 0) {
+                if (s->length && (streamCompareID(&s->last_id, last) > 0)) {
                    serve_synchronously = 1;
                    *gt = *last;
                }
@@ -1417,7 +1444,7 @@ void xreadCommand(client *c) {
        } else {
            /* For consumers without a group, we serve synchronously if we can
             * actually provide at least one item from the stream. */
-            if (streamCompareID(&s->last_id, gt) > 0) {
+            if (s->length && (streamCompareID(&s->last_id, gt) > 0)) {
                serve_synchronously = 1;
            }
        }
@@ -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
@@ -1988,9 +2015,9 @@ void xclaimCommand(client *c) {
     * the client successfully claimed some message, so it should be
     * executed in a "all or nothing" fashion. */
    int j;
-    for (j = 4; j < c->argc; j++) {
+    for (j = 5; 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. */

@@ -2013,12 +2040,12 @@ void xclaimCommand(client *c) {
        } else if (!strcasecmp(opt,"TIME") && moreargs) {
            j++;
            if (getLongLongFromObjectOrReply(c,c->argv[j],&deliverytime,
-                "Invalid IDLE option argument for XCLAIM")
+                "Invalid TIME option argument for XCLAIM")
                != C_OK) return;
        } else if (!strcasecmp(opt,"RETRYCOUNT") && moreargs) {
            j++;
            if (getLongLongFromObjectOrReply(c,c->argv[j],&retrycount,
-                "Invalid IDLE option argument for XCLAIM")
+                "Invalid RETRYCOUNT option argument for XCLAIM")
                != C_OK) return;
        } else {
            addReplyErrorFormat(c,"Unrecognized XCLAIM option '%s'",opt);
@@ -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);
    }

@@ -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. */
@@ -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;
@@ -1 +1 @@
-#define REDIS_VERSION "999.999.999"
+#define REDIS_VERSION "4.9.105"
@@ -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

@@ -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" {
@@ -266,3 +266,47 @@ foreach dl {no yes} {
        }
    }
 }
+
+start_server {tags {"repl"}} {
+    set master [srv 0 client]
+    set master_host [srv 0 host]
+    set master_port [srv 0 port]
+    set load_handle0 [start_write_load $master_host $master_port 3]
+    start_server {} {
+        test "Master stream is correctly processed while the slave has a script in -BUSY state" {
+            set slave [srv 0 client]
+            puts [srv 0 port]
+            $slave config set lua-time-limit 500
+            $slave slaveof $master_host $master_port
+
+            # Wait for the slave to be online
+            wait_for_condition 500 100 {
+                [lindex [$slave role] 3] eq {connected}
+            } else {
+                fail "Slave still not connected after some time"
+            }
+
+            # Wait some time to make sure the master is sending data
+            # to the slave.
+            after 5000
+
+            # Stop the ability of the slave to process data by sendig
+            # a script that will put it in BUSY state.
+            $slave eval {for i=1,3000000000 do end} 0
+
+            # Wait some time again so that more master stream will
+            # be processed.
+            after 2000
+
+            # Stop the write load
+            stop_write_load $load_handle0
+
+            # number of keys
+            wait_for_condition 500 100 {
+                [$master debug digest] eq [$slave debug digest]
+            } else {
+                fail "Different datasets between slave and master"
+            }
+        }
+    }
+}
@@ -17,7 +17,7 @@ test "Basic failover works if the master is down" {
        wait_for_condition 1000 50 {
            [lindex [S $id SENTINEL GET-MASTER-ADDR-BY-NAME mymaster] 1] != $old_port
        } else {
-            fail "At least one Sentinel did not received failover info"
+            fail "At least one Sentinel did not receive failover info"
        }
    }
    restart_instance redis $master_id
@@ -108,7 +108,7 @@ test "Failover works if we configure for absolute agreement" {
        wait_for_condition 1000 50 {
            [lindex [S $id SENTINEL GET-MASTER-ADDR-BY-NAME mymaster] 1] != $old_port
        } else {
-            fail "At least one Sentinel did not received failover info"
+            fail "At least one Sentinel did not receive failover info"
        }
    }
    restart_instance redis $master_id
@@ -16,7 +16,7 @@ test "We can failover with Sentinel 1 crashed" {
            wait_for_condition 1000 50 {
                [lindex [S $id SENTINEL GET-MASTER-ADDR-BY-NAME mymaster] 1] != $old_port
            } else {
-                fail "Sentinel $id did not received failover info"
+                fail "Sentinel $id did not receive failover info"
            }
        }
    }
@@ -30,7 +30,7 @@ test "After Sentinel 1 is restarted, its config gets updated" {
    wait_for_condition 1000 50 {
        [lindex [S 1 SENTINEL GET-MASTER-ADDR-BY-NAME mymaster] 1] != $old_port
    } else {
-        fail "Restarted Sentinel did not received failover info"
+        fail "Restarted Sentinel did not receive failover info"
    }
 }

@@ -36,7 +36,7 @@ proc 02_crash_and_failover {} {
                wait_for_condition 1000 50 {
                    [lindex [S $id SENTINEL GET-MASTER-ADDR-BY-NAME mymaster] 1] != $old_port
                } else {
-                    fail "At least one Sentinel did not received failover info"
+                    fail "At least one Sentinel did not receive failover info"
                }
            }
            restart_instance redis $master_id
@@ -12,7 +12,7 @@ test "Manual failover works" {
        wait_for_condition 1000 50 {
            [lindex [S $id SENTINEL GET-MASTER-ADDR-BY-NAME mymaster] 1] != $old_port
        } else {
-            fail "At least one Sentinel did not received failover info"
+            fail "At least one Sentinel did not receive failover info"
        }
    }
    set addr [S 0 SENTINEL GET-MASTER-ADDR-BY-NAME mymaster]
@@ -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
        }
    }
@@ -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
@@ -33,7 +33,7 @@ start_server {tags {"dump"}} {
        set now [clock milliseconds]
        r restore foo [expr $now+3000] $encoded absttl
        set ttl [r pttl foo]
-        assert {$ttl >= 2998 && $ttl <= 3000}
+        assert {$ttl >= 2990 && $ttl <= 3000}
        r get foo
    } {bar}
    
@@ -44,7 +44,7 @@ start_server {tags {"dump"}} {
        r config set maxmemory-policy allkeys-lru
        r restore foo 0 $encoded idletime 1000
        set idle [r object idletime foo]
-        assert {$idle >= 1000 && $idle <= 1002}
+        assert {$idle >= 1000 && $idle <= 1010}
        r get foo
    } {bar}
    
@@ -1,7 +1,7 @@
 start_server {tags {"introspection"}} {
    test {CLIENT LIST} {
        r client list
-    } {*addr=*:* fd=* age=* idle=* flags=N db=9 sub=0 psub=0 multi=-1 qbuf=0 qbuf-free=* obl=0 oll=0 omem=0 events=r cmd=client*}
+    } {*addr=*:* fd=* age=* idle=* flags=N db=9 sub=0 psub=0 multi=-1 qbuf=26 qbuf-free=* obl=0 oll=0 omem=0 events=r cmd=client*}

    test {MONITOR can log executed commands} {
        set rd [redis_deferring_client]
@@ -143,9 +143,11 @@ start_server {tags {"maxmemory"}} {
    }
 }

-proc test_slave_buffers {cmd_count payload_len limit_memory pipeline} {
+proc test_slave_buffers {test_name cmd_count payload_len limit_memory pipeline} {
    start_server {tags {"maxmemory"}} {
        start_server {} {
+        set slave_pid [s process_id]
+        test "$test_name" {
            set slave [srv 0 client]
            set slave_host [srv 0 host]
            set slave_port [srv 0 port]
@@ -158,8 +160,10 @@ proc test_slave_buffers {cmd_count payload_len limit_memory pipeline} {
                $master setrange "key:$j" 100000 asdf
            }

+            # make sure master doesn't disconnect slave because of timeout
+            $master config set repl-timeout 300 ;# 5 minutes
            $master config set maxmemory-policy allkeys-random
-            $master config set client-output-buffer-limit "slave 100000000 100000000 60"
+            $master config set client-output-buffer-limit "slave 100000000 100000000 300"
            $master config set repl-backlog-size [expr {10*1024}]

            $slave slaveof $master_host $master_port
@@ -182,9 +186,9 @@ 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
+            exec kill -SIGSTOP $slave_pid

-            # send some 10mb woth of commands that don't increase the memory usage
+            # send some 10mb worth of commands that don't increase the memory usage
            if {$pipeline == 1} {
                set rd_master [redis_deferring_client -1]
                for {set k 0} {$k < $cmd_count} {incr k} {
@@ -218,19 +222,21 @@ proc test_slave_buffers {cmd_count payload_len limit_memory pipeline} {
            set delta_no_repl [expr {$killed_used_no_repl - $used_no_repl}]
            assert {$killed_slave_buf == 0}
            assert {$delta_no_repl > -50*1024 && $delta_no_repl < 50*1024} ;# 1 byte unaccounted for, with 1M commands will consume some 1MB
+
+        }
+        # unfreeze slave process (after the 'test' succeeded or failed, but before we attempt to terminate the server
+        exec kill -SIGCONT $slave_pid
        }
    }
 }

-test {slave buffer are counted correctly} {
-    # we wanna use many small commands, and we don't wanna wait long
-    # so we need to use a pipeline (redis_deferring_client)
-    # that may cause query buffer to fill and induce eviction, so we disable it
-    test_slave_buffers 1000000 10 0 1
-}
+# test that slave buffer are counted correctly
+# we wanna use many small commands, and we don't wanna wait long
+# so we need to use a pipeline (redis_deferring_client)
+# that may cause query buffer to fill and induce eviction, so we disable it
+test_slave_buffers {slave buffer are counted correctly} 1000000 10 0 1

-test {slave buffer don't induce eviction} {
-    # test again with fewer (and bigger) commands without pipeline, but with eviction
-    test_slave_buffers 100000 100 1 0
-}
+# test that slave buffer don't induce eviction
+# test again with fewer (and bigger) commands without pipeline, but with eviction
+test_slave_buffers "slave buffer don't induce eviction" 100000 100 1 0

@@ -201,7 +201,7 @@ start_server {tags {"defrag"}} {
                assert {$frag < 1.1}
                # due to high fragmentation, 10hz, and active-defrag-cycle-max set to 75,
                # we expect max latency to be not much higher than 75ms
-                assert {$max_latency <= 80}
+                assert {$max_latency <= 120}
            }
            # verify the data isn't corrupted or changed
            set newdigest [r debug digest]
@@ -148,9 +148,11 @@ start_server {tags {"scripting"}} {

    test {EVAL - Scripts can't run certain commands} {
        set e {}
+        r debug lua-always-replicate-commands 0
        catch {
            r eval "redis.pcall('randomkey'); return redis.pcall('set','x','ciao')" 0
        } e
+        r debug lua-always-replicate-commands 1
        set e
    } {*not allowed after*}

@@ -299,9 +301,12 @@ start_server {tags {"scripting"}} {
    } {b534286061d4b9e4026607613b95c06c06015ae8 loaded}

    test "In the context of Lua the output of random commands gets ordered" {
+        r debug lua-always-replicate-commands 0
        r del myset
        r sadd myset a b c d e f g h i l m n o p q r s t u v z aa aaa azz
-        r eval {return redis.call('smembers',KEYS[1])} 1 myset
+        set res [r eval {return redis.call('smembers',KEYS[1])} 1 myset]
+        r debug lua-always-replicate-commands 1
+        set res
    } {a aa aaa azz b c d e f g h i l m n o p q r s t u v z}

    test "SORT is normally not alpha re-ordered for the scripting engine" {
@@ -655,11 +660,13 @@ start_server {tags {"scripting repl"}} {
        } {1}

        test "Redis.set_repl() must be issued after replicate_commands()" {
+            r debug lua-always-replicate-commands 0
            catch {
                r eval {
                    redis.set_repl(redis.REPL_ALL);
                } 0
            } e
+            r debug lua-always-replicate-commands 1
            set e
        } {*only after turning on*}

@@ -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
+    }
 }