fix indent in evict.cpp

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/IStorage.h

@@ -31,13 +31,13 @@ public:
     virtual void retrieve(const char *key, size_t cchKey, callbackSingle fn) const = 0;
     virtual size_t clear() = 0;
     virtual bool enumerate(callback fn) const = 0;
+    virtual size_t stateful_enumerate(callback fn) = 0;
     virtual size_t count() const = 0;
 
-    virtual void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem) {
+    virtual void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem) {
         beginWriteBatch();
         for (size_t ielem = 0; ielem < celem; ++ielem) {
-            bool fOverwrite = (rgfOverwrite != nullptr) ? rgfOverwrite[ielem] : false;
-            insert(rgkeys[ielem], rgcbkeys[ielem], rgvals[ielem], rgcbvals[ielem], fOverwrite);
+            insert(rgkeys[ielem], rgcbkeys[ielem], rgvals[ielem], rgcbvals[ielem], false);
         }
         endWriteBatch();
     }

src/StorageCache.cpp

@@ -114,14 +114,13 @@ void StorageCache::insert(sds key, const void *data, size_t cbdata, bool fOverwr
 }
 
 long _dictKeyIndex(dict *d, const void *key, uint64_t hash, dictEntry **existing);
-void StorageCache::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem)
+void StorageCache::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem)
 {
     std::vector<dictEntry*> vechashes;
     if (m_pdict != nullptr) {
         vechashes.reserve(celem);
 
         for (size_t ielem = 0; ielem < celem; ++ielem) {
-            if (rgfOverwrite != nullptr && rgfOverwrite[ielem]) continue;
             dictEntry *de = (dictEntry*)zmalloc(sizeof(dictEntry));
             de->key = (void*)dictGenHashFunction(rgkeys[ielem], (int)rgcbkeys[ielem]);
             de->v.u64 = 1;
@@ -153,7 +152,7 @@ void StorageCache::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, si
     }
     ul.unlock();
 
-    m_spstorage->bulkInsert(rgkeys, rgcbkeys, rgvals, rgcbvals, rgfOverwrite, celem);
+    m_spstorage->bulkInsert(rgkeys, rgcbkeys, rgvals, rgcbvals, celem);
 
     bulkInsertsInProgress--;
 }

src/StorageCache.h

@@ -41,7 +41,7 @@ public:
     void clear(void(callback)(void*));
     void clearAsync();
     void insert(sds key, const void *data, size_t cbdata, bool fOverwrite);
-    void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem);
+    void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem);
     void retrieve(sds key, IStorage::callbackSingle fn) const;
     bool erase(sds key);
     void emergencyFreeCache();
@@ -49,6 +49,7 @@ public:
     void expand(uint64_t slots);
 
     bool enumerate(IStorage::callback fn) const { return m_spstorage->enumerate(fn); }
+    size_t stateful_enumerate(IStorage::callback fn) { return m_spstorage->stateful_enumerate(fn); }
 
     void beginWriteBatch();
     void endWriteBatch() { m_spstorage->endWriteBatch(); }

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),

src/db.cpp

@@ -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()));
     }
 }
@@ -2908,13 +2924,15 @@ void redisDbPersistentData::storeDatabase()
     dictReleaseIterator(di);
 }
 
-/* static */ sds redisDbPersistentData::serializeChange(redisDbPersistentData *db, const char *key)
+/* static */ void redisDbPersistentData::serializeAndStoreChange(StorageCache *storage, redisDbPersistentData *db, const char *key, bool fUpdate)
 {
     auto itr = db->find_cached_threadsafe(key);
     if (itr == nullptr)
-        return nullptr;
+        return;
     robj *o = itr.val();
-    return serializeStoredObjectAndExpire(db, (const char*) itr.key(), o);
+    sds temp = serializeStoredObjectAndExpire(db, (const char*) itr.key(), o);
+    storage->insert((sds)key, temp, sdslen(temp), fUpdate);
+    sdsfree(temp);
 }
 
 bool redisDbPersistentData::processChanges(bool fSnapshot)
@@ -2957,30 +2975,10 @@ bool redisDbPersistentData::processChanges(bool fSnapshot)
             {
                 dictIterator *di = dictGetIterator(m_dictChanged);
                 dictEntry *de;
-                std::vector<char*> veckeys;
-                std::vector<size_t> veccbkeys;
-                std::vector<char*> vecvals;
-                std::vector<size_t> veccbvals;
-                std::vector<char> vecoverwrite;
-                veckeys.reserve(dictSize(m_dictChanged));
-                veccbkeys.reserve(dictSize(m_dictChanged));
-                vecvals.reserve(dictSize(m_dictChanged));
-                veccbvals.reserve(dictSize(m_dictChanged));
-                vecoverwrite.reserve(dictSize(m_dictChanged));
                 while ((de = dictNext(di)) != nullptr)
                 {
-                    sds val = serializeChange(this, (const char*)dictGetKey(de));
-                    if (val != nullptr) {
-                        veckeys.push_back((char*)dictGetKey(de));
-                        veccbkeys.push_back(sdslen((sds)dictGetKey(de)));
-                        vecvals.push_back(val);
-                        veccbvals.push_back(sdslen(val));
-                        vecoverwrite.push_back((bool)dictGetVal(de));
-                    }
+                    serializeAndStoreChange(m_spstorage.get(), this, (const char*)dictGetKey(de), (bool)dictGetVal(de));
                 }
-                m_spstorage->bulkInsert(veckeys.data(), veccbkeys.data(), vecvals.data(), veccbvals.data(), vecoverwrite.data(), veckeys.size());
-                for (auto val : vecvals)
-                    sdsfree(val);
                 dictReleaseIterator(di);
             }
         }
@@ -3014,7 +3012,7 @@ void redisDbPersistentData::processChangesAsync(std::atomic<int> &pendingJobs)
             vecvals.push_back(temp);
             veccbvals.push_back(sdslen(temp));
         }
-        m_spstorage->bulkInsert(veckeys.data(), veccbkeys.data(), vecvals.data(), veccbvals.data(), nullptr, veckeys.size());
+        m_spstorage->bulkInsert(veckeys.data(), veccbkeys.data(), vecvals.data(), veccbvals.data(), veckeys.size());
         for (auto val : vecvals)
             sdsfree(val);
         dictReleaseIterator(di);
@@ -3033,7 +3031,7 @@ void redisDbPersistentData::bulkDirectStorageInsert(char **rgKeys, size_t *rgcbK
         }
         aeReleaseLock();
     }
-    m_spstorage->bulkInsert(rgKeys, rgcbKeys, rgVals, rgcbVals, nullptr, celem);
+    m_spstorage->bulkInsert(rgKeys, rgcbKeys, rgVals, rgcbVals, celem);
 }
 
 void redisDbPersistentData::commitChanges(const redisDbPersistentDataSnapshot **psnapshotFree)
@@ -3042,30 +3040,10 @@ void redisDbPersistentData::commitChanges(const redisDbPersistentDataSnapshot **
     {
         dictIterator *di = dictGetIterator(m_dictChangedStorageFlush);
         dictEntry *de;
-        std::vector<char*> veckeys;
-        std::vector<size_t> veccbkeys;
-        std::vector<char*> vecvals;
-        std::vector<size_t> veccbvals;
-        std::vector<char> vecoverwrite;
-        veckeys.reserve(dictSize(m_dictChanged));
-        veccbkeys.reserve(dictSize(m_dictChanged));
-        vecvals.reserve(dictSize(m_dictChanged));
-        veccbvals.reserve(dictSize(m_dictChanged));
-        vecoverwrite.resize(dictSize(m_dictChanged));
         while ((de = dictNext(di)) != nullptr)
         {
-            sds val = serializeChange((redisDbPersistentData*)m_pdbSnapshotStorageFlush, (const char*)dictGetKey(de));
-            if (val != nullptr) {
-                veckeys.push_back((char*)dictGetKey(de));
-                veccbkeys.push_back(sdslen((sds)dictGetKey(de)));
-                vecvals.push_back(val);
-                veccbvals.push_back(sdslen(val));
-                vecoverwrite.push_back((bool)dictGetVal(de));
-            }
+            serializeAndStoreChange(m_spstorage.get(), (redisDbPersistentData*)m_pdbSnapshotStorageFlush, (const char*)dictGetKey(de), (bool)dictGetVal(de));
         }
-        m_spstorage->bulkInsert(veckeys.data(), veccbkeys.data(), vecvals.data(), veccbvals.data(), vecoverwrite.data(), veckeys.size());
-        for (auto val : vecvals)
-            sdsfree(val);
         dictReleaseIterator(di);
         dictRelease(m_dictChangedStorageFlush);
         m_dictChangedStorageFlush = nullptr;
@@ -3294,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())
@@ -3418,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();
@@ -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;

src/expire.cpp

@@ -416,6 +416,33 @@ void activeExpireCycleCore(int type) {
             return true;
         }, &check);
 
+        if (!timelimit_exit && db->getStorageCache() != nullptr)
+            db->getStorageCache()->stateful_enumerate([&] (const char *key, size_t keylen, const void *val, size_t vallen) {
+                sds sdsKey = sdsnewlen(key, keylen);
+                size_t offset = 0;
+                auto e = deserializeExpire(sdsKey, (const char*)val, vallen, &offset);
+                if (e->when() < now)
+                {
+                    expired += activeExpireCycleExpire(db, *e, now, tried);
+                }
+                sdsfree(sdsKey);
+
+                if ((tried % ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP) == 0)
+                {
+                    /* We can't block forever here even if there are many keys to
+                    * expire. So after a given amount of milliseconds return to the
+                    * caller waiting for the other active expire cycle. */
+                    elapsed = ustime()-start;
+                    if (elapsed > timelimit) {
+                        timelimit_exit = 1;
+                        g_pserver->stat_expired_time_cap_reached_count++;
+                        return false;
+                    }
+                    check = ACTIVE_EXPIRE_CYCLE_FAST_DURATION;
+                }
+                return true;
+            });
+
         total_expired += expired;
     }
 
@@ -481,6 +508,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/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

@@ -5172,7 +5172,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

@@ -1209,6 +1209,7 @@ public:
     bool FSnapshot() const { return m_spdbSnapshotHOLDER != nullptr; }
 
     std::unique_ptr<const StorageCache> CloneStorageCache() { return std::unique_ptr<const StorageCache>(m_spstorage->clone()); }
+    std::shared_ptr<StorageCache> getStorageCache() { return m_spstorage; }
     void bulkDirectStorageInsert(char **rgKeys, size_t *rgcbKeys, char **rgVals, size_t *rgcbVals, size_t celem);
 
     dict_iter find_cached_threadsafe(const char *key) const;
@@ -1217,7 +1218,7 @@ protected:
     uint64_t m_mvccCheckpoint = 0;
 
 private:
-    static sds serializeChange(redisDbPersistentData *db, const char *key);
+    static void serializeAndStoreChange(StorageCache *storage, redisDbPersistentData *db, const char *key, bool fUpdate);
 
     void ensure(const char *key);
     void ensure(const char *key, dictEntry **de);
@@ -1371,6 +1372,7 @@ struct redisDb : public redisDbPersistentDataSnapshot
     using redisDbPersistentData::FTrackingChanges;
     using redisDbPersistentData::CloneStorageCache;
     using redisDbPersistentData::bulkDirectStorageInsert;
+    using redisDbPersistentData::getStorageCache;
 
 public:
     const redisDbPersistentDataSnapshot *createSnapshot(uint64_t mvccCheckpoint, bool fOptional) {
@@ -2729,6 +2731,8 @@ 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;
 

src/storage/rocksdb.cpp

@@ -43,9 +43,8 @@ void RocksDBStorageProvider::insert(const char *key, size_t cchKey, void *data,
         ++m_count;
 }
 
-void RocksDBStorageProvider::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem)
+void RocksDBStorageProvider::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem)
 {
-    size_t coverwrites = 0;
     if (celem >= 16384) {
         rocksdb::Options options = DefaultRocksDBOptions();
         rocksdb::SstFileWriter sst_file_writer(rocksdb::EnvOptions(), options, options.comparator);
@@ -93,15 +92,8 @@ void RocksDBStorageProvider::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **
         m_spdb->Write(WriteOptions(), spbatch.get());
     }
 
-    if (rgfOverwrite != nullptr) {
-        for (size_t ielem = 0; ielem < celem; ++ielem) {
-            if (rgfOverwrite[ielem])
-                ++coverwrites;
-        }
-    }
-
     std::unique_lock<fastlock> l(m_lock);
-    m_count += celem - coverwrites;
+    m_count += celem;
 }
 
 bool RocksDBStorageProvider::erase(const char *key, size_t cchKey)
@@ -176,6 +168,28 @@ bool RocksDBStorageProvider::enumerate(callback fn) const
     return !it->Valid();
 }
 
+size_t RocksDBStorageProvider::stateful_enumerate(callback fn)
+{
+    if (m_iter == nullptr || !m_iter->Valid()) {
+        m_iter = std::unique_ptr<rocksdb::Iterator>(m_spdb->NewIterator(ReadOptions(), m_spcolfamily.get()));
+        m_iter->SeekToFirst();
+    }
+    size_t count = 0;
+    while (m_iter->Valid()) {
+        if (FInternalKey(m_iter->key().data(), m_iter->key().size()))
+            continue;
+
+        if (fn(m_iter->key().data(), m_iter->key().size(), m_iter->value().data(), m_iter->value().size())) {
+            m_iter->Next();
+            ++count;
+        } else {
+            break;
+        }
+    }
+    assert(m_iter->status().ok()); // Check for any errors found during the scan
+    return count;
+}
+
 const IStorage *RocksDBStorageProvider::clone() const
 {
     std::unique_lock<fastlock> l(m_lock);

src/storage/rocksdb.h

@@ -22,6 +22,7 @@ class RocksDBStorageProvider : public IStorage
     rocksdb::ReadOptions m_readOptionsTemplate;
     size_t m_count = 0;
     mutable fastlock m_lock {"RocksDBStorageProvider"};
+    std::unique_ptr<rocksdb::Iterator> m_iter;
 
 public:
     RocksDBStorageProvider(RocksDBStorageFactory *pfactory, std::shared_ptr<rocksdb::DB> &spdb, std::shared_ptr<rocksdb::ColumnFamilyHandle> &spcolfam, const rocksdb::Snapshot *psnapshot, size_t count);
@@ -32,13 +33,14 @@ public:
     virtual void retrieve(const char *key, size_t cchKey, callbackSingle fn) const override;
     virtual size_t clear() override;
     virtual bool enumerate(callback fn) const override;
+    virtual size_t stateful_enumerate(callback fn) override;
 
     virtual const IStorage *clone() const override;
 
     virtual void beginWriteBatch() override;
     virtual void endWriteBatch() override;
 
-    virtual void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, char *rgfOverwrite, size_t celem) override;
+    virtual void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem) override;
 
     virtual void batch_lock() override;
     virtual void batch_unlock() override;

src/storage/teststorageprovider.cpp

@@ -73,6 +73,26 @@ bool TestStorageProvider::enumerate(callback fn) const
     }
     return fAll;
 }
+
+size_t TestStorageProvider::stateful_enumerate(callback fn)
+{
+    size_t count = 0;
+    if (m_iter == m_map.end())
+        m_iter = m_map.begin();
+
+    while (m_iter != m_map.end())
+    {
+        if (fn(m_iter->first.data(), m_iter->first.size(), m_iter->second.data(), m_iter->second.size()))
+        {
+            m_iter++;
+            count++;
+        } else
+        {
+            break;
+        }
+    }
+    return count;
+}
     
 size_t TestStorageProvider::count() const
 {

src/storage/teststorageprovider.h

@@ -14,6 +14,7 @@ class TestStorageFactory : public IStorageFactory
 class TestStorageProvider final : public IStorage
 {
     std::unordered_map<std::string, std::string> m_map;
+    std::unordered_map<std::string, std::string>::iterator m_iter = m_map.end();
 
 public:
     TestStorageProvider();
@@ -24,6 +25,7 @@ public:
     virtual void retrieve(const char *key, size_t cchKey, callbackSingle fn) const override;
     virtual size_t clear() override;
     virtual bool enumerate(callback fn) const override;
+    virtual size_t stateful_enumerate(callback fn) override;
     virtual size_t count() const override;
 
     virtual void flush() override;

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/flash.tcl

@@ -120,6 +120,21 @@ if {$::flash_enabled} {
             assert_equal {2} [r scard set1]
         }
 
+        test { Keys in storage are expired } {
+            r flushdb
+            r psetex key1 10000 a
+            r psetex key2 10000 a
+            r psetex key3 10000 a
+            r flushall cache
+
+            after 1000
+            wait_for_condition 1000 50 {
+                [r dbsize] == 0
+            } else {
+                fail "keys in ssd only not expired"
+            }
+        }
+
         r flushall
         # If a weak storage memory model is set, wait for any pending snapshot writes to finish
         after 500 

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