don't do keys or scan async if in transaction

Check the validity of the value before performing the create operation,
prevents new data from being generated even if the request fails to execute.

Co-authored-by: Oran Agra <oran@redislabs.com>
Co-authored-by: chendianqiang <chendianqiang@meituan.com>
Co-authored-by: Binbin <binloveplay1314@qq.com>

src/Makefile

@@ -145,7 +145,9 @@ FINAL_LDFLAGS=$(LDFLAGS) $(KEYDB_LDFLAGS) $(DEBUG)
 FINAL_LIBS+=-lm -lz -lcrypto -lbz2 -lzstd -llz4 -lsnappy
 
 ifneq ($(uname_S),Darwin)
+    ifneq ($(uname_S),FreeBSD)
 	FINAL_LIBS+=-latomic
+    endif
 endif
 # Linux ARM32 needs -latomic at linking time
 ifneq (,$(findstring armv,$(uname_M)))
@@ -211,7 +213,13 @@ ifeq ($(uname_S),NetBSD)
 else
 ifeq ($(uname_S),FreeBSD)
 	# FreeBSD
-	FINAL_LIBS+= -lpthread -lexecinfo
+	FINAL_LIBS+= -lpthread -luuid -lexecinfo
+	FINAL_CFLAGS+= -I/usr/local/include
+	FINAL_CXXFLAGS+= -I/usr/local/include
+	FINAL_LDFLAGS+= -L/usr/local/lib
+	ifeq ($(USE_BACKTRACE),yes)
+	    FINAL_CFLAGS+= -DUSE_BACKTRACE
+	endif
 else
 ifeq ($(uname_S),DragonFly)
 	# DragonFly

src/StorageCache.cpp

@@ -21,7 +21,7 @@ dictType dbStorageCacheType = {
 StorageCache::StorageCache(IStorage *storage, bool fCache)
         : m_spstorage(storage)
 {
-    if (fCache)
+    if (!g_pserver->flash_disable_key_cache && fCache)
         m_pdict = dictCreate(&dbStorageCacheType, nullptr);
 }
 

src/config.cpp

@@ -2806,7 +2806,7 @@ standardConfig configs[] = {
     createBoolConfig("appendonly", NULL, MODIFIABLE_CONFIG, g_pserver->aof_enabled, 0, NULL, updateAppendonly),
     createBoolConfig("cluster-allow-reads-when-down", NULL, MODIFIABLE_CONFIG, g_pserver->cluster_allow_reads_when_down, 0, NULL, NULL),
     createBoolConfig("delete-on-evict", NULL, MODIFIABLE_CONFIG, cserver.delete_on_evict, 0, NULL, NULL),
-    createBoolConfig("use-fork", NULL, IMMUTABLE_CONFIG, cserver.fForkBgSave, 0, NULL, NULL),
+    createBoolConfig("use-fork", NULL, IMMUTABLE_CONFIG, cserver.fForkBgSave, 1, NULL, NULL),
     createBoolConfig("io-threads-do-reads", NULL, IMMUTABLE_CONFIG, fDummy, 0, NULL, NULL),
     createBoolConfig("time-thread-priority", NULL, IMMUTABLE_CONFIG, cserver.time_thread_priority, 0, NULL, NULL),
     createBoolConfig("prefetch-enabled", NULL, MODIFIABLE_CONFIG, g_pserver->prefetch_enabled, 1, NULL, NULL),
@@ -2912,6 +2912,7 @@ standardConfig configs[] = {
     createLongLongConfig("repl-backlog-size", NULL, MODIFIABLE_CONFIG, 1, LLONG_MAX, g_pserver->repl_backlog_size, 1024*1024, MEMORY_CONFIG, NULL, updateReplBacklogSize), /* Default: 1mb */
     createLongLongConfig("repl-backlog-disk-reserve", NULL, IMMUTABLE_CONFIG, 0, LLONG_MAX, cserver.repl_backlog_disk_size, 0, MEMORY_CONFIG, NULL, NULL),
     createLongLongConfig("max-snapshot-slip", NULL, MODIFIABLE_CONFIG, 0, 5000, g_pserver->snapshot_slip, 400, 0, NULL, NULL),
+    createLongLongConfig("max-rand-count", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX/2, g_pserver->rand_total_threshold, LONG_MAX/2, 0, NULL, NULL),
 
     /* Unsigned Long Long configs */
     createULongLongConfig("maxmemory", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, g_pserver->maxmemory, 0, MEMORY_CONFIG, NULL, updateMaxmemory),
@@ -2939,6 +2940,7 @@ standardConfig configs[] = {
     createBoolConfig("allow-write-during-load", NULL, MODIFIABLE_CONFIG, g_pserver->fWriteDuringActiveLoad, 0, NULL, NULL),
     createBoolConfig("force-backlog-disk-reserve", NULL, MODIFIABLE_CONFIG, cserver.force_backlog_disk, 0, NULL, NULL),
     createBoolConfig("soft-shutdown", NULL, MODIFIABLE_CONFIG, g_pserver->config_soft_shutdown, 0, NULL, NULL),
+    createBoolConfig("flash-disable-key-cache", NULL, MODIFIABLE_CONFIG, g_pserver->flash_disable_key_cache, 0, NULL, NULL),
     createSizeTConfig("semi-ordered-set-bucket-size", NULL, MODIFIABLE_CONFIG, 0, 1024, g_semiOrderedSetTargetBucketSize, 0, INTEGER_CONFIG, NULL, NULL),
 
 #ifdef USE_OPENSSL

src/db.cpp

@@ -1059,7 +1059,7 @@ void keysCommand(client *c) {
     sds pattern = szFromObj(c->argv[1]);
 
     const redisDbPersistentDataSnapshot *snapshot = nullptr;
-    if (!(c->flags & (CLIENT_MULTI | CLIENT_BLOCKED | CLIENT_DENY_BLOCKING)))
+    if (!(c->flags & (CLIENT_MULTI | CLIENT_BLOCKED | CLIENT_DENY_BLOCKING)) && !(serverTL->in_eval || serverTL->in_exec))
         snapshot = c->db->createSnapshot(c->mvccCheckpoint, true /* fOptional */);
     if (snapshot != nullptr)
     {
@@ -1224,7 +1224,7 @@ void scanGenericCommand(client *c, robj_roptr o, unsigned long cursor) {
         }
     }
 
-    if (o == nullptr && count >= 100)
+    if (o == nullptr && count >= 100 && !(serverTL->in_eval || serverTL->in_exec))
     {
         // Do an async version
         if (c->asyncCommand(
@@ -1737,6 +1737,9 @@ int dbSwapDatabases(int id1, int id2) {
         id2 < 0 || id2 >= cserver.dbnum) return C_ERR;
     if (id1 == id2) return C_OK;
     std::swap(g_pserver->db[id1], g_pserver->db[id2]);
+    
+    //swap db's id too, otherwise db does not match its id
+    std::swap(g_pserver->db[id1]->id, g_pserver->db[id2]->id);
 
     /* Note that we don't swap blocking_keys,
      * ready_keys and watched_keys, since we want clients to
@@ -1766,7 +1769,7 @@ int dbSwapDatabases(int id1, int id2) {
 
 /* SWAPDB db1 db2 */
 void swapdbCommand(client *c) {
-    int id1, id2;
+    int id1, id2, oriId;
 
     /* Not allowed in cluster mode: we have just DB 0 there. */
     if (g_pserver->cluster_enabled) {
@@ -1783,6 +1786,9 @@ void swapdbCommand(client *c) {
         "invalid second DB index") != C_OK)
         return;
 
+    //get client's original db's id
+    oriId=c->db->id;
+
     /* Swap... */
     if (dbSwapDatabases(id1,id2) == C_ERR) {
         addReplyError(c,"DB index is out of range");
@@ -1791,6 +1797,10 @@ void swapdbCommand(client *c) {
         RedisModuleSwapDbInfo si = {REDISMODULE_SWAPDBINFO_VERSION,(int32_t)id1,(int32_t)id2};
         moduleFireServerEvent(REDISMODULE_EVENT_SWAPDB,0,&si);
         g_pserver->dirty++;
+
+	//set client's db to original db
+        c->db=g_pserver->db[oriId];
+
         addReply(c,shared.ok);
     }
 }
@@ -1931,6 +1941,7 @@ void setExpire(client *c, redisDb *db, robj *key, expireEntry &&e)
  * is associated with this key (i.e. the key is non volatile) */
 expireEntry *redisDbPersistentDataSnapshot::getExpire(const char *key) {
     /* No expire? return ASAP */
+    std::unique_lock<fastlock> ul(g_expireLock);
     if (expireSize() == 0)
         return nullptr;
 
@@ -2727,8 +2738,11 @@ void redisDbPersistentData::clear(void(callback)(void*))
         m_cnewKeysPending = 0;
         m_fAllChanged++;
     }
+    {
+    std::unique_lock<fastlock> ul(g_expireLock);
     delete m_setexpire;
     m_setexpire = new (MALLOC_LOCAL) expireset();
+    }
     if (m_spstorage != nullptr)
         m_spstorage->clear(callback);
     dictEmpty(m_pdictTombstone,callback);
@@ -2861,7 +2875,8 @@ LNotFound:
             {
                 dictAdd(m_pdict, sdsNewKey, o);
                 o->SetFExpires(spexpire != nullptr);
-
+                
+                std::unique_lock<fastlock> ul(g_expireLock);
                 if (spexpire != nullptr)
                 {
                     auto itr = m_setexpire->find(sdsKey);
@@ -2884,6 +2899,7 @@ LNotFound:
     if (*pde != nullptr && dictGetVal(*pde) != nullptr)
     {
         robj *o = (robj*)dictGetVal(*pde);
+        std::unique_lock<fastlock> ul(g_expireLock);
         serverAssert(o->FExpires() == (m_setexpire->find(sdsKey) != m_setexpire->end()));
     }
 }
@@ -3256,6 +3272,7 @@ std::unique_ptr<expireEntry> deserializeExpire(sds key, const char *str, size_t
 
 sds serializeStoredObjectAndExpire(redisDbPersistentData *db, const char *key, robj_roptr o)
 {
+    std::unique_lock<fastlock> ul(g_expireLock);
     auto itrExpire = db->setexpire()->find(key);
     const expireEntry *pexpire = nullptr;
     if (itrExpire != db->setexpire()->end())
@@ -3380,6 +3397,7 @@ void redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
                     dictAdd(m_pdict, sharedKey, o);
                     o->SetFExpires(spexpire != nullptr);
 
+                    std::unique_lock<fastlock> ul(g_expireLock);
                     if (spexpire != nullptr)
                     {
                         auto itr = m_setexpire->find(sharedKey);

src/evict.cpp

@@ -380,7 +380,7 @@ size_t freeMemoryGetNotCountedMemory(void) {
 
     /* also don't count the replication backlog memory
      * that's where the replication clients get their memory from */
-    overhead += g_pserver->repl_backlog_size;
+    overhead += g_pserver->repl_backlog_size - g_pserver->repl_backlog_config_size;
 
     if (g_pserver->aof_state != AOF_OFF) {
         overhead += sdsalloc(g_pserver->aof_buf)+aofRewriteBufferSize();
@@ -552,7 +552,7 @@ static int evictionTimeProc(
     UNUSED(clientData);
     serverAssert(GlobalLocksAcquired());
 
-    if (performEvictions(false) == EVICT_RUNNING) return 0;  /* keep evicting */
+    if (performEvictions((bool)clientData) == EVICT_RUNNING) return 0;  /* keep evicting */
 
     /* For EVICT_OK - things are good, no need to keep evicting.
      * For EVICT_FAIL - there is nothing left to evict.  */
@@ -771,6 +771,7 @@ int performEvictions(bool fPreSnapshot) {
                 if (db->removeCachedValue(bestkey, &deT)) {
                     mem_freed += splazy->addEntry(db->dictUnsafeKeyOnly(), deT);
                     ckeysFailed = 0;
+		    g_pserver->stat_evictedkeys++;
                 }
                 else {
                     delta = 0;
@@ -837,7 +838,7 @@ int performEvictions(bool fPreSnapshot) {
                     if (!isEvictionProcRunning && serverTL->el != nullptr) {
                         isEvictionProcRunning = 1;
                         aeCreateTimeEvent(serverTL->el, 0,
-                                evictionTimeProc, NULL, NULL);
+                                evictionTimeProc, (void*)fPreSnapshot, NULL);
                     }
                     break;
                 }

src/expire.cpp

@@ -481,6 +481,7 @@ void expireSlaveKeys(void) {
     if (slaveKeysWithExpire == NULL ||
         dictSize(slaveKeysWithExpire) == 0) return;
 
+    std::unique_lock<fastlock> ul(g_expireLock);
     int cycles = 0, noexpire = 0;
     mstime_t start = mstime();
     while(1) {

src/module.cpp

@@ -656,7 +656,7 @@ void moduleHandlePropagationAfterCommandCallback(RedisModuleCtx *ctx) {
 
     /* We don't need to do anything here if the server isn't inside
      * a transaction. */
-    if (!g_pserver->propagate_in_transaction) return;
+    if (!serverTL->propagate_in_transaction) return;
 
     /* If this command is executed from with Lua or MULTI/EXEC we do not
      * need to propagate EXEC */
@@ -1814,7 +1814,7 @@ void moduleReplicateMultiIfNeeded(RedisModuleCtx *ctx) {
      * the module command was called by a script. */
     if (serverTL->in_eval || serverTL->in_exec) return;
     /* If we already emitted MULTI return ASAP. */
-    if (g_pserver->propagate_in_transaction) return;
+    if (serverTL->propagate_in_transaction) return;
     /* If this is a thread safe context, we do not want to wrap commands
      * executed into MULTI/EXEC, they are executed as single commands
      * from an external client in essence. */

src/multi.cpp

@@ -118,14 +118,14 @@ void discardCommand(client *c) {
 
 void beforePropagateMulti() {
     /* Propagating MULTI */
-    serverAssert(!g_pserver->propagate_in_transaction);
-    g_pserver->propagate_in_transaction = 1;
+    serverAssert(!serverTL->propagate_in_transaction);
+    serverTL->propagate_in_transaction = 1;
 }
 
 void afterPropagateExec() {
     /* Propagating EXEC */
-    serverAssert(g_pserver->propagate_in_transaction == 1);
-    g_pserver->propagate_in_transaction = 0;
+    serverAssert(serverTL->propagate_in_transaction == 1);
+    serverTL->propagate_in_transaction = 0;
 }
 
 /* Send a MULTI command to all the slaves and AOF file. Check the execCommand
@@ -264,7 +264,7 @@ void execCommand(client *c) {
 
     /* Make sure the EXEC command will be propagated as well if MULTI
      * was already propagated. */
-    if (g_pserver->propagate_in_transaction) {
+    if (serverTL->propagate_in_transaction) {
         int is_master = listLength(g_pserver->masters) == 0;
         g_pserver->dirty++;
         /* If inside the MULTI/EXEC block this instance was suddenly

src/networking.cpp

@@ -2753,7 +2753,7 @@ void readQueryFromClient(connection *conn) {
 
     if (cserver.cthreads > 1 || g_pserver->m_pstorageFactory) {
         parseClientCommandBuffer(c);
-        if (g_pserver->enable_async_commands && !serverTL->disable_async_commands && listLength(g_pserver->monitors) == 0 && (aeLockContention() || serverTL->rgdbSnapshot[c->db->id] || g_fTestMode)) {
+        if (g_pserver->enable_async_commands && !serverTL->disable_async_commands && listLength(g_pserver->monitors) == 0 && (aeLockContention() || serverTL->rgdbSnapshot[c->db->id] || g_fTestMode) && !serverTL->in_eval && !serverTL->in_exec) {
             // Frequent writers aren't good candidates for this optimization, they cause us to renew the snapshot too often
             //  so we exclude them unless the snapshot we need already exists.
             // Note: In test mode we want to create snapshots as often as possibl to excercise them - we don't care about perf
@@ -3995,7 +3995,7 @@ void pauseClients(mstime_t end, pause_type type) {
      * to track this state so that we don't assert
      * in propagate(). */
     if (serverTL->in_exec) {
-        g_pserver->client_pause_in_transaction = 1;
+        serverTL->client_pause_in_transaction = 1;
     }
 }
 

src/object.cpp

@@ -1668,6 +1668,7 @@ robj *deserializeStoredObjectCore(const void *data, size_t cb)
 robj *deserializeStoredObject(const redisDbPersistentData *db, const char *key, const void *data, size_t cb)
 {
     robj *o = deserializeStoredObjectCore(data, cb);
+    std::unique_lock<fastlock> ul(g_expireLock);
     o->SetFExpires(db->setexpire()->exists(key));
     return o;
 }

src/rdb.cpp

@@ -1334,7 +1334,7 @@ int rdbSaveRio(rio *rdb, const redisDbPersistentDataSnapshot **rgpdb, int *error
     if (rdbSaveModulesAux(rdb, REDISMODULE_AUX_BEFORE_RDB) == -1) goto werr;
 
     for (j = 0; j < cserver.dbnum; j++) {
-        const redisDbPersistentDataSnapshot *db = rgpdb[j];
+        const redisDbPersistentDataSnapshot *db = rgpdb != nullptr ? rgpdb[j] : g_pserver->db[j];
         if (db->size() == 0) continue;
 
         /* Write the SELECT DB opcode */
@@ -1715,7 +1715,7 @@ void getTempFileName(char tmpfile[], int tmpfileNum) {
     char tmpfileNumString[214];
 
     /* Generate temp rdb file name using aync-signal safe functions. */
-    int pid_len = ll2string(pid, sizeof(pid), getpid());
+    int pid_len = ll2string(pid, sizeof(pid), g_pserver->in_fork_child ? getppid() : getpid());
     int tmpfileNum_len = ll2string(tmpfileNumString, sizeof(tmpfileNumString), tmpfileNum);
     strcpy(tmpfile, "temp-");
     strncpy(tmpfile+5, pid, pid_len);
@@ -3832,57 +3832,130 @@ int rdbSaveToSlavesSockets(rdbSaveInfo *rsi) {
     }
 
     /* Create the child process. */
-    openChildInfoPipe();
+    if (cserver.fForkBgSave) {
+        pid_t childpid;
+        if ((childpid = redisFork(CHILD_TYPE_RDB)) == 0) {
+            /* Child */
+            int retval, dummy;
+            rio rdb;
 
-    for (int idb = 0; idb < cserver.dbnum; ++idb)
-        args->rgpdb[idb] = g_pserver->db[idb]->createSnapshot(getMvccTstamp(), false /*fOptional*/);
+            rioInitWithFd(&rdb,args->rdb_pipe_write);
 
-    g_pserver->rdbThreadVars.tmpfileNum++;
-    g_pserver->rdbThreadVars.fRdbThreadCancel = false;
-    pthread_attr_t tattr;
-    pthread_attr_init(&tattr);
-    pthread_attr_setstacksize(&tattr, 1 << 23); // 8 MB
-    if (pthread_create(&child, &tattr, rdbSaveToSlavesSocketsThread, args)) {
-        pthread_attr_destroy(&tattr);
-        serverLog(LL_WARNING,"Can't save in background: fork: %s",
-            strerror(errno));
+            redisSetProcTitle("keydb-rdb-to-slaves");
+            redisSetCpuAffinity(g_pserver->bgsave_cpulist);
 
-        /* Undo the state change. The caller will perform cleanup on
-            * all the slaves in BGSAVE_START state, but an early call to
-            * replicationSetupSlaveForFullResync() turned it into BGSAVE_END */
-        listRewind(g_pserver->slaves,&li);
-        while((ln = listNext(&li))) {
-            client *replica = (client*)ln->value;
-            if (replica->replstate == SLAVE_STATE_WAIT_BGSAVE_END) {
-                replica->replstate = SLAVE_STATE_WAIT_BGSAVE_START;
+            retval = rdbSaveRioWithEOFMark(&rdb,nullptr,nullptr,rsi);
+            if (retval == C_OK && rioFlush(&rdb) == 0)
+                retval = C_ERR;
+
+            if (retval == C_OK) {
+                sendChildCowInfo(CHILD_INFO_TYPE_RDB_COW_SIZE, "RDB");
             }
-        }
-        close(args->rdb_pipe_write);
-        close(g_pserver->rdb_pipe_read);
-        zfree(g_pserver->rdb_pipe_conns);
-        close(args->safe_to_exit_pipe);
-        g_pserver->rdb_pipe_conns = NULL;
-        g_pserver->rdb_pipe_numconns = 0;
-        g_pserver->rdb_pipe_numconns_writing = 0;
-        args->rsi.~rdbSaveInfo();
-        zfree(args);
-        closeChildInfoPipe();
-        return C_ERR;
-    }
-    pthread_attr_destroy(&tattr);
-    g_pserver->child_type = CHILD_TYPE_RDB;
 
-    serverLog(LL_NOTICE,"Background RDB transfer started");
-    g_pserver->rdb_save_time_start = time(NULL);
-    serverAssert(!g_pserver->rdbThreadVars.fRdbThreadActive);
-    g_pserver->rdbThreadVars.rdb_child_thread = child;
-    g_pserver->rdbThreadVars.fRdbThreadActive = true;
-    g_pserver->rdb_child_type = RDB_CHILD_TYPE_SOCKET;
-    aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, []{
-        if (aeCreateFileEvent(serverTL->el, g_pserver->rdb_pipe_read, AE_READABLE, rdbPipeReadHandler,NULL) == AE_ERR) {
-            serverPanic("Unrecoverable error creating server.rdb_pipe_read file event.");
+            rioFreeFd(&rdb);
+            /* wake up the reader, tell it we're done. */
+            close(args->rdb_pipe_write);
+            close(g_pserver->rdb_child_exit_pipe); /* close write end so that we can detect the close on the parent. */
+            /* hold exit until the parent tells us it's safe. we're not expecting
+            * to read anything, just get the error when the pipe is closed. */
+            dummy = read(args->safe_to_exit_pipe, pipefds, 1);
+            UNUSED(dummy);
+            exitFromChild((retval == C_OK) ? 0 : 1);
+        } else {
+            /* Parent */
+            close(args->safe_to_exit_pipe);
+            if (childpid == -1) {
+                serverLog(LL_WARNING,"Can't save in background: fork: %s",
+                    strerror(errno));
+
+                /* Undo the state change. The caller will perform cleanup on
+                * all the slaves in BGSAVE_START state, but an early call to
+                * replicationSetupSlaveForFullResync() turned it into BGSAVE_END */
+                listRewind(g_pserver->slaves,&li);
+                while((ln = listNext(&li))) {
+                    client *replica = (client*)ln->value;
+                    if (replica->replstate == SLAVE_STATE_WAIT_BGSAVE_END) {
+                        replica->replstate = SLAVE_STATE_WAIT_BGSAVE_START;
+                    }
+                }
+                close(args->rdb_pipe_write);
+                close(g_pserver->rdb_pipe_read);
+                zfree(g_pserver->rdb_pipe_conns);
+                g_pserver->rdb_pipe_conns = NULL;
+                g_pserver->rdb_pipe_numconns = 0;
+                g_pserver->rdb_pipe_numconns_writing = 0;
+                args->rsi.~rdbSaveInfo();
+                zfree(args);
+            } else {
+                serverLog(LL_NOTICE,"Background RDB transfer started by pid %ld",
+                    (long)childpid);
+                g_pserver->rdb_save_time_start = time(NULL);
+                g_pserver->rdb_child_type = RDB_CHILD_TYPE_SOCKET;
+                g_pserver->rdbThreadVars.fRdbThreadActive = true;
+                updateDictResizePolicy();
+                close(args->rdb_pipe_write); /* close write in parent so that it can detect the close on the child. */
+                aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, []{
+                    if (aeCreateFileEvent(serverTL->el, g_pserver->rdb_pipe_read, AE_READABLE, rdbPipeReadHandler, nullptr) == AE_ERR) {
+                        serverPanic("Unrecoverable error creating g_pserver->rdb_pipe_read file event.");
+                    }
+                });
+            }
+            return (childpid == -1) ? C_ERR : C_OK;
         }
-    });
+    }
+    else {
+        openChildInfoPipe();
+
+        for (int idb = 0; idb < cserver.dbnum; ++idb)
+            args->rgpdb[idb] = g_pserver->db[idb]->createSnapshot(getMvccTstamp(), false /*fOptional*/);
+
+        g_pserver->rdbThreadVars.tmpfileNum++;
+        g_pserver->rdbThreadVars.fRdbThreadCancel = false;
+        pthread_attr_t tattr;
+        pthread_attr_init(&tattr);
+        pthread_attr_setstacksize(&tattr, 1 << 23); // 8 MB
+        if (pthread_create(&child, &tattr, rdbSaveToSlavesSocketsThread, args)) {
+            pthread_attr_destroy(&tattr);
+            serverLog(LL_WARNING,"Can't save in background: fork: %s",
+                strerror(errno));
+
+            /* Undo the state change. The caller will perform cleanup on
+                * all the slaves in BGSAVE_START state, but an early call to
+                * replicationSetupSlaveForFullResync() turned it into BGSAVE_END */
+            listRewind(g_pserver->slaves,&li);
+            while((ln = listNext(&li))) {
+                client *replica = (client*)ln->value;
+                if (replica->replstate == SLAVE_STATE_WAIT_BGSAVE_END) {
+                    replica->replstate = SLAVE_STATE_WAIT_BGSAVE_START;
+                }
+            }
+            close(args->rdb_pipe_write);
+            close(g_pserver->rdb_pipe_read);
+            zfree(g_pserver->rdb_pipe_conns);
+            close(args->safe_to_exit_pipe);
+            g_pserver->rdb_pipe_conns = NULL;
+            g_pserver->rdb_pipe_numconns = 0;
+            g_pserver->rdb_pipe_numconns_writing = 0;
+            args->rsi.~rdbSaveInfo();
+            zfree(args);
+            closeChildInfoPipe();
+            return C_ERR;
+        }
+        pthread_attr_destroy(&tattr);
+        g_pserver->child_type = CHILD_TYPE_RDB;
+
+        serverLog(LL_NOTICE,"Background RDB transfer started");
+        g_pserver->rdb_save_time_start = time(NULL);
+        serverAssert(!g_pserver->rdbThreadVars.fRdbThreadActive);
+        g_pserver->rdbThreadVars.rdb_child_thread = child;
+        g_pserver->rdbThreadVars.fRdbThreadActive = true;
+        g_pserver->rdb_child_type = RDB_CHILD_TYPE_SOCKET;
+        aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, []{
+            if (aeCreateFileEvent(serverTL->el, g_pserver->rdb_pipe_read, AE_READABLE, rdbPipeReadHandler, nullptr) == AE_ERR) {
+                serverPanic("Unrecoverable error creating server.rdb_pipe_read file event.");
+            }
+        });
+    }
 
     return C_OK; /* Unreached. */
 }

src/server.cpp

@@ -3945,8 +3945,6 @@ void initServer(void) {
     g_pserver->pubsub_channels = dictCreate(&keylistDictType,NULL);
     g_pserver->pubsub_patterns = dictCreate(&keylistDictType,NULL);
     g_pserver->cronloops = 0;
-    g_pserver->propagate_in_transaction = 0;
-    g_pserver->client_pause_in_transaction = 0;
     g_pserver->child_pid = -1;
     g_pserver->child_type = CHILD_TYPE_NONE;
     g_pserver->rdbThreadVars.fRdbThreadCancel = false;
@@ -4320,12 +4318,12 @@ void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc,
      * This way we'll deliver the MULTI/..../EXEC block as a whole and
      * both the AOF and the replication link will have the same consistency
      * and atomicity guarantees. */
-    if (serverTL->in_exec && !g_pserver->propagate_in_transaction)
+    if (serverTL->in_exec && !serverTL->propagate_in_transaction)
         execCommandPropagateMulti(dbid);
 
     /* This needs to be unreachable since the dataset should be fixed during 
      * client pause, otherwise data may be lossed during a failover. */
-    serverAssert(!(areClientsPaused() && !g_pserver->client_pause_in_transaction));
+    serverAssert(!(areClientsPaused() && !serverTL->client_pause_in_transaction));
 
     if (g_pserver->aof_state != AOF_OFF && flags & PROPAGATE_AOF)
         feedAppendOnlyFile(cmd,dbid,argv,argc);
@@ -4647,8 +4645,8 @@ void call(client *c, int flags) {
 
     /* Client pause takes effect after a transaction has finished. This needs
      * to be located after everything is propagated. */
-    if (!serverTL->in_exec && g_pserver->client_pause_in_transaction) {
-        g_pserver->client_pause_in_transaction = 0;
+    if (!serverTL->in_exec && serverTL->client_pause_in_transaction) {
+        serverTL->client_pause_in_transaction = 0;
     }
 
     /* If the client has keys tracking enabled for client side caching,
@@ -4739,8 +4737,9 @@ int processCommand(client *c, int callFlags) {
         /* Both EXEC and EVAL call call() directly so there should be
          * no way in_exec or in_eval or propagate_in_transaction is 1.
          * That is unless lua_timedout, in which case client may run
-         * some commands. */
-        serverAssert(!g_pserver->propagate_in_transaction);
+         * some commands. Also possible that some other thread set
+         * propagate_in_transaction if this is an async command. */
+        serverAssert(!serverTL->propagate_in_transaction);
         serverAssert(!serverTL->in_exec);
         serverAssert(!serverTL->in_eval);
     }
@@ -5172,7 +5171,7 @@ int prepareForShutdown(int flags) {
          * to unlink file actully) in background thread.
          * The temp rdb file fd may won't be closed when redis exits quickly,
          * but OS will close this fd when process exits. */
-        rdbRemoveTempFile(g_pserver->child_pid, 0);
+        rdbRemoveTempFile(g_pserver->rdbThreadVars.tmpfileNum, 0);
     }
 
     /* Kill module child if there is one. */

src/server.h

@@ -2197,6 +2197,9 @@ struct redisServerThreadVars {
     bool modulesEnabledThisAeLoop = false; /* In this loop of aeMain, were modules enabled before 
                                               the thread went to sleep? */
     bool disable_async_commands = false; /* this is only valid for one cycle of the AE loop and is reset in afterSleep */
+    
+    int propagate_in_transaction = 0;  /* Make sure we don't propagate nested MULTI/EXEC */
+    int client_pause_in_transaction = 0; /* Was a client pause executed during this Exec? */
     std::vector<client*> vecclientsProcess;
     dictAsyncRehashCtl *rehashCtl = nullptr;
 
@@ -2302,9 +2305,7 @@ struct redisServer {
     int sentinel_mode;          /* True if this instance is a Sentinel. */
     size_t initial_memory_usage; /* Bytes used after initialization. */
     int always_show_logo;       /* Show logo even for non-stdout logging. */
-    int propagate_in_transaction;  /* Make sure we don't propagate nested MULTI/EXEC */
     char *ignore_warnings;      /* Config: warnings that should be ignored. */
-    int client_pause_in_transaction; /* Was a client pause executed during this Exec? */
     pause_type client_pause_type;      /* True if clients are currently paused */
     /* Modules */
     ::dict *moduleapi;            /* Exported core APIs dictionary for modules. */
@@ -2729,9 +2730,13 @@ struct redisServer {
     long long repl_batch_offStart = -1;
     long long repl_batch_idxStart = -1;
 
+    long long rand_total_threshold;
+
     int config_soft_shutdown = false;
     bool soft_shutdown = false;
 
+    int flash_disable_key_cache = false;
+
     /* Lock Contention Ring Buffer */
     static const size_t s_lockContentionSamples = 64;
     uint16_t rglockSamples[s_lockContentionSamples];

src/t_hash.cpp

@@ -729,6 +729,10 @@ void hincrbyfloatCommand(client *c) {
     unsigned int vlen;
 
     if (getLongDoubleFromObjectOrReply(c,c->argv[3],&incr,NULL) != C_OK) return;
+    if (isnan(incr) || isinf(incr)) {
+        addReplyError(c,"value is NaN or Infinity");
+        return;
+    }
     if ((o = hashTypeLookupWriteOrCreate(c,c->argv[1])) == NULL) return;
     if (hashTypeGetValue(o,szFromObj(c->argv[2]),&vstr,&vlen,&ll) == C_OK) {
         if (vstr) {
@@ -1062,6 +1066,8 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
                 addReplyBulkCBuffer(c, key, sdslen(key));
                 if (withvalues)
                     addReplyBulkCBuffer(c, value, sdslen(value));
+                if (c->flags & CLIENT_CLOSE_ASAP)
+                    break;
             }
         } else if (hash->encoding == OBJ_ENCODING_ZIPLIST) {
             ziplistEntry *keys, *vals = NULL;
@@ -1075,6 +1081,8 @@ void hrandfieldWithCountCommand(client *c, long l, int withvalues) {
                 count -= sample_count;
                 ziplistRandomPairs((unsigned char*)ptrFromObj(hash), sample_count, keys, vals);
                 harndfieldReplyWithZiplist(c, sample_count, keys, vals);
+                if (c->flags & CLIENT_CLOSE_ASAP)
+                    break;
             }
             zfree(keys);
             zfree(vals);
@@ -1221,13 +1229,13 @@ void hrandfieldCommand(client *c) {
     ziplistEntry ele;
 
     if (c->argc >= 3) {
-        if (getLongFromObjectOrReply(c,c->argv[2],&l,NULL) != C_OK) return;
+        if (getRangeLongFromObjectOrReply(c,c->argv[2],-LONG_MAX,LONG_MAX,&l,NULL) != C_OK) return;
         if (c->argc > 4 || (c->argc == 4 && strcasecmp(szFromObj(c->argv[3]),"withvalues"))) {
             addReplyErrorObject(c,shared.syntaxerr);
             return;
         } else if (c->argc == 4) {
             withvalues = 1;
-            if (l < LONG_MIN/2 || l > LONG_MAX/2) {
+            if (l < -g_pserver->rand_total_threshold || l > g_pserver->rand_total_threshold) {
                 addReplyError(c,"value is out of range");
                 return;
             }

src/t_set.cpp

@@ -672,7 +672,11 @@ void srandmemberWithCountCommand(client *c) {
 
     dict *d;
 
-    if (getLongFromObjectOrReply(c,c->argv[2],&l,NULL) != C_OK) return;
+    if (getRangeLongFromObjectOrReply(c,c->argv[2],-LONG_MAX,LONG_MAX,&l,NULL) != C_OK) return;
+    if (l < -g_pserver->rand_total_threshold || l > g_pserver->rand_total_threshold) {
+        addReplyError(c,"value is out of range");
+        return;
+    }
     if (l >= 0) {
         count = (unsigned long) l;
     } else {
@@ -706,6 +710,8 @@ void srandmemberWithCountCommand(client *c) {
             } else {
                 addReplyBulkCBuffer(c,ele,sdslen(ele));
             }
+            if (c->flags & CLIENT_CLOSE_ASAP)
+                break;
         }
         return;
     }

src/t_zset.cpp

@@ -4054,6 +4054,8 @@ void zrandmemberWithCountCommand(client *c, long l, int withscores) {
                 addReplyBulkCBuffer(c, key, sdslen(key));
                 if (withscores)
                     addReplyDouble(c, *(double*)dictGetVal(de));
+                if (c->flags & CLIENT_CLOSE_ASAP)
+                    break;
             }
         } else if (zsetobj->encoding == OBJ_ENCODING_ZIPLIST) {
             ziplistEntry *keys, *vals = NULL;
@@ -4067,6 +4069,8 @@ void zrandmemberWithCountCommand(client *c, long l, int withscores) {
                 count -= sample_count;
                 ziplistRandomPairs((unsigned char*)ptrFromObj(zsetobj), sample_count, keys, vals);
                 zarndmemberReplyWithZiplist(c, sample_count, keys, vals);
+                if (c->flags & CLIENT_CLOSE_ASAP)
+                    break;
             }
             zfree(keys);
             zfree(vals);
@@ -4210,13 +4214,13 @@ void zrandmemberCommand(client *c) {
     ziplistEntry ele;
 
     if (c->argc >= 3) {
-        if (getLongFromObjectOrReply(c,c->argv[2],&l,NULL) != C_OK) return;
+        if (getRangeLongFromObjectOrReply(c,c->argv[2],-LONG_MAX,LONG_MAX,&l,NULL) != C_OK) return;
         if (c->argc > 4 || (c->argc == 4 && strcasecmp(szFromObj(c->argv[3]),"withscores"))) {
             addReplyErrorObject(c,shared.syntaxerr);
             return;
         } else if (c->argc == 4) {
             withscores = 1;
-            if (l < LONG_MIN/2 || l > LONG_MAX/2) {
+            if (l < -g_pserver->rand_total_threshold || l > g_pserver->rand_total_threshold) {
                 addReplyError(c,"value is out of range");
                 return;
             }

tests/unit/obuf-limits.tcl

@@ -182,4 +182,12 @@ start_server {tags {"obuf-limits"} overrides { server-threads 1 }} {
         assert_equal "v2" [r get k2]
         assert_equal "v3" [r get k3]
     }
+
+    test "Obuf limit, HRANDFIELD with huge count stopped mid-run" {
+        r config set client-output-buffer-limit {normal 1000000 0 0}
+        r hset myhash a b
+        catch {r hrandfield myhash -999999999} e
+        assert_match "*I/O error*" $e
+        reconnect
+    }
 }

tests/unit/type/hash.tcl

@@ -71,6 +71,8 @@ start_server {tags {"hash"}} {
     test "HRANDFIELD count overflow" {
         r hmset myhash a 1
         assert_error {*value is out of range*} {r hrandfield myhash -9223372036854770000 withvalues}
+        assert_error {*value is out of range*} {r hrandfield myhash -9223372036854775808 withvalues}
+        assert_error {*value is out of range*} {r hrandfield myhash -9223372036854775808}
     } {}
 
     test "HRANDFIELD with <count> against non existing key" {
@@ -831,4 +833,8 @@ start_server {tags {"hash"}} {
         set _ $k
     } {ZIP_INT_8B 127 ZIP_INT_16B 32767 ZIP_INT_32B 2147483647 ZIP_INT_64B 9223372036854775808 ZIP_INT_IMM_MIN 0 ZIP_INT_IMM_MAX 12}
 
+    test {HINCRBYFLOAT does not allow NaN or Infinity} {
+        assert_error "*value is NaN or Infinity*" {r hincrbyfloat hfoo field +inf}
+        assert_equal 0 [r exists hfoo]
+    }
 }

tests/unit/type/set.tcl

@@ -588,6 +588,11 @@ start_server {
         r srandmember nonexisting_key 100
     } {}
 
+    test "SRANDMEMBER count overflow" {
+        r sadd myset a
+        assert_error {*value is out of range*} {r srandmember myset -9223372036854775808}
+    } {}
+
     # Make sure we can distinguish between an empty array and a null response
     r readraw 1
 

tests/unit/type/zset.tcl

@@ -1717,6 +1717,8 @@ start_server {tags {"zset"}} {
     test "ZRANDMEMBER count overflow" {
         r zadd myzset 0 a
         assert_error {*value is out of range*} {r zrandmember myzset -9223372036854770000 withscores}
+        assert_error {*value is out of range*} {r zrandmember myzset -9223372036854775808 withscores}
+        assert_error {*value is out of range*} {r zrandmember myzset -9223372036854775808}
     } {}
 
     # Make sure we can distinguish between an empty array and a null response

utils/create-cluster/create-cluster

@@ -26,7 +26,11 @@ then
     while [ $((PORT < ENDPORT)) != "0" ]; do
         PORT=$((PORT+1))
         echo "Starting $PORT"
-        $BIN_PATH/keydb-server --port $PORT  --protected-mode $PROTECTED_MODE --cluster-enabled yes --cluster-config-file nodes-${PORT}.conf --cluster-node-timeout $TIMEOUT --appendonly yes --appendfilename appendonly-${PORT}.aof --dbfilename dump-${PORT}.rdb --logfile ${PORT}.log --daemonize yes ${ADDITIONAL_OPTIONS}
+        if [ "$2" == "flash" ]
+        then
+            ADDITIONAL_OPTIONS="--save \"\" \"\" \"\" --semi-ordered-set-bucket-size 8 --client-output-buffer-limit replica 1 1 0 --maxmemory 100000000 --storage-provider flash ./$PORT.flash"
+        fi
+        $BIN_PATH/keydb-server --server-threads 4 --port $PORT  --protected-mode $PROTECTED_MODE --cluster-enabled yes --cluster-config-file nodes-${PORT}.conf --cluster-node-timeout $TIMEOUT --dbfilename dump-${PORT}.rdb --logfile ${PORT}.log --daemonize yes ${ADDITIONAL_OPTIONS}
     done
     exit 0
 fi
@@ -97,6 +101,8 @@ then
     rm -rf appendonly*.aof
     rm -rf dump*.rdb
     rm -rf nodes*.conf
+    rm -rf *.flash
+    rm -rf temp*.rdb
     exit 0
 fi