add config to disable storage key cache

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>

deps/Makefile

@@ -95,7 +95,7 @@ JEMALLOC_LDFLAGS= $(LDFLAGS)
 
 jemalloc: .make-prerequisites
 	@printf '%b %b\n' $(MAKECOLOR)MAKE$(ENDCOLOR) $(BINCOLOR)$@$(ENDCOLOR)
-	cd jemalloc && ./configure --with-version=5.2.1-0-g0 --with-lg-quantum=3 --with-jemalloc-prefix=je_ --disable-cxx CFLAGS="$(JEMALLOC_CFLAGS)" LDFLAGS="$(JEMALLOC_LDFLAGS)"
+	cd jemalloc && ./configure --with-version=5.2.1-0-g0 --with-lg-quantum=3 --disable-cxx CFLAGS="$(JEMALLOC_CFLAGS)" LDFLAGS="$(JEMALLOC_LDFLAGS)"
 	cd jemalloc && $(MAKE) CFLAGS="$(JEMALLOC_CFLAGS)" LDFLAGS="$(JEMALLOC_LDFLAGS)" lib/libjemalloc.a
 
 .PHONY: jemalloc

src/Makefile

@@ -25,7 +25,7 @@ NODEPS:=clean distclean
 
 # Default settings
 STD=-pedantic -DREDIS_STATIC=''
-CXX_STD=-std=c++14 -pedantic -fno-rtti -D__STDC_FORMAT_MACROS  
+CXX_STD=-std=c++17 -pedantic -fno-rtti -D__STDC_FORMAT_MACROS  
 ifneq (,$(findstring clang,$(CC)))
   STD+=-Wno-c11-extensions
 else

src/SnapshotPayloadParseState.cpp

@@ -140,7 +140,7 @@ void SnapshotPayloadParseState::flushQueuedKeys() {
     auto &insertsInFlightTmp = insertsInFlight; // C++ GRRRRRRRRRRRRRRRR, we don't want to capute "this" because that's dangerous
     if (current_database < cserver.dbnum) {
         g_pserver->asyncworkqueue->AddWorkFunction([idb, vecqueuedKeys = std::move(this->vecqueuedKeys), vecqueuedKeysCb = std::move(this->vecqueuedKeysCb), vecqueuedVals = std::move(this->vecqueuedVals), vecqueuedValsCb = std::move(this->vecqueuedValsCb), &insertsInFlightTmp, pallocator = m_spallocator.release()]() mutable {
-            g_pserver->db[idb]->bulkStorageInsert(vecqueuedKeys.data(), vecqueuedKeysCb.data(), vecqueuedVals.data(), vecqueuedValsCb.data(), vecqueuedKeys.size());
+            g_pserver->db[idb]->bulkDirectStorageInsert(vecqueuedKeys.data(), vecqueuedKeysCb.data(), vecqueuedVals.data(), vecqueuedValsCb.data(), vecqueuedKeys.size());
             --insertsInFlightTmp;
             delete pallocator;
         });

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/ae.cpp

@@ -859,6 +859,11 @@ void aeReleaseForkLock()
     g_forkLock.downgradeWrite();
 }
 
+void aeForkLockInChild()
+{
+    g_forkLock.setNotify(false);
+}
+
 int aeThreadOwnsLock()
 {
     return fOwnLockOverride || g_lock.fOwnLock();

src/ae.h

@@ -171,6 +171,7 @@ int aeTryAcquireLock(int fWeak);
 void aeThreadOffline();
 void aeReleaseLock();
 void aeReleaseForkLock();
+void aeForkLockInChild();
 int aeThreadOwnsLock();
 void aeSetThreadOwnsLockOverride(int fOverride);
 int aeLockContested(int threshold);

src/anet.c

@@ -441,7 +441,7 @@ static int _anetTcpServer(char *err, int port, const char *bindaddr, int af, int
     char _port[6];  /* strlen("65535") */
     struct addrinfo hints, *servinfo, *p;
 
-    snprintf(_port,6,"%d",port);
+    snprintf(_port,sizeof(_port),"%d",port);
     memset(&hints,0,sizeof(hints));
     hints.ai_family = af;
     hints.ai_socktype = SOCK_STREAM;

src/aof.cpp

@@ -752,7 +752,7 @@ void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int a
      * accumulate the differences between the child DB and the current one
      * in a buffer, so that when the child process will do its work we
      * can append the differences to the new append only file. */
-    if (g_pserver->child_type == CHILD_TYPE_AOF)
+    if (hasActiveChildProcess() && g_pserver->child_type == CHILD_TYPE_AOF)
         aofRewriteBufferAppend((unsigned char*)buf,sdslen(buf));
 
     sdsfree(buf);
@@ -1658,7 +1658,7 @@ int rewriteAppendOnlyFile(char *filename) {
 { // BEGIN GOTO SCOPED VARIABLES
     /* Note that we have to use a different temp name here compared to the
      * one used by rewriteAppendOnlyFileBackground() function. */
-    snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid());
+    snprintf(tmpfile,sizeof(tmpfile),"temp-rewriteaof-%d.aof", (int) getpid());
     fp = fopen(tmpfile,"w");
     if (!fp) {
         serverLog(LL_WARNING, "Opening the temp file for AOF rewrite in rewriteAppendOnlyFile(): %s", strerror(errno));
@@ -1895,7 +1895,7 @@ int rewriteAppendOnlyFileBackground(void) {
         /* Child */
         redisSetProcTitle("keydb-aof-rewrite");
         redisSetCpuAffinity(g_pserver->aof_rewrite_cpulist);
-        snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
+        snprintf(tmpfile,sizeof(tmpfile),"temp-rewriteaof-bg-%d.aof", (int) getpid());
         if (rewriteAppendOnlyFile(tmpfile) == C_OK) {
             sendChildCowInfo(CHILD_INFO_TYPE_AOF_COW_SIZE, "AOF rewrite");
             exitFromChild(0);
@@ -1944,10 +1944,10 @@ void bgrewriteaofCommand(client *c) {
 void aofRemoveTempFile(pid_t childpid) {
     char tmpfile[256];
 
-    snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) childpid);
+    snprintf(tmpfile,sizeof(tmpfile),"temp-rewriteaof-bg-%d.aof", (int) childpid);
     bg_unlink(tmpfile);
 
-    snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) childpid);
+    snprintf(tmpfile,sizeof(tmpfile),"temp-rewriteaof-%d.aof", (int) childpid);
     bg_unlink(tmpfile);
 }
 
@@ -1985,7 +1985,7 @@ void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
         /* Flush the differences accumulated by the parent to the
          * rewritten AOF. */
         latencyStartMonitor(latency);
-        snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof",
+        snprintf(tmpfile,sizeof(tmpfile),"temp-rewriteaof-bg-%d.aof",
             (int)g_pserver->child_pid);
         newfd = open(tmpfile,O_WRONLY|O_APPEND);
         if (newfd == -1) {

src/childinfo.cpp

@@ -42,6 +42,7 @@ typedef struct {
  * RDB / AOF saving process from the child to the parent (for instance
  * the amount of copy on write memory used) */
 void openChildInfoPipe(void) {
+    serverAssert(g_pserver->child_info_pipe[0] == -1);
     if (pipe(g_pserver->child_info_pipe) == -1) {
         /* On error our two file descriptors should be still set to -1,
          * but we call anyway closeChildInfoPipe() since can't hurt. */

src/cluster.cpp

@@ -5107,7 +5107,7 @@ void createDumpPayload(rio *payload, robj_roptr o, robj *key) {
     serverAssert(rdbSaveObject(payload,o,key));
     char szT[32];
     uint64_t mvcc = mvccFromObj(o);
-    snprintf(szT, 32, "%" PRIu64, mvcc);
+    snprintf(szT, sizeof(szT), "%" PRIu64, mvcc);
     serverAssert(rdbSaveAuxFieldStrStr(payload,"mvcc-tstamp", szT) != -1);
 
     /* Write the footer, this is how it looks like:

src/config.cpp

@@ -35,6 +35,7 @@
 
 #include <fcntl.h>
 #include <sys/stat.h>
+#include <sys/wait.h>
 #ifdef __linux__
 #include <sys/sysinfo.h>
 #endif
@@ -2456,6 +2457,32 @@ static int isValidAOFfilename(char *val, const char **err) {
     return 1;
 }
 
+static int isValidS3Bucket(char *s3bucket, const char **err) {
+    int status = EXIT_FAILURE;
+    pid_t pid = fork();
+    if (pid < 0)
+    {
+        *err = "couldn't fork to call aws cli";
+        return 0;
+    }
+
+    if (pid == 0)
+    {
+        execlp("aws", "aws", "s3", "ls", s3bucket);
+        exit(EXIT_FAILURE);
+    }
+    else 
+    {
+        waitpid(pid, &status, 0);
+    }
+
+    if (status != EXIT_SUCCESS) {
+        *err = "could not access s3 bucket";
+        return 0;
+    }
+    return 1;
+}
+
 /* Validate specified string is a valid proc-title-template */
 static int isValidProcTitleTemplate(char *val, const char **err) {
     if (!validateProcTitleTemplate(val)) {
@@ -2588,7 +2615,7 @@ static int updateMaxclients(long long val, long long prev, const char **err) {
         adjustOpenFilesLimit();
         if (g_pserver->maxclients != val) {
             static char msg[128];
-            snprintf(msg, 128, "The operating system is not able to handle the specified number of clients, try with %d", g_pserver->maxclients);
+            snprintf(msg, sizeof(msg), "The operating system is not able to handle the specified number of clients, try with %d", g_pserver->maxclients);
             *err = msg;
             if (g_pserver->maxclients > prev) {
                 g_pserver->maxclients = prev;
@@ -2627,7 +2654,7 @@ static int updateMaxclients(long long val, long long prev, const char **err) {
 
                 if (res != AE_OK){
                     static char msg[128];
-                    snprintf(msg, 128,"Failed to post the request to change setsize for Thread %d", iel);
+                    snprintf(msg, sizeof(msg),"Failed to post the request to change setsize for Thread %d", iel);
                     *err = msg;
                     return 0;
                 }
@@ -2779,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),
@@ -2803,6 +2830,7 @@ standardConfig configs[] = {
     createStringConfig("cluster-announce-ip", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->cluster_announce_ip, NULL, NULL, NULL),
     createStringConfig("syslog-ident", NULL, IMMUTABLE_CONFIG, ALLOW_EMPTY_STRING, g_pserver->syslog_ident, "redis", NULL, NULL),
     createStringConfig("dbfilename", NULL, MODIFIABLE_CONFIG, ALLOW_EMPTY_STRING, g_pserver->rdb_filename, CONFIG_DEFAULT_RDB_FILENAME, isValidDBfilename, NULL),
+    createStringConfig("db-s3-object", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->rdb_s3bucketpath, NULL, isValidS3Bucket, NULL),
     createStringConfig("appendfilename", NULL, IMMUTABLE_CONFIG, ALLOW_EMPTY_STRING, g_pserver->aof_filename, "appendonly.aof", isValidAOFfilename, NULL),
     createStringConfig("server_cpulist", NULL, IMMUTABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->server_cpulist, NULL, NULL, NULL),
     createStringConfig("bio_cpulist", NULL, IMMUTABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->bio_cpulist, NULL, NULL, NULL),
@@ -2884,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),
@@ -2911,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

@@ -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;
 
@@ -2607,12 +2618,15 @@ void clusterStorageLoadCallback(const char *rgchkey, size_t cch, void *)
     slotToKeyUpdateKeyCore(rgchkey, cch, true /*add*/);
 }
 
-void moduleLoadCallback(const char * rgchKey, size_t cchKey, void *data) {
-    if (g_pserver->cluster_enabled) {
-        clusterStorageLoadCallback(rgchKey, cchKey, data);
-    }
-    robj *keyobj = createEmbeddedStringObject(rgchKey, cchKey);
-    moduleNotifyKeyspaceEvent(NOTIFY_LOADED, "loaded", keyobj, *(int *)data);
+void moduleLoadCallback(const char * rgchKey, size_t, void *data) {
+    redisObjectStack keyobj;
+    initStaticStringObject(keyobj, const_cast<char *>(rgchKey));
+    moduleNotifyKeyspaceEvent(NOTIFY_LOADED, "loaded", &keyobj, *(int *)data);
+}
+
+void moduleClusterLoadCallback(const char * rgchKey, size_t cchKey, void *data) {
+    clusterStorageLoadCallback(rgchKey, cchKey, data);
+    moduleLoadCallback(rgchKey, cchKey, data);
 }
 
 void redisDb::initialize(int id)
@@ -2633,7 +2647,7 @@ void redisDb::storageProviderInitialize()
 {
     if (g_pserver->m_pstorageFactory != nullptr)
     {
-        IStorageFactory::key_load_iterator itr = moduleLoadCallback;
+        IStorageFactory::key_load_iterator itr = g_pserver->cluster_enabled ? moduleClusterLoadCallback : moduleLoadCallback;
         this->setStorageProvider(StorageCache::create(g_pserver->m_pstorageFactory, id, itr, &id));
     }
 }
@@ -2724,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);
@@ -2858,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);
@@ -2881,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()));
     }
 }
@@ -3002,8 +3021,16 @@ void redisDbPersistentData::processChangesAsync(std::atomic<int> &pendingJobs)
     });
 }
 
-void redisDbPersistentData::bulkStorageInsert(char **rgKeys, size_t *rgcbKeys, char **rgVals, size_t *rgcbVals, size_t celem)
+/* This function is to bulk insert directly to storage provider bypassing in memory, assumes rgKeys and rgVals are not sds strings */
+void redisDbPersistentData::bulkDirectStorageInsert(char **rgKeys, size_t *rgcbKeys, char **rgVals, size_t *rgcbVals, size_t celem)
 {
+    if (g_pserver->cluster_enabled) {
+        aeAcquireLock();
+        for (size_t i = 0; i < celem; i++) {
+            slotToKeyUpdateKeyCore(rgKeys[i], rgcbKeys[i], 1);
+        }
+        aeReleaseLock();
+    }
     m_spstorage->bulkInsert(rgKeys, rgcbKeys, rgVals, rgcbVals, celem);
 }
 
@@ -3245,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())
@@ -3265,7 +3293,7 @@ int dbnumFromDb(redisDb *db)
     serverPanic("invalid database pointer");
 }
 
-bool redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command, bool fExecOK)
+void redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command)
 {
     if (m_spstorage == nullptr) {
 #if defined(__x86_64__) || defined(__i386__)
@@ -3277,7 +3305,7 @@ bool redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
             const char *cmd = szFromObj(command.argv[0]);
             if (!strcasecmp(cmd, "set") || !strcasecmp(cmd, "get")) {
                 if (c->db->m_spdbSnapshotHOLDER != nullptr)
-                    return false; // this is dangerous enough without a snapshot around
+                    return; // this is dangerous enough without a snapshot around
                 auto h = dictSdsHash(szFromObj(command.argv[1]));
                 for (int iht = 0; iht < 2; ++iht) {
                     auto hT = h & c->db->m_pdict->ht[iht].sizemask;
@@ -3297,7 +3325,7 @@ bool redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
             }
         }
 #endif
-        return false;
+        return;
     }
 
     AeLocker lock;
@@ -3307,10 +3335,10 @@ bool redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
     getKeysResult result = GETKEYS_RESULT_INIT;
     auto cmd = lookupCommand(szFromObj(command.argv[0]));
     if (cmd == nullptr)
-        return false; // Bad command? It's not for us to judge, just bail
+        return; // Bad command? It's not for us to judge, just bail
     
     if (command.argc < std::abs(cmd->arity))
-        return false; // Invalid number of args
+        return; // Invalid number of args
     
     int numkeys = getKeysFromCommand(cmd, command.argv, command.argc, &result);
     for (int ikey = 0; ikey < numkeys; ++ikey)
@@ -3343,7 +3371,6 @@ bool redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
         }
     }
 
-    bool fNoInsert = false;
     if (!vecInserts.empty()) {
         lock.arm(c);
         for (auto &tuple : vecInserts)
@@ -3359,19 +3386,18 @@ bool redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
                     // While unlocked this was already ensured
                     decrRefCount(o);
                     sdsfree(sharedKey);
-                    fNoInsert = true;
                 }
                 else
                 {
                     if (spexpire != nullptr) {
                         if (spexpire->when() < mstime()) {
-                            fNoInsert = true;
                             break;
                         }
                     }
                     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);
@@ -3392,12 +3418,5 @@ bool redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
         lock.disarm();
     }
 
-    if (fExecOK && !fNoInsert && cmd->proc == getCommand && !vecInserts.empty()) {
-        robj *o = std::get<1>(vecInserts[0]);
-        if (o != nullptr) {
-            addReplyBulk(c, o);
-            return true;
-        }
-    }
-    return false;
-}
+    return;
+}

src/debug.cpp

@@ -341,10 +341,10 @@ void mallctl_int(client *c, robj **argv, int argc) {
     }
     size_t sz = sizeof(old);
     while (sz > 0) {
-        if ((ret=je_mallctl(szFromObj(argv[0]), &old, &sz, argc > 1? &val: NULL, argc > 1?sz: 0))) {
+        if ((ret=mallctl(szFromObj(argv[0]), &old, &sz, argc > 1? &val: NULL, argc > 1?sz: 0))) {
             if (ret == EPERM && argc > 1) {
                 /* if this option is write only, try just writing to it. */
-                if (!(ret=je_mallctl(szFromObj(argv[0]), NULL, 0, &val, sz))) {
+                if (!(ret=mallctl(szFromObj(argv[0]), NULL, 0, &val, sz))) {
                     addReply(c, shared.ok);
                     return;
                 }
@@ -375,7 +375,7 @@ void mallctl_string(client *c, robj **argv, int argc) {
     char *old;
     size_t sz = sizeof(old);
     /* for strings, it seems we need to first get the old value, before overriding it. */
-    if ((rret=je_mallctl(szFromObj(argv[0]), &old, &sz, NULL, 0))) {
+    if ((rret=mallctl(szFromObj(argv[0]), &old, &sz, NULL, 0))) {
         /* return error unless this option is write only. */
         if (!(rret == EPERM && argc > 1)) {
             addReplyErrorFormat(c,"%s", strerror(rret));
@@ -387,7 +387,7 @@ void mallctl_string(client *c, robj **argv, int argc) {
         char **valref = &val;
         if ((!strcmp(val,"VOID")))
             valref = NULL, sz = 0;
-        wret = je_mallctl(szFromObj(argv[0]), NULL, 0, valref, sz);
+        wret = mallctl(szFromObj(argv[0]), NULL, 0, valref, sz);
     }
     if (!rret)
         addReplyBulkCString(c, old);

src/dict.cpp

@@ -1577,7 +1577,7 @@ char *stringFromLongLong(long long value) {
     int len;
     char *s;
 
-    len = snprintf(buf,32,"%lld",value);
+    len = snprintf(buf,sizeof(buf),"%lld",value);
     s = zmalloc(len+1);
     memcpy(s, buf, len);
     s[len] = '\0';

src/endianconv.c

@@ -112,15 +112,15 @@ int endianconvTest(int argc, char *argv[], int accurate) {
     UNUSED(argv);
     UNUSED(accurate);
 
-    snprintf(buf,32,"ciaoroma");
+    snprintf(buf,sizeof(buf),"ciaoroma");
     memrev16(buf);
     printf("%s\n", buf);
 
-    snprintf(buf,32,"ciaoroma");
+    snprintf(buf,sizeof(buf),"ciaoroma");
     memrev32(buf);
     printf("%s\n", buf);
 
-    snprintf(buf,32,"ciaoroma");
+    snprintf(buf,sizeof(buf),"ciaoroma");
     memrev64(buf);
     printf("%s\n", buf);
 

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

@@ -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/keydb-diagnostic-tool.cpp

@@ -904,7 +904,7 @@ int main(int argc, const char **argv) {
 
     while (self_threads < config.max_threads) {
         for (int i = 0; i < config.numclients; i++) {
-            snprintf(command, 63, "SET %d %s\r\n", self_threads * config.numclients + i, set_value);
+            snprintf(command, sizeof(command), "SET %d %s\r\n", self_threads * config.numclients + i, set_value);
             createClient(command, strlen(command), NULL,self_threads);
         }
 

src/modules/hellotimer.c

@@ -51,8 +51,9 @@ int TimerCommand_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int
 
     for (int j = 0; j < 10; j++) {
         int delay = rand() % 5000;
-        char *buf = RedisModule_Alloc(256);
-        snprintf(buf,256,"After %d", delay);
+        int bufsize = 256;
+        char *buf = RedisModule_Alloc(bufsize);
+        snprintf(buf,bufsize,"After %d", delay);
         RedisModuleTimerID tid = RedisModule_CreateTimer(ctx,delay,timerHandler,buf);
         REDISMODULE_NOT_USED(tid);
     }

src/networking.cpp

@@ -145,7 +145,7 @@ client *createClient(connection *conn, int iel) {
     client_id = g_pserver->next_client_id.fetch_add(1);
     c->iel = iel;
     c->id = client_id;
-    snprintf(c->lock.szName, 56, "client %" PRIu64, client_id);
+    snprintf(c->lock.szName, sizeof(c->lock.szName), "client %" PRIu64, client_id);
     c->resp = 2;
     c->conn = conn;
     c->name = NULL;
@@ -777,11 +777,11 @@ void setDeferredAggregateLen(client *c, void *node, long length, char prefix) {
          * we return NULL in addReplyDeferredLen() */
         if (node == NULL) return;
         char lenstr[128];
-        size_t lenstr_len = snprintf(lenstr, 128, "%c%ld\r\n", prefix, length);
+        size_t lenstr_len = snprintf(lenstr, sizeof(lenstr), "%c%ld\r\n", prefix, length);
         setDeferredReply(c, node, lenstr, lenstr_len);
     } else {
         char lenstr[128];
-        int lenstr_len = snprintf(lenstr, 128, "%c%ld\r\n", prefix, length);
+        int lenstr_len = snprintf(lenstr, sizeof(lenstr), "%c%ld\r\n", prefix, length);
 
         size_t idxSplice = (size_t)node;
         serverAssert(idxSplice <= c->replyAsync->used);
@@ -2595,9 +2595,7 @@ void parseClientCommandBuffer(client *c) {
             (g_pserver->m_pstorageFactory || aeLockContested(cserver.cthreads/2) || cserver.cthreads == 1) && !GlobalLocksAcquired()) {
             auto &query = c->vecqueuedcmd.back();
             if (query.argc > 0 && query.argc == query.argcMax) {
-                if (c->db->prefetchKeysAsync(c, query, c->vecqueuedcmd.size() == 1)) {
-                    c->vecqueuedcmd.erase(c->vecqueuedcmd.begin());
-                }
+                c->db->prefetchKeysAsync(c, query);
             }
         }
         c->reqtype = 0;

src/new.cpp

@@ -41,10 +41,3 @@ void operator delete(void *p, std::size_t) noexcept
 }
 
 #endif
-
-#if defined(USE_JEMALLOC)
-extern "C" size_t malloc_usable_size(void *ptr)
-{
-    return zmalloc_usable_size(ptr);
-}
-#endif

src/object.cpp

@@ -1541,7 +1541,7 @@ void memoryCommand(client *c) {
     } else if (!strcasecmp(szFromObj(c->argv[1]),"malloc-stats") && c->argc == 2) {
 #if defined(USE_JEMALLOC)
         sds info = sdsempty();
-        je_malloc_stats_print(inputCatSds, &info, NULL);
+        malloc_stats_print(inputCatSds, &info, NULL);
         addReplyVerbatim(c,info,sdslen(info),"txt");
         sdsfree(info);
 #else
@@ -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

@@ -1163,7 +1163,7 @@ int rdbSaveKeyValuePair(rio *rdb, robj_roptr key, robj_roptr val, const expireEn
 
     char szT[32];
     if (g_pserver->fActiveReplica) {
-        snprintf(szT, 32, "%" PRIu64, mvccFromObj(val));
+        snprintf(szT, sizeof(szT), "%" PRIu64, mvccFromObj(val));
         if (rdbSaveAuxFieldStrStr(rdb,"mvcc-tstamp", szT) == -1) return -1;
     }
 
@@ -1190,7 +1190,7 @@ int rdbSaveKeyValuePair(rio *rdb, robj_roptr key, robj_roptr val, const expireEn
         {
             if (itr.subkey() == nullptr)
                 continue;   // already saved
-            snprintf(szT, 32, "%lld", itr.when());
+            snprintf(szT, sizeof(szT), "%lld", itr.when());
             rdbSaveAuxFieldStrStr(rdb,"keydb-subexpire-key",itr.subkey());
             rdbSaveAuxFieldStrStr(rdb,"keydb-subexpire-when",szT);
         }
@@ -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 */
@@ -1490,7 +1490,7 @@ int rdbSaveFile(char *filename, const redisDbPersistentDataSnapshot **rgpdb, rdb
     rio rdb;
     int error = 0;
 
-    snprintf(tmpfile,256,"temp-%d-%d.rdb", getpid(), g_pserver->rdbThreadVars.tmpfileNum);
+    getTempFileName(tmpfile, g_pserver->rdbThreadVars.tmpfileNum);
     fp = fopen(tmpfile,"w");
     if (!fp) {
         char *cwdp = getcwd(cwd,MAXPATHLEN);
@@ -1660,12 +1660,14 @@ int launchRdbSaveThread(pthread_t &child, rdbSaveInfo *rsi)
         pthread_attr_t tattr;
         pthread_attr_init(&tattr);
         pthread_attr_setstacksize(&tattr, 1 << 23); // 8 MB
+        openChildInfoPipe();
         if (pthread_create(&child, &tattr, rdbSaveThread, args)) {
             pthread_attr_destroy(&tattr);
             for (int idb = 0; idb < cserver.dbnum; ++idb)
                 g_pserver->db[idb]->endSnapshot(args->rgpdb[idb]);
             args->~rdbSaveThreadArgs();
             zfree(args);
+            closeChildInfoPipe();
             return C_ERR;
         }
         pthread_attr_destroy(&tattr);
@@ -1683,13 +1685,11 @@ int rdbSaveBackground(rdbSaveInfo *rsi) {
 
     g_pserver->dirty_before_bgsave = g_pserver->dirty;
     g_pserver->lastbgsave_try = time(NULL);
-    openChildInfoPipe();
 
     start = ustime();
     latencyStartMonitor(g_pserver->rdb_save_latency);
 
     if (launchRdbSaveThread(child, rsi) != C_OK) {
-        closeChildInfoPipe();
         g_pserver->lastbgsave_status = C_ERR;
         serverLog(LL_WARNING,"Can't save in background: fork: %s",
             strerror(errno));
@@ -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);
@@ -3694,7 +3694,7 @@ void killRDBChild(bool fSynchronous) {
     serverAssert(GlobalLocksAcquired());
 
     if (cserver.fForkBgSave) {
-        kill(g_pserver->rdb_child_pid,SIGUSR1);
+        kill(g_pserver->child_pid,SIGUSR1);
     } else { 
         g_pserver->rdbThreadVars.fRdbThreadCancel = true;
         if (g_pserver->rdb_child_type == RDB_CHILD_TYPE_SOCKET) {
@@ -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/rdb.h

@@ -151,6 +151,7 @@ int rdbLoad(rdbSaveInfo *rsi, int rdbflags);
 int rdbLoadFile(const char *filename, rdbSaveInfo *rsi, int rdbflags);
 int rdbSaveBackground(rdbSaveInfo *rsi);
 int rdbSaveToSlavesSockets(rdbSaveInfo *rsi);
+void getTempFileName(char tmpfile[], int tmpfileNum);
 void rdbRemoveTempFile(pid_t childpid, int from_signal);
 int rdbSave(const redisDbPersistentDataSnapshot **rgpdb, rdbSaveInfo *rsi);
 int rdbSaveFile(char *filename, const redisDbPersistentDataSnapshot **rgpdb, rdbSaveInfo *rsi);

src/readwritelock.h

@@ -8,6 +8,7 @@ class readWriteLock {
     int m_readCount = 0;
     int m_writeCount = 0;
     bool m_writeWaiting = false;
+    bool m_notify = true;
 public:
     readWriteLock(const char *name) : m_readLock(name), m_writeLock(name) {}
 
@@ -65,7 +66,8 @@ public:
     void releaseRead() {
         std::unique_lock<fastlock> rm(m_readLock);
         m_readCount--;
-        m_cv.notify_all();
+        if (m_notify)
+            m_cv.notify_all();
     }
 
     void releaseWrite(bool exclusive = true) {
@@ -74,7 +76,8 @@ public:
         if (exclusive)
             m_writeLock.unlock();
         m_writeCount--;
-        m_cv.notify_all();
+        if (m_notify)
+            m_cv.notify_all();
     }
 
     void downgradeWrite(bool exclusive = true) {
@@ -82,6 +85,10 @@ public:
         acquireRead();
     }
 
+    void setNotify(bool notify) {
+        m_notify = notify;
+    }
+
     bool hasReader() {
         return m_readCount > 0;
     }

src/redis-cli.c

@@ -3293,8 +3293,8 @@ static redisReply *clusterManagerMigrateKeysInReply(clusterManagerNode *source,
     argv_len = zcalloc(argc * sizeof(size_t), MALLOC_LOCAL);
     char portstr[255];
     char timeoutstr[255];
-    snprintf(portstr, 10, "%d", target->port);
-    snprintf(timeoutstr, 10, "%d", timeout);
+    snprintf(portstr, sizeof(portstr), "%d", target->port);
+    snprintf(timeoutstr, sizeof(timeoutstr), "%d", timeout);
     argv[0] = "MIGRATE";
     argv_len[0] = 7;
     argv[1] = target->ip;
@@ -6825,7 +6825,7 @@ static long getLongInfoField(char *info, char *field) {
 
 /* Convert number of bytes into a human readable string of the form:
  * 100B, 2G, 100M, 4K, and so forth. */
-void bytesToHuman(char *s, long long n) {
+void bytesToHuman(char *s, long long n, size_t bufsize) {
     double d;
 
     if (n < 0) {
@@ -6835,17 +6835,17 @@ void bytesToHuman(char *s, long long n) {
     }
     if (n < 1024) {
         /* Bytes */
-        snprintf(s,256,"%lldB",n);
+        snprintf(s,bufsize,"%lldB",n);
         return;
     } else if (n < (1024*1024)) {
         d = (double)n/(1024);
-        snprintf(s,256,"%.2fK",d);
+        snprintf(s,bufsize,"%.2fK",d);
     } else if (n < (1024LL*1024*1024)) {
         d = (double)n/(1024*1024);
-        snprintf(s,256,"%.2fM",d);
+        snprintf(s,bufsize,"%.2fM",d);
     } else if (n < (1024LL*1024*1024*1024)) {
         d = (double)n/(1024LL*1024*1024);
-        snprintf(s,256,"%.2fG",d);
+        snprintf(s,bufsize,"%.2fG",d);
     }
 }
 
@@ -6875,38 +6875,38 @@ static void statMode(void) {
         for (j = 0; j < 20; j++) {
             long k;
 
-            snprintf(buf,64,"db%d:keys",j);
+            snprintf(buf,sizeof(buf),"db%d:keys",j);
             k = getLongInfoField(reply->str,buf);
             if (k == LONG_MIN) continue;
             aux += k;
         }
-        snprintf(buf,64,"%ld",aux);
+        snprintf(buf,sizeof(buf),"%ld",aux);
         printf("%-11s",buf);
 
         /* Used memory */
         aux = getLongInfoField(reply->str,"used_memory");
-        bytesToHuman(buf,aux);
+        bytesToHuman(buf,aux,sizeof(buf));
         printf("%-8s",buf);
 
         /* Clients */
         aux = getLongInfoField(reply->str,"connected_clients");
-        snprintf(buf,64,"%ld",aux);
+        snprintf(buf,sizeof(buf),"%ld",aux);
         printf(" %-8s",buf);
 
         /* Blocked (BLPOPPING) Clients */
         aux = getLongInfoField(reply->str,"blocked_clients");
-        snprintf(buf,64,"%ld",aux);
+        snprintf(buf,sizeof(buf),"%ld",aux);
         printf("%-8s",buf);
 
         /* Requests */
         aux = getLongInfoField(reply->str,"total_commands_processed");
-        snprintf(buf,64,"%ld (+%ld)",aux,requests == 0 ? 0 : aux-requests);
+        snprintf(buf,sizeof(buf),"%ld (+%ld)",aux,requests == 0 ? 0 : aux-requests);
         printf("%-19s",buf);
         requests = aux;
 
         /* Connections */
         aux = getLongInfoField(reply->str,"total_connections_received");
-        snprintf(buf,64,"%ld",aux);
+        snprintf(buf,sizeof(buf),"%ld",aux);
         printf(" %-12s",buf);
 
         /* Children */

src/replication.cpp

@@ -3736,7 +3736,7 @@ retry_connect:
         while(maxtries--) {
             auto dt = std::chrono::system_clock::now().time_since_epoch();
             auto dtMillisecond = std::chrono::duration_cast<std::chrono::milliseconds>(dt);
-            snprintf(tmpfile,256,
+            snprintf(tmpfile,sizeof(tmpfile),
                 "temp-%d.%ld.rdb",(int)dtMillisecond.count(),(long int)getpid());
             dfd = open(tmpfile,O_CREAT|O_WRONLY|O_EXCL,0644);
             if (dfd != -1) break;
@@ -3797,9 +3797,11 @@ int connectWithMaster(redisMaster *mi) {
     connSetPrivateData(mi->repl_transfer_s, mi);
     if (connConnect(mi->repl_transfer_s, mi->masterhost, mi->masterport,
                 NET_FIRST_BIND_ADDR, syncWithMaster) == C_ERR) {
-        int sev = g_pserver->enable_multimaster ? LL_NOTICE : LL_WARNING;   // with multimaster its not unheard of to intentiallionall have downed masters
-        serverLog(sev,"Unable to connect to MASTER: %s",
-                connGetLastError(mi->repl_transfer_s));
+        const char *err = "Unknown Error";
+        if (mi->repl_transfer_s->last_errno != 0)
+            err = connGetLastError(mi->repl_transfer_s);
+        int sev = g_pserver->enable_multimaster ? LL_NOTICE : LL_WARNING;   // with multimaster its not unheard of to intentionally have downed masters
+        serverLog(sev, "Unable to connect to MASTER: %s", err);
         connClose(mi->repl_transfer_s);
         mi->repl_transfer_s = NULL;
         return C_ERR;

src/scripting.cpp

@@ -1264,6 +1264,9 @@ void scriptingInit(int setup) {
 
     /* Finally set the table as 'redis' global var. */
     lua_setglobal(lua,"redis");
+    /* Set table as 'keydb' global var as well */
+    lua_getglobal(lua,"redis");
+    lua_setglobal(lua,"keydb");
 
     /* Replace math.random and math.randomseed with our implementations. */
     lua_getglobal(lua,"math");

src/server.cpp

@@ -2219,6 +2219,7 @@ void checkChildrenDone(void) {
             g_pserver->rdbThreadVars.fRdbThreadCancel = false;
             g_pserver->rdbThreadVars.fDone = false;
             if (exitcode == 0) receiveChildInfo();
+            closeChildInfoPipe();
         }
     }
     else if ((pid = waitpid(-1, &statloc, WNOHANG)) != 0) {
@@ -5171,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. */
@@ -5471,30 +5472,30 @@ NULL
 
 /* Convert an amount of bytes into a human readable string in the form
  * of 100B, 2G, 100M, 4K, and so forth. */
-void bytesToHuman(char *s, unsigned long long n) {
+void bytesToHuman(char *s, unsigned long long n, size_t bufsize) {
     double d;
 
     if (n < 1024) {
         /* Bytes */
-        snprintf(s,256,"%lluB",n);
+        snprintf(s,bufsize,"%lluB",n);
     } else if (n < (1024*1024)) {
         d = (double)n/(1024);
-        snprintf(s,256,"%.2fK",d);
+        snprintf(s,bufsize,"%.2fK",d);
     } else if (n < (1024LL*1024*1024)) {
         d = (double)n/(1024*1024);
-        snprintf(s,256,"%.2fM",d);
+        snprintf(s,bufsize,"%.2fM",d);
     } else if (n < (1024LL*1024*1024*1024)) {
         d = (double)n/(1024LL*1024*1024);
-        snprintf(s,256,"%.2fG",d);
+        snprintf(s,bufsize,"%.2fG",d);
     } else if (n < (1024LL*1024*1024*1024*1024)) {
         d = (double)n/(1024LL*1024*1024*1024);
-        snprintf(s,256,"%.2fT",d);
+        snprintf(s,bufsize,"%.2fT",d);
     } else if (n < (1024LL*1024*1024*1024*1024*1024)) {
         d = (double)n/(1024LL*1024*1024*1024*1024);
-        snprintf(s,256,"%.2fP",d);
+        snprintf(s,bufsize,"%.2fP",d);
     } else {
         /* Let's hope we never need this */
-        snprintf(s,256,"%lluB",n);
+        snprintf(s,bufsize,"%lluB",n);
     }
 }
 
@@ -5670,13 +5671,13 @@ sds genRedisInfoString(const char *section) {
         if (zmalloc_used > g_pserver->stat_peak_memory)
             g_pserver->stat_peak_memory = zmalloc_used;
 
-        bytesToHuman(hmem,zmalloc_used);
-        bytesToHuman(peak_hmem,g_pserver->stat_peak_memory);
-        bytesToHuman(total_system_hmem,total_system_mem);
-        bytesToHuman(used_memory_lua_hmem,memory_lua);
-        bytesToHuman(used_memory_scripts_hmem,mh->lua_caches);
-        bytesToHuman(used_memory_rss_hmem,g_pserver->cron_malloc_stats.process_rss);
-        bytesToHuman(maxmemory_hmem,g_pserver->maxmemory);
+        bytesToHuman(hmem,zmalloc_used,sizeof(hmem));
+        bytesToHuman(peak_hmem,g_pserver->stat_peak_memory,sizeof(peak_hmem));
+        bytesToHuman(total_system_hmem,total_system_mem,sizeof(total_system_hmem));
+        bytesToHuman(used_memory_lua_hmem,memory_lua,sizeof(used_memory_lua_hmem));
+        bytesToHuman(used_memory_scripts_hmem,mh->lua_caches,sizeof(used_memory_scripts_hmem));
+        bytesToHuman(used_memory_rss_hmem,g_pserver->cron_malloc_stats.process_rss,sizeof(used_memory_rss_hmem));
+        bytesToHuman(maxmemory_hmem,g_pserver->maxmemory,sizeof(maxmemory_hmem));
 
         if (sections++) info = sdscat(info,"\r\n");
         info = sdscatprintf(info,
@@ -6563,7 +6564,8 @@ void usage(void) {
 
 void redisAsciiArt(void) {
 #include "asciilogo.h"
-    char *buf = (char*)zmalloc(1024*16, MALLOC_LOCAL);
+    size_t bufsize = 1024*16;
+    char *buf = (char*)zmalloc(bufsize, MALLOC_LOCAL);
     const char *mode;
 
     if (g_pserver->cluster_enabled) mode = "cluster";
@@ -6585,7 +6587,7 @@ void redisAsciiArt(void) {
         );
     } else {
         sds motd = fetchMOTD(true, cserver.enable_motd);
-        snprintf(buf,1024*16,ascii_logo,
+        snprintf(buf,bufsize,ascii_logo,
             KEYDB_REAL_VERSION,
             redisGitSHA1(),
             strtol(redisGitDirty(),NULL,10) > 0,
@@ -6888,6 +6890,7 @@ int redisFork(int purpose) {
     latencyAddSampleIfNeeded("fork-lock",(ustime()-startWriteLock)/1000);
     if ((childpid = fork()) == 0) {
         /* Child */
+        aeForkLockInChild();
         aeReleaseForkLock();
         g_pserver->in_fork_child = purpose;
         setOOMScoreAdj(CONFIG_OOM_BGCHILD);

src/server.h

@@ -1204,12 +1204,12 @@ public:
     void disableKeyCache();
     bool keycacheIsEnabled();
 
-    bool prefetchKeysAsync(client *c, struct parsed_command &command, bool fExecOK);
+    void prefetchKeysAsync(client *c, struct parsed_command &command);
 
     bool FSnapshot() const { return m_spdbSnapshotHOLDER != nullptr; }
 
     std::unique_ptr<const StorageCache> CloneStorageCache() { return std::unique_ptr<const StorageCache>(m_spstorage->clone()); }
-    void bulkStorageInsert(char **rgKeys, size_t *rgcbKeys, char **rgVals, size_t *rgcbVals, size_t celem);
+    void bulkDirectStorageInsert(char **rgKeys, size_t *rgcbKeys, char **rgVals, size_t *rgcbVals, size_t celem);
 
     dict_iter find_cached_threadsafe(const char *key) const;
 
@@ -1370,7 +1370,7 @@ struct redisDb : public redisDbPersistentDataSnapshot
     using redisDbPersistentData::FRehashing;
     using redisDbPersistentData::FTrackingChanges;
     using redisDbPersistentData::CloneStorageCache;
-    using redisDbPersistentData::bulkStorageInsert;
+    using redisDbPersistentData::bulkDirectStorageInsert;
 
 public:
     const redisDbPersistentDataSnapshot *createSnapshot(uint64_t mvccCheckpoint, bool fOptional) {
@@ -2729,9 +2729,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];
@@ -3894,13 +3898,13 @@ void incrementMvccTstamp();
 
 #if __GNUC__ >= 7 && !defined(NO_DEPRECATE_FREE)
  [[deprecated]]
-void *calloc(size_t count, size_t size);
+void *calloc(size_t count, size_t size) noexcept;
  [[deprecated]]
-void free(void *ptr);
+void free(void *ptr) noexcept;
  [[deprecated]]
-void *malloc(size_t size);
+void *malloc(size_t size) noexcept;
  [[deprecated]]
-void *realloc(void *ptr, size_t size);
+void *realloc(void *ptr, size_t size) noexcept;
 #endif
 
 /* Debugging stuff */

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

src/tls.cpp

@@ -616,8 +616,9 @@ bool tlsValidateCertificateName(tls_connection* conn){
     /* If neither the CN nor the SANs match, update the SSL error and return false */
     conn->c.last_errno = 0;
     if (conn->ssl_error) zfree(conn->ssl_error);
-    conn->ssl_error = (char*)zmalloc(512);
-    snprintf(conn->ssl_error, 512, "Client CN (%s) and SANs not found in allowlist.", commonName);
+    size_t bufsize = 512;
+    conn->ssl_error = (char*)zmalloc(bufsize);
+    snprintf(conn->ssl_error, bufsize, "Client CN (%s) and SANs not found in allowlist.", commonName);
     return false;
 }
 

src/ziplist.c

@@ -1704,17 +1704,17 @@ static unsigned char *createIntList() {
     unsigned char *zl = ziplistNew();
     char buf[32];
 
-    snprintf(buf, 32, "100");
+    snprintf(buf, sizeof(buf), "100");
     zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_TAIL);
-    snprintf(buf, 32, "128000");
+    snprintf(buf, sizeof(buf), "128000");
     zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_TAIL);
-    snprintf(buf, 32, "-100");
+    snprintf(buf, sizeof(buf), "-100");
     zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_HEAD);
-    snprintf(buf, 32, "4294967296");
+    snprintf(buf, sizeof(buf), "4294967296");
     zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_HEAD);
-    snprintf(buf, 32, "non integer");
+    snprintf(buf, sizeof(buf), "non integer");
     zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_TAIL);
-    snprintf(buf, 32, "much much longer non integer");
+    snprintf(buf, sizeof(buf), "much much longer non integer");
     zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_TAIL);
     return zl;
 }
@@ -2228,7 +2228,7 @@ int ziplistTest(int argc, char **argv, int accurate) {
         char buf[32];
         int i,len;
         for (i = 0; i < 1000; i++) {
-            len = snprintf(buf,32,"%d",i);
+            len = snprintf(buf,sizeof(buf),"%d",i);
             zl = ziplistPush(zl,(unsigned char*)buf,len,ZIPLIST_TAIL);
         }
         for (i = 0; i < 1000; i++) {
@@ -2375,13 +2375,13 @@ int ziplistTest(int argc, char **argv, int accurate) {
                 } else {
                     switch(rand() % 3) {
                     case 0:
-                        buflen = snprintf(buf,1024,"%lld",(0LL + rand()) >> 20);
+                        buflen = snprintf(buf,sizeof(buf),"%lld",(0LL + rand()) >> 20);
                         break;
                     case 1:
-                        buflen = snprintf(buf,1024,"%lld",(0LL + rand()));
+                        buflen = snprintf(buf,sizeof(buf),"%lld",(0LL + rand()));
                         break;
                     case 2:
-                        buflen = snprintf(buf,1024,"%lld",(0LL + rand()) << 20);
+                        buflen = snprintf(buf,sizeof(buf),"%lld",(0LL + rand()) << 20);
                         break;
                     default:
                         assert(NULL);
@@ -2410,7 +2410,7 @@ int ziplistTest(int argc, char **argv, int accurate) {
 
                 assert(ziplistGet(p,&sstr,&slen,&sval));
                 if (sstr == NULL) {
-                    buflen = snprintf(buf,1024,"%lld",sval);
+                    buflen = snprintf(buf,sizeof(buf),"%lld",sval);
                 } else {
                     buflen = slen;
                     memcpy(buf,sstr,buflen);

src/zmalloc.cpp

@@ -80,12 +80,12 @@ static_assert((PREFIX_SIZE % 16) == 0, "Our prefix must be modulo 16-bytes or ou
 #define realloc(ptr,size, type) tc_realloc(ptr,size)
 #define free(ptr) tc_free(ptr)
 #elif defined(USE_JEMALLOC)
-#define malloc(size, type) je_malloc(size)
-#define calloc(count,size,type) je_calloc(count,size)
-#define realloc(ptr,size,type) je_realloc(ptr,size)
-#define free(ptr) je_free(ptr)
-#define mallocx(size,flags) je_mallocx(size,flags)
-#define dallocx(ptr,flags) je_dallocx(ptr,flags)
+#define malloc(size, type) malloc(size)
+#define calloc(count,size,type) calloc(count,size)
+#define realloc(ptr,size,type) realloc(ptr,size)
+#define free(ptr) free(ptr)
+#define mallocx(size,flags) mallocx(size,flags)
+#define dallocx(ptr,flags) dallocx(ptr,flags)
 #else
 #define malloc(size, type) malloc(size)
 #define calloc(count,size,type) calloc(count,size)
@@ -370,7 +370,7 @@ size_t zmalloc_get_rss(void) {
     int fd, count;
     char *p, *x;
 
-    snprintf(filename,256,"/proc/%ld/stat",(long) getpid());
+    snprintf(filename,sizeof(filename),"/proc/%ld/stat",(long) getpid());
     if ((fd = open(filename,O_RDONLY)) == -1) return 0;
     if (read(fd,buf,4096) <= 0) {
         close(fd);
@@ -462,7 +462,7 @@ size_t zmalloc_get_rss(void) {
     char filename[256];
     int fd;
 
-    snprintf(filename,256,"/proc/%ld/psinfo",(long) getpid());
+    snprintf(filename,sizeof(filename),"/proc/%ld/psinfo",(long) getpid());
 
     if ((fd = open(filename,O_RDONLY)) == -1) return 0;
     if (ioctl(fd, PIOCPSINFO, &info) == -1) {
@@ -494,17 +494,17 @@ int zmalloc_get_allocator_info(size_t *allocated,
     *allocated = *resident = *active = 0;
     /* Update the statistics cached by mallctl. */
     sz = sizeof(epoch);
-    je_mallctl("epoch", &epoch, &sz, &epoch, sz);
+    mallctl("epoch", &epoch, &sz, &epoch, sz);
     sz = sizeof(size_t);
     /* Unlike RSS, this does not include RSS from shared libraries and other non
      * heap mappings. */
-    je_mallctl("stats.resident", resident, &sz, NULL, 0);
+    mallctl("stats.resident", resident, &sz, NULL, 0);
     /* Unlike resident, this doesn't not include the pages jemalloc reserves
      * for re-use (purge will clean that). */
-    je_mallctl("stats.active", active, &sz, NULL, 0);
+    mallctl("stats.active", active, &sz, NULL, 0);
     /* Unlike zmalloc_used_memory, this matches the stats.resident by taking
      * into account all allocations done by this process (not only zmalloc). */
-    je_mallctl("stats.allocated", allocated, &sz, NULL, 0);
+    mallctl("stats.allocated", allocated, &sz, NULL, 0);
     return 1;
 }
 
@@ -512,7 +512,7 @@ void set_jemalloc_bg_thread(int enable) {
     /* let jemalloc do purging asynchronously, required when there's no traffic 
      * after flushdb */
     char val = !!enable;
-    je_mallctl("background_thread", NULL, 0, &val, 1);
+    mallctl("background_thread", NULL, 0, &val, 1);
 }
 
 int jemalloc_purge() {
@@ -520,9 +520,9 @@ int jemalloc_purge() {
     char tmp[32];
     unsigned narenas = 0;
     size_t sz = sizeof(unsigned);
-    if (!je_mallctl("arenas.narenas", &narenas, &sz, NULL, 0)) {
-        snprintf(tmp, 32, "arena.%d.purge", narenas);
-        if (!je_mallctl(tmp, NULL, 0, NULL, 0))
+    if (!mallctl("arenas.narenas", &narenas, &sz, NULL, 0)) {
+        snprintf(tmp, sizeof(tmp), "arena.%d.purge", narenas);
+        if (!mallctl(tmp, NULL, 0, NULL, 0))
             return 0;
     }
     return -1;

src/zmalloc.h

@@ -55,7 +55,7 @@
 #include <jemalloc/jemalloc.h>
 #if (JEMALLOC_VERSION_MAJOR == 2 && JEMALLOC_VERSION_MINOR >= 1) || (JEMALLOC_VERSION_MAJOR > 2)
 #define HAVE_MALLOC_SIZE 1
-#define zmalloc_size(p) je_malloc_usable_size(p)
+#define zmalloc_size(p) malloc_usable_size(p)
 #else
 #error "Newer version of jemalloc required"
 #endif

tests/unit/introspection.tcl

@@ -162,15 +162,16 @@ start_server {tags {"introspection"}} {
             aof_rewrite_cpulist
             time-thread-priority
             bgsave_cpulist
-	        storage-cache-mode
-	        storage-provider-options
-	        use-fork
+	    storage-cache-mode
+	    storage-provider-options
+	    use-fork
             multi-master
             active-replica
             bind
             set-proc-title
             repl-backlog-disk-reserve
 	    tls-allowlist
+            db-s3-object
         }
 
         if {!$::tls} {

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

@@ -43,10 +43,15 @@ start_server {tags {"scripting"}} {
         r eval {return redis.call('get',KEYS[1])} 1 mykey
     } {myval}
 
-    test { EVAL - keys command works? } {
+    test {EVAL - keys command works? } {
         r eval {return redis.call('keys', 'test')} 0
     }
 
+    test {EVAL - KeyDB global works } {
+        r eval {return keydb.call('get', KEYS[1])} 1 mykey
+        assert_equal [r eval {return redis.call('get',KEYS[1])} 1 mykey] [r eval {return keydb.call('get', KEYS[1])} 1 mykey]
+    }
+
     test {EVALSHA - Can we call a SHA1 if already defined?} {
         r evalsha fd758d1589d044dd850a6f05d52f2eefd27f033f 1 mykey
     } {myval}

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