update clusterStorageLoadCallback

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 --disable-cxx CFLAGS="$(JEMALLOC_CFLAGS)" LDFLAGS="$(JEMALLOC_LDFLAGS)"
+	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 && $(MAKE) CFLAGS="$(JEMALLOC_CFLAGS)" LDFLAGS="$(JEMALLOC_LDFLAGS)" lib/libjemalloc.a
 
 .PHONY: jemalloc

src/IStorage.h

@@ -33,11 +33,10 @@ public:
     virtual bool enumerate(callback fn) const = 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/Makefile

@@ -25,7 +25,7 @@ NODEPS:=clean distclean
 
 # Default settings
 STD=-pedantic -DREDIS_STATIC=''
-CXX_STD=-std=c++17 -pedantic -fno-rtti -D__STDC_FORMAT_MACROS  
+CXX_STD=-std=c++14 -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]->bulkDirectStorageInsert(vecqueuedKeys.data(), vecqueuedKeysCb.data(), vecqueuedVals.data(), vecqueuedValsCb.data(), vecqueuedKeys.size());
+            g_pserver->db[idb]->bulkStorageInsert(vecqueuedKeys.data(), vecqueuedKeysCb.data(), vecqueuedVals.data(), vecqueuedValsCb.data(), vecqueuedKeys.size());
             --insertsInFlightTmp;
             delete pallocator;
         });

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();

src/childinfo.cpp

@@ -42,7 +42,6 @@ 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/config.cpp

@@ -35,7 +35,6 @@
 
 #include <fcntl.h>
 #include <sys/stat.h>
-#include <sys/wait.h>
 #ifdef __linux__
 #include <sys/sysinfo.h>
 #endif
@@ -2457,32 +2456,6 @@ 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)) {
@@ -2830,7 +2803,6 @@ 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),

src/db.cpp

@@ -53,7 +53,7 @@ struct dbBackup {
  *----------------------------------------------------------------------------*/
 
 int expireIfNeeded(redisDb *db, robj *key, robj *o);
-void slotToKeyUpdateKeyCore(const char *key, size_t keylen, int add);
+void slotToKeyUpdateKeyCore(const char *key, size_t keylen, int add, int noModify);
 
 std::unique_ptr<expireEntry> deserializeExpire(sds key, const char *str, size_t cch, size_t *poffset);
 sds serializeStoredObjectAndExpire(redisDbPersistentData *db, const char *key, robj_roptr o);
@@ -2471,11 +2471,11 @@ int xreadGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult
  * a fast way a key that belongs to a specified hash slot. This is useful
  * while rehashing the cluster and in other conditions when we need to
  * understand if we have keys for a given hash slot. */
-void slotToKeyUpdateKey(sds key, int add) {
-    slotToKeyUpdateKeyCore(key, sdslen(key), add);
+void slotToKeyUpdateKey(sds key, int add, int noModify) {
+    slotToKeyUpdateKeyCore(key, sdslen(key), add, noModify);
 }
 
-void slotToKeyUpdateKeyCore(const char *key, size_t keylen, int add) {
+void slotToKeyUpdateKeyCore(const char *key, size_t keylen, int add, int noModify) {
     serverAssert(GlobalLocksAcquired());
 
     unsigned int hashslot = keyHashSlot(key,keylen);
@@ -2489,23 +2489,28 @@ void slotToKeyUpdateKeyCore(const char *key, size_t keylen, int add) {
     memcpy(indexed+2,key,keylen);
     int fModified = false;
     if (add) {
-        fModified = raxInsert(g_pserver->cluster->slots_to_keys,indexed,keylen+2,NULL,NULL);
+        if (noModify) fModified = raxTryInsert(g_pserver->cluster->slots_to_keys,indexed,keylen+2,NULL,NULL);
+        else fModified = raxInsert(g_pserver->cluster->slots_to_keys,indexed,keylen+2,NULL,NULL);
     } else {
         fModified = raxRemove(g_pserver->cluster->slots_to_keys,indexed,keylen+2,NULL);
     }
     // This assert is disabled when a snapshot depth is >0 because prepOverwriteForSnapshot will add in a tombstone,
     //  this prevents ensure from adding the key to the dictionary which means the caller isn't aware we're already tracking
     //  the key.
-    serverAssert(fModified || g_pserver->db[0]->snapshot_depth() > 0);
+    serverAssert(fModified || g_pserver->db[0]->snapshot_depth() > 0 || noModify);
     if (indexed != buf) zfree(indexed);
 }
 
+void slotToKeyTryAdd(sds key) {
+    slotToKeyUpdateKey(key,1,1);
+}
+
 void slotToKeyAdd(sds key) {
-    slotToKeyUpdateKey(key,1);
+    slotToKeyUpdateKey(key,1,0);
 }
 
 void slotToKeyDel(sds key) {
-    slotToKeyUpdateKey(key,0);
+    slotToKeyUpdateKey(key,0,0);
 }
 
 /* Release the radix tree mapping Redis Cluster keys to slots. If 'async'
@@ -2604,18 +2609,15 @@ void redisDbPersistentData::endStorageProvider()
 
 void clusterStorageLoadCallback(const char *rgchkey, size_t cch, void *)
 {
-    slotToKeyUpdateKeyCore(rgchkey, cch, true /*add*/);
+    slotToKeyUpdateKeyCore(rgchkey, cch, true /*add*/, false /*noModify*/);
 }
 
-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 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 redisDb::initialize(int id)
@@ -2636,7 +2638,7 @@ void redisDb::storageProviderInitialize()
 {
     if (g_pserver->m_pstorageFactory != nullptr)
     {
-        IStorageFactory::key_load_iterator itr = g_pserver->cluster_enabled ? moduleClusterLoadCallback : moduleLoadCallback;
+        IStorageFactory::key_load_iterator itr = moduleLoadCallback;
         this->setStorageProvider(StorageCache::create(g_pserver->m_pstorageFactory, id, itr, &id));
     }
 }
@@ -2861,6 +2863,9 @@ LNotFound:
             {
                 dictAdd(m_pdict, sdsNewKey, o);
                 o->SetFExpires(spexpire != nullptr);
+                if (g_pserver->cluster_enabled) {
+                    slotToKeyTryAdd(sdsNewKey);
+                }
 
                 if (spexpire != nullptr)
                 {
@@ -2908,13 +2913,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 +2964,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 +3001,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);
@@ -3023,17 +3010,9 @@ void redisDbPersistentData::processChangesAsync(std::atomic<int> &pendingJobs)
     });
 }
 
-/* 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)
+void redisDbPersistentData::bulkStorageInsert(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, nullptr, celem);
+    m_spstorage->bulkInsert(rgKeys, rgcbKeys, rgVals, rgcbVals, celem);
 }
 
 void redisDbPersistentData::commitChanges(const redisDbPersistentDataSnapshot **psnapshotFree)
@@ -3042,30 +3021,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;

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=mallctl(szFromObj(argv[0]), &old, &sz, argc > 1? &val: NULL, argc > 1?sz: 0))) {
+        if ((ret=je_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=mallctl(szFromObj(argv[0]), NULL, 0, &val, sz))) {
+                if (!(ret=je_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=mallctl(szFromObj(argv[0]), &old, &sz, NULL, 0))) {
+    if ((rret=je_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 = mallctl(szFromObj(argv[0]), NULL, 0, valref, sz);
+        wret = je_mallctl(szFromObj(argv[0]), NULL, 0, valref, sz);
     }
     if (!rret)
         addReplyBulkCString(c, old);

src/new.cpp

@@ -41,3 +41,10 @@ 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();
-        malloc_stats_print(inputCatSds, &info, NULL);
+        je_malloc_stats_print(inputCatSds, &info, NULL);
         addReplyVerbatim(c,info,sdslen(info),"txt");
         sdsfree(info);
 #else

src/rdb.cpp

@@ -1490,7 +1490,7 @@ int rdbSaveFile(char *filename, const redisDbPersistentDataSnapshot **rgpdb, rdb
     rio rdb;
     int error = 0;
 
-    getTempFileName(tmpfile, g_pserver->rdbThreadVars.tmpfileNum);
+    snprintf(tmpfile,sizeof(tmpfile),"temp-%d-%d.rdb", getpid(), g_pserver->rdbThreadVars.tmpfileNum);
     fp = fopen(tmpfile,"w");
     if (!fp) {
         char *cwdp = getcwd(cwd,MAXPATHLEN);
@@ -1660,14 +1660,12 @@ 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);
@@ -1685,11 +1683,13 @@ 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));

src/rdb.h

@@ -151,7 +151,6 @@ 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/replication.cpp

@@ -3797,11 +3797,9 @@ 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) {
-        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);
+        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));
         connClose(mi->repl_transfer_s);
         mi->repl_transfer_s = NULL;
         return C_ERR;

src/scripting.cpp

@@ -1264,9 +1264,6 @@ 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.h

@@ -1209,7 +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()); }
-    void bulkDirectStorageInsert(char **rgKeys, size_t *rgcbKeys, char **rgVals, size_t *rgcbVals, size_t celem);
+    void bulkStorageInsert(char **rgKeys, size_t *rgcbKeys, char **rgVals, size_t *rgcbVals, size_t celem);
 
     dict_iter find_cached_threadsafe(const char *key) const;
 
@@ -1217,7 +1217,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);
@@ -1370,7 +1370,7 @@ struct redisDb : public redisDbPersistentDataSnapshot
     using redisDbPersistentData::FRehashing;
     using redisDbPersistentData::FTrackingChanges;
     using redisDbPersistentData::CloneStorageCache;
-    using redisDbPersistentData::bulkDirectStorageInsert;
+    using redisDbPersistentData::bulkStorageInsert;
 
 public:
     const redisDbPersistentDataSnapshot *createSnapshot(uint64_t mvccCheckpoint, bool fOptional) {
@@ -3894,13 +3894,13 @@ void incrementMvccTstamp();
 
 #if __GNUC__ >= 7 && !defined(NO_DEPRECATE_FREE)
  [[deprecated]]
-void *calloc(size_t count, size_t size) noexcept;
+void *calloc(size_t count, size_t size);
  [[deprecated]]
-void free(void *ptr) noexcept;
+void free(void *ptr);
  [[deprecated]]
-void *malloc(size_t size) noexcept;
+void *malloc(size_t size);
  [[deprecated]]
-void *realloc(void *ptr, size_t size) noexcept;
+void *realloc(void *ptr, size_t size);
 #endif
 
 /* Debugging stuff */

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)

src/storage/rocksdb.h

@@ -38,7 +38,7 @@ public:
     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/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) 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)
+#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)
 #else
 #define malloc(size, type) malloc(size)
 #define calloc(count,size,type) calloc(count,size)
@@ -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);
-    mallctl("epoch", &epoch, &sz, &epoch, sz);
+    je_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. */
-    mallctl("stats.resident", resident, &sz, NULL, 0);
+    je_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). */
-    mallctl("stats.active", active, &sz, NULL, 0);
+    je_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). */
-    mallctl("stats.allocated", allocated, &sz, NULL, 0);
+    je_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;
-    mallctl("background_thread", NULL, 0, &val, 1);
+    je_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 (!mallctl("arenas.narenas", &narenas, &sz, NULL, 0)) {
+    if (!je_mallctl("arenas.narenas", &narenas, &sz, NULL, 0)) {
         snprintf(tmp, sizeof(tmp), "arena.%d.purge", narenas);
-        if (!mallctl(tmp, NULL, 0, NULL, 0))
+        if (!je_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) malloc_usable_size(p)
+#define zmalloc_size(p) je_malloc_usable_size(p)
 #else
 #error "Newer version of jemalloc required"
 #endif

tests/unit/introspection.tcl

@@ -162,16 +162,15 @@ 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/scripting.tcl

@@ -43,15 +43,10 @@ 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}