update slotToKey during fast load

* close tempfile when done bgsave

* declare getTempFileName

* open close child pipes in correct place

* assert no existing child pipe when opening new child pipe

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

@@ -3002,8 +3002,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);
 }
 
@@ -3265,7 +3273,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 +3285,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 +3305,7 @@ bool redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
             }
         }
 #endif
-        return false;
+        return;
     }
 
     AeLocker lock;
@@ -3307,10 +3315,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 +3351,6 @@ bool redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
         }
     }
 
-    bool fNoInsert = false;
     if (!vecInserts.empty()) {
         lock.arm(c);
         for (auto &tuple : vecInserts)
@@ -3359,13 +3366,11 @@ 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;
                         }
                     }
@@ -3392,12 +3397,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/networking.cpp

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

src/rdb.cpp

@@ -1490,7 +1490,7 @@ int rdbSaveFile(char *filename, const redisDbPersistentDataSnapshot **rgpdb, rdb
     rio rdb;
     int error = 0;
 
-    snprintf(tmpfile,sizeof(tmpfile),"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));

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/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) {

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) {
@@ -3894,13 +3894,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/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)
@@ -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)) {
+    if (!mallctl("arenas.narenas", &narenas, &sz, NULL, 0)) {
         snprintf(tmp, sizeof(tmp), "arena.%d.purge", narenas);
-        if (!je_mallctl(tmp, NULL, 0, NULL, 0))
+        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