init iteration

pkg/docker/Dockerfile

@@ -53,7 +53,7 @@ RUN set -eux; \
         grep -E '^ *createBoolConfig[(]"protected-mode",.*, *1 *,.*[)],$' ./src/config.cpp; \
         sed -ri 's!^( *createBoolConfig[(]"protected-mode",.*, *)1( *,.*[)],)$!\10\2!' ./src/config.cpp; \
         grep -E '^ *createBoolConfig[(]"protected-mode",.*, *0 *,.*[)],$' ./src/config.cpp; \
-        make -j$(nproc) BUILD_TLS=yes ENABLE_FLASH=yes; \
+        make -j$(nproc) BUILD_TLS=yes; \
         cd src; \
         strip keydb-cli keydb-benchmark keydb-check-rdb keydb-check-aof keydb-diagnostic-tool keydb-sentinel keydb-server; \
         mv keydb-server keydb-cli keydb-benchmark keydb-check-rdb keydb-check-aof keydb-diagnostic-tool keydb-sentinel /usr/local/bin/; \
@@ -83,7 +83,6 @@ RUN \
         sed -i 's/^\(logfile .*\)$/# \1/' /etc/keydb/keydb.conf; \
         sed -i 's/protected-mode yes/protected-mode no/g' /etc/keydb/keydb.conf; \
         sed -i 's/^\(bind .*\)$/# \1/' /etc/keydb/keydb.conf; \
-        cd /usr/local/bin; \
         ln -s keydb-cli redis-cli; \
         cd /etc/keydb; \
         ln -s keydb.conf redis.conf; \

pkg/docker/Dockerfile_Alpine

@@ -5,7 +5,7 @@ RUN mkdir -p /etc/keydb
 ARG BRANCH
 RUN set -eux; \
         \
-        apk add --no-cache su-exec tini; \
+        apk add --no-cache su-exec; \
         apk add --no-cache --virtual .build-deps \
                 coreutils \
                 gcc \
@@ -32,7 +32,7 @@ RUN set -eux; \
         grep -E '^ *createBoolConfig[(]"protected-mode",.*, *1 *,.*[)],$' ./src/config.cpp; \
         sed -ri 's!^( *createBoolConfig[(]"protected-mode",.*, *)1( *,.*[)],)$!\10\2!' ./src/config.cpp; \
         grep -E '^ *createBoolConfig[(]"protected-mode",.*, *0 *,.*[)],$' ./src/config.cpp; \
-        make -j$(nproc) BUILD_TLS=yes ENABLE_FLASH=yes; \
+        make -j$(nproc) BUILD_TLS=yes; \
         cd src; \
         strip keydb-cli keydb-benchmark keydb-check-rdb keydb-check-aof keydb-diagnostic-tool keydb-sentinel keydb-server; \
         mv keydb-server keydb-cli keydb-benchmark keydb-check-rdb keydb-check-aof keydb-diagnostic-tool keydb-sentinel /usr/local/bin/; \
@@ -54,7 +54,6 @@ RUN set -eux; \
         sed -i 's/^\(logfile .*\)$/# \1/' /etc/keydb/keydb.conf; \
         sed -i 's/protected-mode yes/protected-mode no/g' /etc/keydb/keydb.conf; \
         sed -i 's/^\(bind .*\)$/# \1/' /etc/keydb/keydb.conf; \
-        cd /usr/local/bin; \
         ln -s keydb-cli redis-cli; \
         cd /etc/keydb; \
         ln -s keydb.conf redis.conf; \
@@ -77,6 +76,6 @@ RUN set -eux; \
         chmod +x /usr/local/bin/docker-entrypoint.sh
 VOLUME /data
 WORKDIR /data
-ENTRYPOINT ["tini", "--", "docker-entrypoint.sh"]
+ENTRYPOINT ["docker-entrypoint.sh"]
 EXPOSE 6379
 CMD ["keydb-server", "/etc/keydb/keydb.conf"]

src/IStorage.h

@@ -44,7 +44,7 @@ public:
         endWriteBatch();
     }
 
-    virtual std::vector<std::string> getExpirationCandidates(unsigned int count) = 0;
+    virtual std::vector<std::string> getExpirationCandidates(unsigned int count, long long time) = 0;
     virtual std::vector<std::string> getEvictionCandidates(unsigned int count) = 0;
     virtual void setExpire(const char *key, size_t cchKey, long long expire) = 0;
     virtual void removeExpire(const char *key, size_t cchKey, long long expire) = 0;

src/StorageCache.h

@@ -51,7 +51,7 @@ public:
     bool enumerate(IStorage::callback fn) const { return m_spstorage->enumerate(fn); }
     bool enumerate_hashslot(IStorage::callback fn, unsigned int hashslot) const { return m_spstorage->enumerate_hashslot(fn, hashslot); }
 
-    std::vector<std::string> getExpirationCandidates(unsigned int count) { return m_spstorage->getExpirationCandidates(count); }
+    std::vector<std::string> getExpirationCandidates(unsigned int count, long long time) { return m_spstorage->getExpirationCandidates(count, time); }
     std::vector<std::string> getEvictionCandidates(unsigned int count) { return m_spstorage->getEvictionCandidates(count); }
     void setExpire(const char *key, size_t cchKey, long long expire) { m_spstorage->setExpire(key, cchKey, expire); }
     void removeExpire(const char *key, size_t cchKey, long long expire) { m_spstorage->removeExpire(key, cchKey, expire); }

src/ae.cpp

@@ -113,8 +113,6 @@ enum class AE_ASYNC_OP
     PostCppFunction,
     DeleteFileEvent,
     CreateFileEvent,
-    PostAsynDBFunction,
-
 };
 
 struct aeCommand
@@ -127,8 +125,6 @@ struct aeCommand
         aePostFunctionProc *proc;
         aeFileProc *fproc;
         std::function<void()> *pfn;
-        aePostFunctionTokenProc* tproc;
-
     };
     void *clientData;
 };
@@ -179,17 +175,6 @@ void aeProcessCmd(aeEventLoop *eventLoop, int fd, void *, int )
             (*cmd.pfn)();
 
             delete cmd.pfn;
-            break;
-        }
-        case AE_ASYNC_OP::PostAsynDBFunction:
-        {   //added to support async api IStorage
-           if (cmd.fLock && !ulock.owns_lock()) {
-                g_forkLock.releaseRead();
-                ulock.lock();
-                g_forkLock.acquireRead();
-            }  
-            ((aePostFunctionTokenProc*)cmd.tproc)(eventLoop,(StorageToken*)cmd.clientData);
-            break; 
         }
             break;
         }
@@ -271,20 +256,6 @@ int aePostFunction(aeEventLoop *eventLoop, aePostFunctionProc *proc, void *arg)
     return AE_OK;
 }
 
-int aePostFunction(aeEventLoop *eventLoop, aePostFunctionTokenProc *proc, StorageToken *token)
-{
-  //added to support async api IStorage
-    aeCommand cmd = {};
-    cmd.op = AE_ASYNC_OP::PostAsynDBFunction; 
-    cmd.tproc = proc;
-    cmd.clientData = (void*)token; 
-    cmd.fLock = true;
-    auto size = write(eventLoop->fdCmdWrite, &cmd, sizeof(cmd));
-    if (size != sizeof(cmd))
-        return AE_ERR;
-    return AE_OK;
-}
-
 int aePostFunction(aeEventLoop *eventLoop, std::function<void()> fn, bool fLock, bool fForceQueue)
 {
     if (eventLoop == g_eventLoopThisThread && !fForceQueue)
@@ -890,7 +861,7 @@ void aeReleaseForkLock()
 
 void aeForkLockInChild()
 {
-    g_forkLock.setMulti(false);
+    g_forkLock.setNotify(false);
 }
 
 int aeThreadOwnsLock()

src/ae.h

@@ -72,15 +72,13 @@ extern "C" {
 #define AE_NOTUSED(V) ((void) V)
 
 struct aeEventLoop;
+
 /* Types and data structures */
 typedef void aeFileProc(struct aeEventLoop *eventLoop, int fd, void *clientData, int mask);
 typedef int aeTimeProc(struct aeEventLoop *eventLoop, long long id, void *clientData);
 typedef void aeEventFinalizerProc(struct aeEventLoop *eventLoop, void *clientData);
 typedef void aeBeforeSleepProc(struct aeEventLoop *eventLoop);
 typedef void aePostFunctionProc(void *pvArgs);
-//added to support async api IStorage
-struct StorageToken;
-typedef void aePostFunctionTokenProc(struct aeEventLoop *el, struct StorageToken *token);
 
 /* File event structure */
 typedef struct aeFileEvent {
@@ -136,8 +134,6 @@ int aePostFunction(aeEventLoop *eventLoop, aePostFunctionProc *proc, void *arg);
 #ifdef __cplusplus
 }   // EXTERN C
 int aePostFunction(aeEventLoop *eventLoop, std::function<void()> fn, bool fLock = true, bool fForceQueue = false);
-//added to support async api IStorage
-int aePostFunction(aeEventLoop *eventLoop, aePostFunctionTokenProc *proc, StorageToken *token);
 extern "C" {
 #endif
 void aeDeleteEventLoop(aeEventLoop *eventLoop);

src/cluster.cpp

@@ -5991,9 +5991,7 @@ clusterNode *getNodeByQuery(client *c, struct redisCommand *cmd, robj **argv, in
                 /* If it is not the first key, make sure it is exactly
                  * the same key as the first we saw. */
                 if (!equalStringObjects(firstkey,thiskey)) {
-                    clusterNode* nThisKey = g_pserver->cluster->slots[thisslot];
-
-                    if ((slot != thisslot) && (nThisKey != n || migrating_slot || importing_slot || g_pserver->cluster->migrating_slots_to[slot] != nullptr || g_pserver->cluster->importing_slots_from[slot] != nullptr)) {
+                    if (slot != thisslot) {
                         /* Error: multiple keys from different slots. */
                         getKeysFreeResult(&result);
                         if (error_code)

src/config.cpp

@@ -2946,7 +2946,7 @@ standardConfig configs[] = {
     createLongLongConfig("latency-monitor-threshold", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, g_pserver->latency_monitor_threshold, 0, INTEGER_CONFIG, NULL, NULL),
     createLongLongConfig("proto-max-bulk-len", NULL, MODIFIABLE_CONFIG, 1024*1024, LLONG_MAX, g_pserver->proto_max_bulk_len, 512ll*1024*1024, MEMORY_CONFIG, NULL, NULL), /* Bulk request max size */
     createLongLongConfig("stream-node-max-entries", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, g_pserver->stream_node_max_entries, 100, INTEGER_CONFIG, NULL, NULL),
-    createLongLongConfig("repl-backlog-size", NULL, MODIFIABLE_CONFIG, 1, LLONG_MAX, g_pserver->repl_backlog_config_size, 1024*1024, MEMORY_CONFIG, NULL, updateReplBacklogSize), /* Default: 1mb */
+    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),
@@ -2982,7 +2982,6 @@ standardConfig configs[] = {
     createSDSConfig("availability-zone", NULL, MODIFIABLE_CONFIG, 0, g_pserver->sdsAvailabilityZone, "", NULL, NULL),
     createIntConfig("overload-protect-percent", NULL, MODIFIABLE_CONFIG, 0, 200, g_pserver->overload_protect_threshold, 0, INTEGER_CONFIG, NULL, NULL),
     createIntConfig("force-eviction-percent", NULL, MODIFIABLE_CONFIG, 0, 100, g_pserver->force_eviction_percent, 0, INTEGER_CONFIG, NULL, NULL),
-    createBoolConfig("enable-async-rehash", NULL, MODIFIABLE_CONFIG, g_pserver->enable_async_rehash, 1, NULL, NULL),
 
 #ifdef USE_OPENSSL
     createIntConfig("tls-port", NULL, MODIFIABLE_CONFIG, 0, 65535, g_pserver->tls_port, 0, INTEGER_CONFIG, NULL, updateTLSPort), /* TCP port. */

src/db.cpp

@@ -2703,6 +2703,11 @@ void redisDbPersistentData::prepOverwriteForSnapshot(char *key)
         auto itr = m_pdbSnapshot->find_cached_threadsafe(key);
         if (itr.key() != nullptr)
         {
+            if (itr.val()->FExpires()) {
+                // Note: I'm sure we could handle this, but its too risky at the moment.
+                //  There are known bugs doing this with expires
+                return;
+            }
             sds keyNew = sdsdupshared(itr.key());
             if (dictAdd(m_pdictTombstone, keyNew, (void*)dictHashKey(m_pdict, key)) != DICT_OK)
                 sdsfree(keyNew);
@@ -2740,13 +2745,6 @@ void redisDbPersistentData::clear(void(callback)(void*))
     if (m_spstorage != nullptr)
         m_spstorage->clear(callback);
     dictEmpty(m_pdictTombstone,callback);
-
-    // To avoid issues with async rehash we completly free the old dict and create a fresh one
-    dictRelease(m_pdict);
-    dictRelease(m_pdictTombstone);
-    m_pdict = dictCreate(&dbDictType, this);
-    m_pdictTombstone = dictCreate(&dbTombstoneDictType, this);
-
     m_pdbSnapshot = nullptr;
     m_numexpires = 0;
 }
@@ -2882,10 +2880,10 @@ LNotFound:
             {
                 dictAdd(m_pdict, sdsNewKey, o);
                 
-                o->SetFExpires(spexpire != nullptr);
                 if (spexpire != nullptr) {
                     o->expire = std::move(*spexpire);
                 }
+                o->SetFExpires(spexpire != nullptr);
                 g_pserver->stat_storage_provider_read_hits++;
             } else {
                 sdsfree(sdsNewKey);

src/debug.cpp

@@ -79,7 +79,6 @@ void printCrashReport(void);
 void bugReportEnd(int killViaSignal, int sig);
 void logStackTrace(void *eip, int uplevel);
 void getTempFileName(char tmpfile[], int tmpfileNum);
-bool initializeStorageProvider(const char **err);
 
 /* ================================= Debugging ============================== */
 
@@ -309,7 +308,7 @@ void computeDatasetDigest(unsigned char *final) {
         mixDigest(final,&aux,sizeof(aux));
 
         /* Iterate this DB writing every entry */
-        db->iterate_threadsafe([final](const char *key, robj_roptr o)->bool {
+        db->iterate_threadsafe([final, db](const char *key, robj_roptr o)->bool {
             unsigned char digest[20];
             robj *keyobj;
 
@@ -536,75 +535,30 @@ NULL
                 return;
             }
         }
-        
-        if (g_pserver->m_pstorageFactory) {
-            protectClient(c);
 
-            for (int idb = 0; idb < cserver.dbnum; ++idb) {
-                if (g_pserver->db[idb]->processChanges(false))
-                    g_pserver->db[idb]->commitChanges();
-                g_pserver->db[idb]->storageProviderDelete();
-            }
-            delete g_pserver->metadataDb;
-
-            if (flush) emptyDb(-1,EMPTYDB_NO_FLAGS,NULL);
-
-            delete g_pserver->m_pstorageFactory;
-
-            const char *err;
-            if (!initializeStorageProvider(&err))
-            {
-                serverLog(LL_WARNING, "Failed to initialize storage provider: %s",err);
-                exit(EXIT_FAILURE);
-            }
-
-            g_pserver->metadataDb = g_pserver->m_pstorageFactory->createMetadataDb();
-            for (int idb = 0; idb < cserver.dbnum; ++idb)
-            {
-                int dbid = idb;
-                std::string dbid_key = "db-" + std::to_string(idb);
-                g_pserver->metadataDb->retrieve(dbid_key.c_str(), dbid_key.length(), [&](const char *, size_t, const void *data, size_t){
-                    dbid = *(int*)data;
-                });
-                delete g_pserver->db[idb];
-                g_pserver->db[idb] = new (MALLOC_LOCAL) redisDb();
-                g_pserver->db[idb]->initialize(dbid);
-            }
-
-            moduleFireServerEvent(REDISMODULE_EVENT_LOADING, REDISMODULE_SUBEVENT_LOADING_FLASH_START, NULL);
-            for (int idb = 0; idb < cserver.dbnum; ++idb)
-            {
-                g_pserver->db[idb]->storageProviderInitialize();
-                g_pserver->db[idb]->trackChanges(false);
-            }
-            moduleFireServerEvent(REDISMODULE_EVENT_LOADING, REDISMODULE_SUBEVENT_LOADING_ENDED, NULL);
-            
-            unprotectClient(c);
-        } else {
-            /* The default behavior is to save the RDB file before loading
-            * it back. */
-            if (save) {
-                rdbSaveInfo rsi, *rsiptr;
-                rsiptr = rdbPopulateSaveInfo(&rsi);
-                if (rdbSave(nullptr, rsiptr) != C_OK) {
-                    addReply(c,shared.err);
-                    return;
-                }
-            }
-
-            /* The default behavior is to remove the current dataset from
-            * memory before loading the RDB file, however when MERGE is
-            * used together with NOFLUSH, we are able to merge two datasets. */
-            if (flush) emptyDb(-1,EMPTYDB_NO_FLAGS,NULL);
-
-            protectClient(c);
-            int ret = rdbLoadFile(g_pserver->rdb_filename,NULL,flags);
-            unprotectClient(c);
-            if (ret != C_OK) {
-                addReplyError(c,"Error trying to load the RDB dump");
+        /* The default behavior is to save the RDB file before loading
+         * it back. */
+        if (save) {
+            rdbSaveInfo rsi, *rsiptr;
+            rsiptr = rdbPopulateSaveInfo(&rsi);
+            if (rdbSave(nullptr, rsiptr) != C_OK) {
+                addReply(c,shared.err);
                 return;
             }
         }
+
+        /* The default behavior is to remove the current dataset from
+         * memory before loading the RDB file, however when MERGE is
+         * used together with NOFLUSH, we are able to merge two datasets. */
+        if (flush) emptyDb(-1,EMPTYDB_NO_FLAGS,NULL);
+
+        protectClient(c);
+        int ret = rdbLoadFile(g_pserver->rdb_filename,NULL,flags);
+        unprotectClient(c);
+        if (ret != C_OK) {
+            addReplyError(c,"Error trying to load the RDB dump");
+            return;
+        }
         serverLog(LL_WARNING,"DB reloaded by DEBUG RELOAD");
         addReply(c,shared.ok);
     } else if (!strcasecmp(szFromObj(c->argv[1]),"loadaof")) {

src/expire.cpp

@@ -518,13 +518,11 @@ void pexpireMemberAtCommand(client *c)
             } while (sampled == 0 ||
                     (expired*100/sampled) > config_cycle_acceptable_stale);
         } else {
-            long prev_expired;
-            long long now = mstime();
+            long long now;
             size_t tried = 0;
             std::vector<std::string> keys;
             do {
-                prev_expired = total_expired;
-                keys = db->getStorageCache()->getExpirationCandidates(ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP);
+                keys = db->getStorageCache()->getExpirationCandidates(ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP, now);
                 for (std::string key : keys) {
                     robj* keyobj = createStringObject(key.c_str(), key.size());
                     db->find(szFromObj(keyobj));
@@ -535,7 +533,7 @@ void pexpireMemberAtCommand(client *c)
                 }
                 total_sampled += keys.size();
                 elapsed = ustime()-start;
-            } while (keys.size() > 0 && (elapsed < timelimit) && (total_expired - prev_expired) > 0);
+            } while (keys.size() > 0 && (elapsed < timelimit));
 
             if (ustime()-start > timelimit) {
                 timelimit_exit = 1;

src/expire.h

@@ -160,8 +160,6 @@ public:
 
     expireEntry &operator=(expireEntry &&e)
     {
-        if (FFat())
-            delete pfatentry();
         s = e.s;
         e.s.m_whenAndPtrUnion = 0;
         e.s.fFat = false;
@@ -169,8 +167,6 @@ public:
     }
 
     expireEntry &operator=(expireEntry &e) {
-        if (FFat())
-            delete pfatentry();
         if (e.FFat()) {
             s.m_whenAndPtrUnion = reinterpret_cast<long long>(new expireEntryFat(*e.pfatentry()));
             s.fFat = true;

src/meminfo.cpp

@@ -19,7 +19,6 @@ static size_t getMemKey(std::string key) {
     }
     return 0;
 # else
-    (void)key;
     return 0;
 # endif
 }

src/module.cpp

@@ -690,9 +690,8 @@ void moduleHandlePropagationAfterCommandCallback(RedisModuleCtx *ctx) {
 }
 
 /* Free the context after the user function was called. */
-void moduleFreeContext(RedisModuleCtx *ctx, bool propogate) {
-    if (propogate)
-        moduleHandlePropagationAfterCommandCallback(ctx);
+void moduleFreeContext(RedisModuleCtx *ctx) {
+    moduleHandlePropagationAfterCommandCallback(ctx);
     autoMemoryCollect(ctx);
     poolAllocRelease(ctx);
     if (ctx->postponed_arrays) {
@@ -7058,7 +7057,7 @@ RedisModuleString *RM_DictPrev(RedisModuleCtx *ctx, RedisModuleDictIter *di, voi
 /* Compare the element currently pointed by the iterator to the specified
  * element given by key/keylen, according to the operator 'op' (the set of
  * valid operators are the same valid for RedisModule_DictIteratorStart).
- * If the comparison is successful the command returns REDISMODULE_OK
+ * If the comparision is successful the command returns REDISMODULE_OK
  * otherwise REDISMODULE_ERR is returned.
  *
  * This is useful when we want to just emit a lexicographical range, so

src/modules/keydb_modstatsd/Makefile

@@ -4,7 +4,7 @@ OBJECT_FILES := modmain.o
 MODSNAP_CXX_FLAGS := -std=gnu++14
 
 %.o: %.cpp
-	$(CXX) -o $@ -c $< $(MODULE_FLAGS) -I../../../deps/cpp-statsd-client/include $(MODSNAP_CXX_FLAGS) -g
+	$(CXX) -o $@ -c $< $(MODULE_FLAGS) -I../../../deps/cpp-statsd-client/include $(MODSNAP_CXX_FLAGS)
 
 modstatsd.so: $(OBJECT_FILES)
 	$(CXX) -shared $(OBJECT_FILES) -o modstatsd.so

src/modules/keydb_modstatsd/modmain.cpp

@@ -76,7 +76,7 @@ std::string m_strPrefix { "keydb" };
 const std::regex g_replica_or_db_info_regex { "^(slave|db)(\\d+)" };
 const char *g_string_counter_separator = "__";
 const uint64_t g_stats_buffer_size_bytes = 1600;
-std::string nodeName;
+utsname sysName;
 int unameResult;
 
 enum class StatsD_Type {
@@ -582,7 +582,7 @@ void event_cron_handler(struct RedisModuleCtx *ctx, RedisModuleEvent eid, uint64
         size_t startTime = ustime();
 
 #ifdef __linux__
-	/* Log CPU Usage */
+        /* Log CPU Usage */
 	static long long s_mscpuLast = 0;
 	struct rusage self_ru;
 	getrusage(RUSAGE_SELF, &self_ru);
@@ -591,7 +591,7 @@ void event_cron_handler(struct RedisModuleCtx *ctx, RedisModuleEvent eid, uint64
 			+ (self_ru.ru_stime.tv_sec * 1000) + (self_ru.ru_stime.tv_usec / 1000);
 
 
-	g_stats->gauge("cpu_load_perc", ((double)(mscpuCur - s_mscpuLast) / ((curTime - lastTime)*1000))*100, false /* prefixOnly */);   
+        g_stats->gauge("cpu_load_perc", (double)(mscpuCur - s_mscpuLast) / ((curTime - lastTime)*1000), true /* prefixOnly */);   
 	s_mscpuLast = mscpuCur;
 #endif
 
@@ -600,7 +600,7 @@ void event_cron_handler(struct RedisModuleCtx *ctx, RedisModuleEvent eid, uint64
 
         /* node name */
         if (unameResult == 0) {
-            g_stats->increment("node_name" + std::string(g_string_counter_separator) + nodeName);
+            g_stats->increment("node_name" + std::string(g_string_counter_separator) + sysName.nodename);
         }
 
         /* Log INFO Fields */
@@ -653,7 +653,7 @@ void event_cron_handler(struct RedisModuleCtx *ctx, RedisModuleEvent eid, uint64
         reply = RedisModule_Call(ctx, "dbsize", "");
         long long keys = RedisModule_CallReplyInteger(reply);
         RedisModule_FreeCallReply(reply);
-        g_stats->gauge("keys", keys, false /* prefixOnly */);
+        g_stats->gauge("keys", keys);
         RedisModule_Log(ctx, REDISMODULE_LOGLEVEL_DEBUG, "Emitting metric \"keys\": %llu", keys);
         g_stats->timing("emit_keys_metric_time_taken_us", ustime() - commandStartTime);
 
@@ -672,20 +672,17 @@ extern "C" int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv,
     RedisModule_AutoMemory(ctx);
     /* Use pod name if available*/
     const char *podName = getenv("POD_NAME");
-    utsname sysName;
     unameResult = uname(&sysName);
-    if (unameResult == 0) {
-        nodeName = std::string(sysName.nodename);
-        std::replace(nodeName.begin(), nodeName.end(), '.', '-');
-    }
     if (podName != nullptr) {
         m_strPrefix = podName;
-        std::replace(m_strPrefix.begin(), m_strPrefix.end(), '.', '-');
     }
-    else if (unameResult == 0) {
-        m_strPrefix = nodeName;
-        unameResult = 1;
+    else {
+        if (unameResult == 0) {
+            m_strPrefix = std::string(sysName.nodename);
+            unameResult = 1;
+        }
     }
+    std::replace(m_strPrefix.begin(), m_strPrefix.end(), '.', '-');
 
     for (int iarg = 0; iarg < argc; ++iarg) {
         size_t len = 0;

src/object.cpp

@@ -46,7 +46,6 @@ robj *createObject(int type, void *ptr) {
     char *oB = (char*)zcalloc(sizeof(robj)+mvccExtraBytes, MALLOC_SHARED);
     robj *o = reinterpret_cast<robj*>(oB + mvccExtraBytes);
     
-    new (o) redisObject;
     o->type = type;
     o->encoding = OBJ_ENCODING_RAW;
     o->m_ptr = ptr;
@@ -419,7 +418,6 @@ void decrRefCount(robj_roptr o) {
         case OBJ_NESTEDHASH: freeNestedHashObject(o); break;
         default: serverPanic("Unknown object type"); break;
         }
-        o->~redisObject();
         if (g_pserver->fActiveReplica) {
             zfree(reinterpret_cast<redisObjectExtended*>(o.unsafe_robjcast())-1);
         } else {

src/rdb.cpp

@@ -2542,7 +2542,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int *error, uint64_t mvcc_ts
          * encoding version in the lower 10 bits of the module ID. */
         void *ptr = mt->rdb_load(&io,moduleid&1023);
         if (io.ctx) {
-            moduleFreeContext(io.ctx, false /* propogate */);
+            moduleFreeContext(io.ctx);
             zfree(io.ctx);
         }
 

src/readwritelock.h

@@ -8,46 +8,38 @@ class readWriteLock {
     int m_readCount = 0;
     int m_writeCount = 0;
     bool m_writeWaiting = false;
-    bool m_multi = true;
+    bool m_notify = true;
 public:
     readWriteLock(const char *name) : m_readLock(name), m_writeLock(name) {}
 
     void acquireRead() {
-        std::unique_lock<fastlock> rm(m_readLock, std::defer_lock);
-        if (m_multi) {
-            rm.lock();
-            while (m_writeCount > 0 || m_writeWaiting)
-                m_cv.wait(rm);
-        }
+        std::unique_lock<fastlock> rm(m_readLock);
+        while (m_writeCount > 0 || m_writeWaiting)
+            m_cv.wait(rm);
         m_readCount++;
     }
     
     bool tryAcquireRead() {
         std::unique_lock<fastlock> rm(m_readLock, std::defer_lock);
-        if (m_multi) {
-            if (!rm.try_lock())
-                return false;
-            if (m_writeCount > 0 || m_writeWaiting)
-                return false;
-        }
+        if (!rm.try_lock())
+            return false;
+        if (m_writeCount > 0 || m_writeWaiting)
+            return false;
         m_readCount++;
         return true;
     }
 
     void acquireWrite(bool exclusive = true) {
-        std::unique_lock<fastlock> rm(m_readLock, std::defer_lock);
-        if (m_multi) {
-            rm.lock();
-            m_writeWaiting = true;
-            while (m_readCount > 0)
+        std::unique_lock<fastlock> rm(m_readLock);
+        m_writeWaiting = true;
+        while (m_readCount > 0)
+            m_cv.wait(rm);
+        if (exclusive) {
+            /* Another thread might have the write lock while we have the read lock
+               but won't be able to release it until they can acquire the read lock
+               so release the read lock and try again instead of waiting to avoid deadlock */
+            while(!m_writeLock.try_lock())
                 m_cv.wait(rm);
-            if (exclusive) {
-                /* Another thread might have the write lock while we have the read lock
-                but won't be able to release it until they can acquire the read lock
-                so release the read lock and try again instead of waiting to avoid deadlock */
-                while(!m_writeLock.try_lock())
-                    m_cv.wait(rm);
-            }
         }
         m_writeCount++;
         m_writeWaiting = false;
@@ -60,38 +52,32 @@ public:
 
     bool tryAcquireWrite(bool exclusive = true) {
         std::unique_lock<fastlock> rm(m_readLock, std::defer_lock);
-        if (m_multi) {
-            if (!rm.try_lock())
+        if (!rm.try_lock())
+            return false;
+        if (m_readCount > 0)
+            return false;
+        if (exclusive)
+            if (!m_writeLock.try_lock())
                 return false;
-            if (m_readCount > 0)
-                return false;
-            if (exclusive)
-                if (!m_writeLock.try_lock())
-                    return false;
-        }
         m_writeCount++;
         return true;
     }
 
     void releaseRead() {
-        std::unique_lock<fastlock> rm(m_readLock, std::defer_lock);
-        if (m_multi) {
-            rm.lock();
-            m_cv.notify_all();
-        }
+        std::unique_lock<fastlock> rm(m_readLock);
         m_readCount--;
+        if (m_notify)
+            m_cv.notify_all();
     }
 
     void releaseWrite(bool exclusive = true) {
-        std::unique_lock<fastlock> rm(m_readLock, std::defer_lock);
+        std::unique_lock<fastlock> rm(m_readLock);
         serverAssert(m_writeCount > 0);
-        if (m_multi) {
-            rm.lock();
-            if (exclusive)
-                m_writeLock.unlock();
-            m_cv.notify_all();
-        }
+        if (exclusive)
+            m_writeLock.unlock();
         m_writeCount--;
+        if (m_notify)
+            m_cv.notify_all();
     }
 
     void downgradeWrite(bool exclusive = true) {
@@ -99,8 +85,8 @@ public:
         acquireRead();
     }
 
-    void setMulti(bool multi) {
-        m_multi = multi;
+    void setNotify(bool notify) {
+        m_notify = notify;
     }
 
     bool hasReader() {

src/replication.cpp

@@ -188,8 +188,6 @@ int bg_unlink(const char *filename) {
 /* ---------------------------------- MASTER -------------------------------- */
 
 bool createDiskBacklog() {
-    if (g_pserver->repl_backlog_disk != nullptr)
-        return true;    // already exists
     // Lets create some disk backed pages and add them here
     std::string path = "./repl-backlog-temp" + std::to_string(gettid());
 #if (defined __APPLE__ || defined __FreeBSD__)
@@ -3735,7 +3733,7 @@ retry_connect:
     }
 
     /* Prepare a suitable temp file for bulk transfer */
-    if (!useDisklessLoad() && !mi->isRocksdbSnapshotRepl) {
+    if (!useDisklessLoad()) {
         while(maxtries--) {
             auto dt = std::chrono::system_clock::now().time_since_epoch();
             auto dtMillisecond = std::chrono::duration_cast<std::chrono::milliseconds>(dt);

src/server.cpp

@@ -2137,7 +2137,7 @@ void databasesCron(bool fMainThread) {
                 ::dict *dict = g_pserver->db[rehash_db]->dictUnsafeKeyOnly();
                 /* Are we async rehashing? And if so is it time to re-calibrate? */
                 /* The recalibration limit is a prime number to ensure balancing across threads */
-                if (g_pserver->enable_async_rehash && rehashes_per_ms > 0 && async_rehashes < 131 && !cserver.active_defrag_enabled && cserver.cthreads > 1 && dictSize(dict) > 2048 && dictIsRehashing(dict) && !g_pserver->loading && aeLockContention() > 1) {
+                if (rehashes_per_ms > 0 && async_rehashes < 131 && !cserver.active_defrag_enabled && cserver.cthreads > 1 && dictSize(dict) > 2048 && dictIsRehashing(dict) && !g_pserver->loading) {
                     serverTL->rehashCtl = dictRehashAsyncStart(dict, rehashes_per_ms * ((1000 / g_pserver->hz) / 10));  // Estimate 10% CPU time spent in lock contention
                     if (serverTL->rehashCtl)
                         ++async_rehashes;
@@ -5663,8 +5663,7 @@ sds genRedisInfoString(const char *section) {
             "lru_clock:%u\r\n"
             "executable:%s\r\n"
             "config_file:%s\r\n"
-            "availability_zone:%s\r\n"
-            "features:%s\r\n",
+            "availability_zone:%s\r\n",
             KEYDB_SET_VERSION,
             redisGitSHA1(),
             strtol(redisGitDirty(),NULL,10) > 0,
@@ -5691,8 +5690,7 @@ sds genRedisInfoString(const char *section) {
             lruclock,
             cserver.executable ? cserver.executable : "",
             cserver.configfile ? cserver.configfile : "",
-            g_pserver->sdsAvailabilityZone,
-            "cluster_mget");
+            g_pserver->sdsAvailabilityZone);
     }
 
     /* Clients */
@@ -7413,7 +7411,7 @@ static void validateConfiguration()
         exit(EXIT_FAILURE);
     }
 
-    g_pserver->repl_backlog_size = g_pserver->repl_backlog_config_size; // this is normally set in the update logic, but not on initial config
+    g_pserver->repl_backlog_config_size = g_pserver->repl_backlog_size; // this is normally set in the update logic, but not on initial config
 }
 
 int iAmMaster(void) {

src/server.h

@@ -966,7 +966,6 @@ struct redisObjectExtended {
 
 typedef struct redisObject {
     friend redisObject *createEmbeddedStringObject(const char *ptr, size_t len);
-    friend redisObject *createObject(int type, void *ptr);
 protected:
     redisObject() {}
 
@@ -2306,7 +2305,6 @@ struct redisServer {
     rax *errors;                /* Errors table */
     int activerehashing;        /* Incremental rehash in serverCron() */
     int active_defrag_running;  /* Active defragmentation running (holds current scan aggressiveness) */
-    int enable_async_rehash = 1;    /* Should we use the async rehash feature? */
     int cronloops;              /* Number of times the cron function run */
     char runid[CONFIG_RUN_ID_SIZE+1];  /* ID always different at every exec. */
     int sentinel_mode;          /* True if this instance is a Sentinel. */
@@ -2919,7 +2917,7 @@ int moduleGetCommandKeysViaAPI(struct redisCommand *cmd, robj **argv, int argc,
 moduleType *moduleTypeLookupModuleByID(uint64_t id);
 void moduleTypeNameByID(char *name, uint64_t moduleid);
 const char *moduleTypeModuleName(moduleType *mt);
-void moduleFreeContext(struct RedisModuleCtx *ctx, bool propogate = true);
+void moduleFreeContext(struct RedisModuleCtx *ctx);
 void unblockClientFromModule(client *c);
 void moduleHandleBlockedClients(int iel);
 void moduleBlockedClientTimedOut(client *c);

src/snapshot.cpp

@@ -587,12 +587,8 @@ bool redisDbPersistentDataSnapshot::iterate_threadsafe_core(std::function<bool(c
                 if (!fKeyOnly)
                 {
                     size_t offset = 0;
-                    std::unique_ptr<expireEntry> spexpire = deserializeExpire((const char*)data, cbData, &offset);
+                    deserializeExpire((const char*)data, cbData, &offset);
                     o = deserializeStoredObject(reinterpret_cast<const char*>(data)+offset, cbData-offset);
-                    o->SetFExpires(spexpire != nullptr);
-                    if (spexpire != nullptr) {
-                        o->expire = std::move(*spexpire);
-                    }
                 }
                 fContinue = fn(sdsKey, o);
                 if (o != nullptr)

src/storage/rocksdb.cpp

@@ -149,22 +149,11 @@ size_t RocksDBStorageProvider::clear()
 
     rocksdb::ColumnFamilyHandle *handle = nullptr;
     rocksdb::ColumnFamilyOptions cf_options(m_pfactory->RocksDbOptions());
-    cf_options.level_compaction_dynamic_level_bytes = true;
     m_spdb->CreateColumnFamily(cf_options, strName, &handle);
     m_spcolfamily = std::shared_ptr<rocksdb::ColumnFamilyHandle>(handle);
 
     if (!status.ok())
         throw status.ToString();
-   
-    status = m_spdb->DropColumnFamily(m_spexpirecolfamily.get());
-    strName = m_spexpirecolfamily->GetName();
-
-    m_spdb->CreateColumnFamily(cf_options, strName, &handle);
-    m_spexpirecolfamily = std::shared_ptr<rocksdb::ColumnFamilyHandle>(handle);
-
-    if (!status.ok())
-        throw status.ToString();
-
     m_count = 0;
     return celem;
 }
@@ -253,14 +242,21 @@ void RocksDBStorageProvider::removeExpire(const char *key, size_t cchKey, long l
         throw status.ToString();
 }
 
-std::vector<std::string> RocksDBStorageProvider::getExpirationCandidates(unsigned int count)
+std::vector<std::string> RocksDBStorageProvider::getExpirationCandidates(unsigned int count, long long time)
 {
     std::vector<std::string> result;
     std::unique_ptr<rocksdb::Iterator> it = std::unique_ptr<rocksdb::Iterator>(m_spdb->NewIterator(ReadOptions(), m_spexpirecolfamily.get()));
-    for (it->SeekToFirst(); it->Valid() && result.size() < count; it->Next()) {
+    for (it->SeekToFirst(); it->Valid() && (*((long long *)it->key().data()) < time) && (result.size() < count); it->Next()) {
         if (FInternalKey(it->key().data(), it->key().size()))
             continue;
         result.emplace_back(it->value().data(), it->value().size());
+        rocksdb::Status status;
+        if (m_spbatch)
+            status = m_spbatch->Delete(m_spexpirecolfamily.get(), it->key());
+        else
+            status = m_spdb->Delete(WriteOptions(), m_spexpirecolfamily.get(), it->key());
+        if (!status.ok())
+            throw status.ToString();
     }
     return result;
 }
@@ -281,7 +277,7 @@ std::vector<std::string> RocksDBStorageProvider::getEvictionCandidates(unsigned
         }
     } else {
         std::unique_ptr<rocksdb::Iterator> it = std::unique_ptr<rocksdb::Iterator>(m_spdb->NewIterator(ReadOptions(), m_spexpirecolfamily.get()));
-        for (it->SeekToFirst(); it->Valid() && result.size() < count; it->Next()) {
+        for (it->SeekToFirst(); it->Valid() && result.size() < 16; it->Next()) {
             if (FInternalKey(it->key().data(), it->key().size()))
                 continue;
             result.emplace_back(it->value().data(), it->value().size());

src/storage/rocksdb.h

@@ -36,7 +36,7 @@ public:
     virtual bool enumerate(callback fn) const override;
     virtual bool enumerate_hashslot(callback fn, unsigned int hashslot) const override;
 
-    virtual std::vector<std::string> getExpirationCandidates(unsigned int count) override;
+    virtual std::vector<std::string> getExpirationCandidates(unsigned int count, long long time) override;
     virtual std::vector<std::string> getEvictionCandidates(unsigned int count) override;
     virtual void setExpire(const char *key, size_t cchKey, long long expire) override;
     virtual void removeExpire(const char *key, size_t cchKey, long long expire) override;

src/storage/teststorageprovider.h

@@ -29,7 +29,7 @@ public:
     virtual bool enumerate_hashslot(callback fn, unsigned int hashslot) const override;
     virtual size_t count() const override;
 
-    virtual std::vector<std::string> getExpirationCandidates(unsigned int) override { return std::vector<std::string>(); }
+    virtual std::vector<std::string> getExpirationCandidates(unsigned int, long long) override { return std::vector<std::string>(); }
     virtual std::vector<std::string> getEvictionCandidates(unsigned int) override { return std::vector<std::string>(); }
     virtual void setExpire(const char *, size_t, long long) override {}
     virtual void removeExpire(const char *, size_t, long long) override {}

tests/unit/flash.tcl

@@ -81,6 +81,21 @@ if {$::flash_enabled} {
             }
         }
 
+        test { EXPIRE reduces used disk space } {
+            r flushall
+            r set testkey foo ex 10
+            r debug flush-storage
+            set used [r debug get-storage-usage]
+            assert_equal {1} [r dbsize]
+            wait_for_condition 50 1000 {
+                [r dbsize] == 0
+            } else {
+                fail "key is not expired"
+            }
+            set after [r debug get-storage-usage]
+            assert {$after < $used}
+        }
+
         test { SUBKEY EXPIRE persists after cache flush } {
             r flushall
             r sadd testkey foo bar baz