set version to 6.3.4

* fix auditlog test titles, and enable tls tests for CI

* fix tls auditlog key string

build.yaml

@@ -21,13 +21,13 @@ machamp:
         parent: make-build
         # https://github.sc-corp.net/Snapchat/img/tree/master/keydb/ubuntu-20-04
         builder_image: us.gcr.io/snapchat-build-artifacts/prod/snapchat/img/keydb/keydb-ubuntu-20-04@sha256:cf869a3f5d1de1e1d976bb906689c37b7031938eb68661b844a38c532f27248c
-        command: ./runtest --clients 4 --verbose 
+        command: ./runtest --clients 4 --verbose --tls
       cluster-test:
         type: cmd
         parent: make-build
         # https://github.sc-corp.net/Snapchat/img/tree/master/keydb/ubuntu-20-04
         builder_image: us.gcr.io/snapchat-build-artifacts/prod/snapchat/img/keydb/keydb-ubuntu-20-04@sha256:cf869a3f5d1de1e1d976bb906689c37b7031938eb68661b844a38c532f27248c
-        command: ./runtest-cluster 
+        command: ./runtest-cluster --tls
       sentinel-test:
         type: cmd
         parent: make-build

machamp_scripts/build.sh

@@ -2,7 +2,7 @@
 
 # make the build
 git submodule init && git submodule update
-make BUILD_TLS=yes -j$(nproc) KEYDB_CFLAGS='-Werror' KEYDB_CXXFLAGS='-Werror'
+make BUILD_TLS=yes ENABLE_FLASH=yes -j$(nproc) KEYDB_CFLAGS='-Werror' KEYDB_CXXFLAGS='-Werror'
 
 # gen-cert
 ./utils/gen-test-certs.sh

pkg/deb/master_changelog

@@ -1,3 +1,86 @@
+keydb (6:6.3.4-1distribution_placeholder) codename_placeholder; urgency=medium
+
+  * Add keydb_modstatsd, a module for providing keydb stats to a local statsd server. Can be found under src/modules/keydb_modstatsd
+  * Fixed FreeBSD compile(Thanks to @bra-fsn)
+  * Fixed a bug in s3 bucket config that blocked startup(Thanks to Alexandre Chichmanian)
+  * Fixed a bug causing crash if keys command is called after a blocking command (Fixes issues #659, #605, #532)
+  * Added proper error checking of replica configs, now KeyDB will throw an error if replica-of config is passed before active-replica or multimaster configs
+  * Fixed double free bug in lazy free
+  * Added "overload-protect-percent" config, when enabled this will load shed clients whenever CPU usage exceeds configured value
+  * Added "availability-zone" config, this can be passed any string which will then be reported by info command
+  * Updated eviction logic to account for total system memory availability using sysinfo
+  * Fixed a bug where repl-backlog-size config was modified in keydb.conf with the runtime value during config rewrite
+  * Fixed a possible deadlock when running CLIENT KILL with large number of clients
+  * Fixed a bug where KeyDB would overcommit memory during fork BG save
+  * Fixed a bug with disk repl_backlog causing double free
+  * Fixed integer overflow issue in the temp rdb file naming (Thanks to @karthyuom)
+  * Fixed compile issue with GCC 13.1.1 (Thanks to @michieldwitte)
+  * Removed expireset and restored redis expire behaviour
+  * Fixed a bug causing forked processes to hang, specifically affecting RDB and AOF(Fixes issues #675, #619)
+  * Added "CLUSTER REPLICATE NO ONE" to turn a replica into an empty primary
+  * Added RDB-less full sync, can be enabled with config 'enable-keydb-fastsync'
+  * Added "flash-disable-key-cache" config to disable key cache which stores every key(but not value) in memory
+  * Moved cluster slot to key map to being stored by storage provider, rather than in memory
+  * Moved expires to be stored by storage provider, rather than in memory
+  * Enabled expiry/eviction from storage provider, previously only data stored in memory could be expired/evicted(Fixes #645 along with the 2 previous updates)
+  * Fixed a bug where swapdb result was not recovered after keydb restarts in FLASH mode (Thanks to @karthyuom)
+  * Fixed double free bug when fast sync was canceled early
+  * Fixed a bug where a temp rdb file with zero bytes is generated in flash mode (Thanks to @karthyuom)
+  * Fixed a bug where flash CF options are being reset to default after flushall (Thanks to @karthyuom)(Fixes issue #717)
+  * Updated debug reload command to work with flash
+  * Added support for KEYDB_PASSWORD env variable(Thanks to @einar-pexip)
+  * Compile with flash in Dockerfile(Thanks to @der-eismann)
+  * Fix broken redis-cli symlink (Thanks to @pimvandenbroek)
+  * Use tini for alpine docker(Thanks to @rofafor)
+
+-- Malavan Sotheeswaran <msotheeswaran@snapchat.com>  Sun, 29 Oct 2023 8:39:21 +0000
+
+keydb (6:6.3.3-1distribution_placeholder) codename_placeholder; urgency=medium
+
+  * To help accelerate our development efforts for KeyDB, this will be the last release containing support for Centos 7, Ubuntu 16.04, Ubuntu 18.04, Debian 9 and 32-bit builds
+  * This release contains fixes for 11 issues along with improvements to the KeyDB FLASH feature:
+  * Fixed race condition with expireset access (Issue #597)
+  * Fixed keys command with lua, and added keydb as an alternative to redis as lua variable to access db (Issue #562 and #594)
+  * Fixed hang in aof child (Issue #554)
+  * Fixed leaking fds from RDB save (Issue #453 and #584)
+  * Added config for S3 RDB load/save (Issue #584)
+  * Enabled active defrag during forkless background save to improve average memory efficiency (Issue #460)
+  * Fixed crash with fork background save during replication (Issue #567)
+  * RocksDB has been updated to v7.9.2
+  * Fixed missing slot_to_key map in FLASH cluster mode (Issue #574)
+  * Added keyspace notifications for keys loaded from FLASH at startup
+  * Fixed race condition in prefetchKeysAsync with FLASH enabled (Issue #571)
+  * Fixed integer overflow in rand family of commands(Issue #631, #632, #633)
+  * Fixed bad value in hincrbyfloat(Issue #634)
+  * Fixed OOM hang in rand family of commands(Issue #635)
+  * Added config to limit count of return values in rand family of commands(Issue #636)
+
+-- Ben Schermel <ben@eqalpha.com>  Tue, 18 Apr 2023 10:00:37 +0000
+
+keydb (6:6.3.2-1distribution_placeholder) codename_placeholder; urgency=medium
+
+  * This release contains Beta level support for KeyDB FLASH, new ASYNC commands, latency improvements and a number of bug fixes. 
+  * KeyDB FLASH is included as a Beta feature. FLASH allows KeyDB to persist data to the storage medium it is written to, avoiding the need for AOF/RDB files. KeyDB uses rocksdb as the persistent storage provider and can be enabled with config "storage-provider flash /path/to/rocksdb/output". Read more at https://docs.keydb.dev/docs/flash/
+  * In addition to GET/MGET, ASYNC support has been added for the following commands: HGET, HMGET, HKEYS, HVALS, HGETALL, HSCAN and can be enable with config "enable-async-commands yes”
+  * Packaging support for Ubuntu 22.04 (Jammy) and Debian 12 (Bookworm) has been included with this release
+  * Added new soft shutdown feature, can be enabled with config "soft-shutdown yes".
+  * If soft shutdown is enabled, instead of shutting down right away, the server will wait until all clients have disconnected, and will reject all new connection attempts.
+  * Fixed memory leak with tls certificates when tls allowlist is enabled
+  * Fixed bug in rdb load with flash enabled to ensure all dbs are safe to load (previously only checked db[0])
+  * Fixed race conditions in rdb load and replication
+  * Fixed memory access of rdb file after it should have been deleted
+  * Fixed integer overflow bug in flash(Issue #486)
+  * Improve TLS latency by queueing new commands before executing instead of after
+  * Removed O(n) count of memory usage from info command(replaced by O(1) estimate)
+  * Improved latency of clearing large number of flash DBs(Thanks to Paul Chen for this fix)(Issue #516)
+  * replaced sprintf with snprintf to avoid potential security bugs
+  * Fixed bug where a failed move due to key already existing in move target would result in the key being removed from move source(Thanks to Paul Chen for this fix)(Issue #497)
+  * Fixed usage of deprecated OpenSSL api in OpenSSL v>3.0.1(Issue #392)
+  * Imported security fixes from Redis (CVE-2023-22458 CVE-2022-35977)
+  * Other fixed issues: #480 #477 #454 #452 #303 #425 #492 #541
+
+-- Ben Schermel <ben@eqalpha.com>  Fri, 20 Jan 2023 20:00:37 +0000
+
 keydb (6:6.3.1-1distribution_placeholder) codename_placeholder; urgency=medium
 
   * This point release contains fixes to bugs related to expires, active-rep, and rdb saving

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; \
+        make -j$(nproc) BUILD_TLS=yes ENABLE_FLASH=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,6 +83,7 @@ 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; \
@@ -96,6 +97,10 @@ RUN set -eux; \
         echo 'if [ "${1#-}" != "$1" ] || [ "${1%.conf}" != "$1" ]; then' >> /usr/local/bin/docker-entrypoint.sh; \
         echo '        set -- keydb-server "$@"' >> /usr/local/bin/docker-entrypoint.sh; \
         echo 'fi' >> /usr/local/bin/docker-entrypoint.sh; \
+        echo '# if KEYDB_PASSWORD is set, add it to the arguments' >> /usr/local/bin/docker-entrypoint.sh; \
+        echo 'if [ -n "$KEYDB_PASSWORD" ]; then' >> /usr/local/bin/docker-entrypoint.sh; \
+        echo '        set -- "$@" --requirepass "${KEYDB_PASSWORD}"' >> /usr/local/bin/docker-entrypoint.sh; \
+        echo 'fi' >> /usr/local/bin/docker-entrypoint.sh; \
         echo "# allow the container to be started with `--user`" >> /usr/local/bin/docker-entrypoint.sh; \
         echo 'if [ "$1" = "keydb-server" -a "$(id -u)" = "0" ]; then' >> /usr/local/bin/docker-entrypoint.sh; \
         echo "        find . \! -user keydb -exec chown keydb '{}' +" >> /usr/local/bin/docker-entrypoint.sh; \

pkg/docker/Dockerfile_Alpine

@@ -1,11 +1,11 @@
-FROM alpine:3.12
+FROM alpine:3.18
 # add our user and group first to make sure their IDs get assigned consistently, regardless of whatever dependencies get added
 RUN addgroup -S -g 1000 keydb && adduser -S -G keydb -u 999 keydb
 RUN mkdir -p /etc/keydb
 ARG BRANCH
 RUN set -eux; \
         \
-        apk add --no-cache su-exec; \
+        apk add --no-cache su-exec tini; \
         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; \
+        make -j$(nproc) BUILD_TLS=yes ENABLE_FLASH=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,6 +54,7 @@ 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; \
@@ -67,6 +68,10 @@ RUN set -eux; \
         echo 'if [ "${1#-}" != "$1" ] || [ "${1%.conf}" != "$1" ]; then' >> /usr/local/bin/docker-entrypoint.sh; \
         echo '        set -- keydb-server "$@"' >> /usr/local/bin/docker-entrypoint.sh; \
         echo 'fi' >> /usr/local/bin/docker-entrypoint.sh; \
+        echo '# if KEYDB_PASSWORD is set, add it to the arguments' >> /usr/local/bin/docker-entrypoint.sh; \
+        echo 'if [ -n "$KEYDB_PASSWORD" ]; then' >> /usr/local/bin/docker-entrypoint.sh; \
+        echo '        set -- "$@" --requirepass "${KEYDB_PASSWORD}"' >> /usr/local/bin/docker-entrypoint.sh; \
+        echo 'fi' >> /usr/local/bin/docker-entrypoint.sh; \
         echo "# allow the container to be started with `--user`" >> /usr/local/bin/docker-entrypoint.sh; \
         echo 'if [ "$1" = "keydb-server" -a "$(id -u)" = "0" ]; then' >> /usr/local/bin/docker-entrypoint.sh; \
         echo "        find . \! -user keydb -exec chown keydb '{}' +" >> /usr/local/bin/docker-entrypoint.sh; \
@@ -76,6 +81,6 @@ RUN set -eux; \
         chmod +x /usr/local/bin/docker-entrypoint.sh
 VOLUME /data
 WORKDIR /data
-ENTRYPOINT ["docker-entrypoint.sh"]
+ENTRYPOINT ["tini", "--", "docker-entrypoint.sh"]
 EXPOSE 6379
 CMD ["keydb-server", "/etc/keydb/keydb.conf"]

src/IStorage.h

@@ -1,6 +1,7 @@
 #pragma once
 #include <functional>
 #include "sds.h"
+#include <string>
 
 #define METADATA_DB_IDENTIFIER "c299fde0-6d42-4ec4-b939-34f680ffe39f"
 
@@ -43,6 +44,11 @@ public:
         endWriteBatch();
     }
 
+    virtual std::vector<std::string> getExpirationCandidates(unsigned int count) = 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;
+
     virtual void beginWriteBatch() {} // NOP
     virtual void endWriteBatch() {} // NOP
 

src/Makefile

@@ -358,6 +358,7 @@ endif
 OS := $(shell cat /etc/os-release | grep ID= | head -n 1 | cut -d'=' -f2)
 ifeq ($(OS),alpine)
     FINAL_CXXFLAGS+=-DUNW_LOCAL_ONLY
+	FINAL_CXXFLAGS+=-DALPINE
     FINAL_LIBS += -lunwind
 endif
 

src/SnapshotPayloadParseState.cpp

@@ -136,14 +136,33 @@ void SnapshotPayloadParseState::flushQueuedKeys() {
     int idb = current_database;
     serverAssert(vecqueuedKeys.size() == vecqueuedVals.size());
     auto sizePrev = vecqueuedKeys.size();
-    (*insertsInFlight)++;
-    std::weak_ptr<std::atomic<int>> 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());
-            (*(insertsInFlightTmp.lock()))--;
-            delete pallocator;
-        });
+        if (g_pserver->m_pstorageFactory) {
+            (*insertsInFlight)++;
+            std::weak_ptr<std::atomic<int>> insertsInFlightTmp = insertsInFlight; // C++ GRRRRRRRRRRRRRRRR, we don't want to capute "this" because that's dangerous
+            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());
+                (*(insertsInFlightTmp.lock()))--;
+                delete pallocator;
+            });
+        } else {
+            for (size_t ival = 0; ival < vecqueuedKeys.size(); ++ival) {
+                size_t offset = 0;
+                auto spexpire = deserializeExpire(vecqueuedVals[ival], vecqueuedValsCb[ival], &offset);    
+                auto o = deserializeStoredObject(vecqueuedVals[ival] + offset, vecqueuedValsCb[ival] - offset);
+                sds sdsKey = sdsnewlen(vecqueuedKeys[ival], -static_cast<ssize_t>(vecqueuedKeysCb[ival]));
+                if (dbMerge(g_pserver->db[idb], sdsKey, o, false)) {
+                    if (spexpire != nullptr)
+                        g_pserver->db[idb]->setExpire(sdsKey, std::move(*spexpire));
+                } else {
+                    sdsfree(sdsKey);
+                }
+            }
+            vecqueuedKeys.clear();
+            vecqueuedKeysCb.clear();
+            vecqueuedVals.clear();
+            vecqueuedValsCb.clear();
+        }
     } else {
         // else drop the data
         vecqueuedKeys.clear();

src/StorageCache.cpp

@@ -84,19 +84,31 @@ void StorageCache::cacheKey(const char *rgch, size_t cch)
 
 bool StorageCache::erase(sds key)
 {
+    unsigned long long when = 0;
+    m_spstorage->retrieve(key, sdslen(key), [&when](const char *, size_t, const void * data, size_t cbdata) {
+        auto e = deserializeExpire((const char *)data, cbdata, nullptr);
+        if (e != nullptr)
+            when = e->when();
+    });
     bool result = m_spstorage->erase(key, sdslen(key));
     std::unique_lock<fastlock> ul(m_lock);
-    if (result && m_pdict != nullptr)
+    if (result)
     {
-        uint64_t hash = dictSdsHash(key);
-        dictEntry *de = dictFind(m_pdict, reinterpret_cast<void*>(hash));
-        serverAssert(de != nullptr);
-        de->v.s64--;
-        serverAssert(de->v.s64 >= 0);
-        if (de->v.s64 == 0) {
-            dictDelete(m_pdict, reinterpret_cast<void*>(hash));
-        } else {
-            m_collisionCount--;
+        if (m_pdict != nullptr)
+        {
+            uint64_t hash = dictSdsHash(key);
+            dictEntry *de = dictFind(m_pdict, reinterpret_cast<void*>(hash));
+            serverAssert(de != nullptr);
+            de->v.s64--;
+            serverAssert(de->v.s64 >= 0);
+            if (de->v.s64 == 0) {
+                dictDelete(m_pdict, reinterpret_cast<void*>(hash));
+            } else {
+                m_collisionCount--;
+            }
+        }
+        if (when != 0) {
+            m_spstorage->removeExpire(key, sdslen(key), when);
         }
     }
     return result;
@@ -111,6 +123,9 @@ void StorageCache::insert(sds key, const void *data, size_t cbdata, bool fOverwr
     }
     ul.unlock();
     m_spstorage->insert(key, sdslen(key), (void*)data, cbdata, fOverwrite);
+    auto e = deserializeExpire((const char *)data, cbdata, nullptr);
+    if (e != nullptr)
+        m_spstorage->setExpire(key, sdslen(key), e->when());
 }
 
 long _dictKeyIndex(dict *d, const void *key, uint64_t hash, dictEntry **existing);
@@ -119,13 +134,18 @@ void StorageCache::bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, si
     std::vector<dictEntry*> vechashes;
     if (m_pdict != nullptr) {
         vechashes.reserve(celem);
+    }
 
-        for (size_t ielem = 0; ielem < celem; ++ielem) {
+    for (size_t ielem = 0; ielem < celem; ++ielem) {
+        if (m_pdict != nullptr) {
             dictEntry *de = (dictEntry*)zmalloc(sizeof(dictEntry));
             de->key = (void*)dictGenHashFunction(rgkeys[ielem], (int)rgcbkeys[ielem]);
             de->v.u64 = 1;
             vechashes.push_back(de);
         }
+        auto e = deserializeExpire(rgvals[ielem], rgcbvals[ielem], nullptr);
+        if (e != nullptr)
+            m_spstorage->setExpire(rgkeys[ielem], rgcbkeys[ielem], e->when());
     }
 
     std::unique_lock<fastlock> ul(m_lock);

src/StorageCache.h

@@ -51,11 +51,18 @@ 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> 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); }
+
     void beginWriteBatch();
     void endWriteBatch() { m_spstorage->endWriteBatch(); }
     void batch_lock() { return m_spstorage->batch_lock(); }
     void batch_unlock() { return m_spstorage->batch_unlock(); }
 
+    void flush() { m_spstorage->flush(); }
+
     size_t count() const;
 
     const StorageCache *clone();

src/ae.cpp

@@ -113,6 +113,8 @@ enum class AE_ASYNC_OP
     PostCppFunction,
     DeleteFileEvent,
     CreateFileEvent,
+    PostAsynDBFunction,
+
 };
 
 struct aeCommand
@@ -125,6 +127,8 @@ struct aeCommand
         aePostFunctionProc *proc;
         aeFileProc *fproc;
         std::function<void()> *pfn;
+        aePostFunctionTokenProc* tproc;
+
     };
     void *clientData;
 };
@@ -175,6 +179,17 @@ 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;
         }
@@ -256,6 +271,20 @@ 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)
@@ -861,7 +890,7 @@ void aeReleaseForkLock()
 
 void aeForkLockInChild()
 {
-    g_forkLock.setNotify(false);
+    g_forkLock.setMulti(false);
 }
 
 int aeThreadOwnsLock()

src/ae.h

@@ -72,13 +72,15 @@ 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 {
@@ -134,6 +136,8 @@ 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,7 +5991,9 @@ 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)) {
-                    if (slot != thisslot) {
+                    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)) {
                         /* Error: multiple keys from different slots. */
                         getKeysFreeResult(&result);
                         if (error_code)

src/config.cpp

@@ -2597,6 +2597,19 @@ static int updateMaxmemory(long long val, long long prev, const char **err) {
     return 1;
 }
 
+static int updateFlashMaxmemory(long long val, long long prev, const char **err) {
+    UNUSED(prev);
+    UNUSED(err);
+    if (val && g_pserver->m_pstorageFactory) {
+        size_t used = g_pserver->m_pstorageFactory->totalDiskspaceUsed();
+        if ((unsigned long long)val < used) {
+            serverLog(LL_WARNING,"WARNING: the new maxstorage value set via CONFIG SET (%llu) is smaller than the current storage usage (%zu). This will result in key eviction and/or the inability to accept new write commands depending on the maxmemory-policy.", g_pserver->maxstorage, used);
+        }
+        performEvictions(false /*fPreSnapshot*/);
+    }
+    return 1;
+}
+
 static int updateGoodSlaves(long long val, long long prev, const char **err) {
     UNUSED(val);
     UNUSED(prev);
@@ -2933,14 +2946,14 @@ 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_size, 1024*1024, MEMORY_CONFIG, NULL, updateReplBacklogSize), /* Default: 1mb */
+    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-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),
-    createULongLongConfig("maxstorage", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, g_pserver->maxstorage, 0, MEMORY_CONFIG, NULL, NULL),
+    createULongLongConfig("maxstorage", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, g_pserver->maxstorage, 0, MEMORY_CONFIG, NULL, updateFlashMaxmemory),
 
     /* Size_t configs */
     createSizeTConfig("hash-max-ziplist-entries", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, g_pserver->hash_max_ziplist_entries, 512, INTEGER_CONFIG, NULL, NULL),
@@ -2969,6 +2982,8 @@ 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),
+    createBoolConfig("enable-keydb-fastsync", NULL, MODIFIABLE_CONFIG, g_pserver->fEnableFastSync, 0, 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

@@ -56,7 +56,6 @@ int expireIfNeeded(redisDb *db, robj *key, robj *o);
 void slotToKeyUpdateKeyCore(const char *key, size_t keylen, int add);
 
 std::unique_ptr<expireEntry> deserializeExpire(const char *str, size_t cch, size_t *poffset);
-sds serializeStoredObjectAndExpire(robj_roptr o);
 
 dictType dictChangeDescType {
     dictSdsHash,                /* hash function */
@@ -2703,11 +2702,6 @@ 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);
@@ -2745,6 +2739,13 @@ 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;
 }
@@ -2880,10 +2881,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);
@@ -3254,8 +3255,8 @@ std::unique_ptr<expireEntry> deserializeExpire(const char *str, size_t cch, size
         if (subkey)
             sdsfree(subkey);
     }
-
-    *poffset = offset;
+    if (poffset != nullptr)
+        *poffset = offset;
     return spexpire;
 }
 

src/debug.cpp

@@ -79,6 +79,7 @@ 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 ============================== */
 
@@ -308,7 +309,7 @@ void computeDatasetDigest(unsigned char *final) {
         mixDigest(final,&aux,sizeof(aux));
 
         /* Iterate this DB writing every entry */
-        db->iterate_threadsafe([final, db](const char *key, robj_roptr o)->bool {
+        db->iterate_threadsafe([final](const char *key, robj_roptr o)->bool {
             unsigned char digest[20];
             robj *keyobj;
 
@@ -535,30 +536,75 @@ NULL
                 return;
             }
         }
+        
+        if (g_pserver->m_pstorageFactory) {
+            protectClient(c);
 
-        /* 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);
+            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");
                 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")) {
@@ -932,6 +978,21 @@ NULL
         mallctl_string(c, c->argv+2, c->argc-2);
         return;
 #endif
+    } else if(!strcasecmp(szFromObj(c->argv[1]),"flush-storage") && c->argc == 2) {
+        if (g_pserver->m_pstorageFactory != nullptr) {
+            for (int i = 0; i < cserver.dbnum; i++) {
+                g_pserver->db[i]->getStorageCache()->flush();
+            }
+            addReply(c,shared.ok);
+        } else {
+            addReplyError(c, "Can't flush storage if no storage provider is set");
+        }
+    } else if (!strcasecmp(szFromObj(c->argv[1]),"get-storage-usage") && c->argc == 2) {
+        if (g_pserver->m_pstorageFactory != nullptr) {
+            addReplyLongLong(c, g_pserver->m_pstorageFactory->totalDiskspaceUsed());
+        } else {
+            addReplyLongLong(c, 0);
+        }
     } else {
         addReplySubcommandSyntaxError(c);
         return;

src/evict.cpp

@@ -100,6 +100,36 @@ unsigned long long estimateObjectIdleTime(robj_roptr o) {
     }
 }
 
+unsigned long long getIdle(robj *obj, const expireEntry *e) {
+    unsigned long long idle;
+    /* Calculate the idle time according to the policy. This is called
+        * idle just because the code initially handled LRU, but is in fact
+        * just a score where an higher score means better candidate. */
+    if (g_pserver->maxmemory_policy & MAXMEMORY_FLAG_LRU) {
+        idle = (obj != nullptr) ? estimateObjectIdleTime(obj) : 0;
+    } else if (g_pserver->maxmemory_policy & MAXMEMORY_FLAG_LFU) {
+        /* When we use an LRU policy, we sort the keys by idle time
+            * so that we expire keys starting from greater idle time.
+            * However when the policy is an LFU one, we have a frequency
+            * estimation, and we want to evict keys with lower frequency
+            * first. So inside the pool we put objects using the inverted
+            * frequency subtracting the actual frequency to the maximum
+            * frequency of 255. */
+        idle = 255-LFUDecrAndReturn(obj);
+    } else if (g_pserver->maxmemory_policy == MAXMEMORY_VOLATILE_TTL) {
+        /* In this case the sooner the expire the better. */
+        if (e != nullptr)
+            idle = ULLONG_MAX - e->when();
+        else
+            idle = 0;
+    } else if (g_pserver->maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS) {
+        idle = ULLONG_MAX;
+    } else {
+        serverPanic("Unknown eviction policy in storage eviction");
+    }
+    return idle;
+}
+
 /* LRU approximation algorithm
  *
  * Redis uses an approximation of the LRU algorithm that runs in constant
@@ -137,28 +167,7 @@ void evictionPoolAlloc(void) {
 
 void processEvictionCandidate(int dbid, sds key, robj *o, const expireEntry *e, struct evictionPoolEntry *pool)
 {
-    unsigned long long idle;
-
-    /* Calculate the idle time according to the policy. This is called
-        * idle just because the code initially handled LRU, but is in fact
-        * just a score where an higher score means better candidate. */
-    if (g_pserver->maxmemory_policy & MAXMEMORY_FLAG_LRU) {
-        idle = (o != nullptr) ? estimateObjectIdleTime(o) : 0;
-    } else if (g_pserver->maxmemory_policy & MAXMEMORY_FLAG_LFU) {
-        /* When we use an LRU policy, we sort the keys by idle time
-            * so that we expire keys starting from greater idle time.
-            * However when the policy is an LFU one, we have a frequency
-            * estimation, and we want to evict keys with lower frequency
-            * first. So inside the pool we put objects using the inverted
-            * frequency subtracting the actual frequency to the maximum
-            * frequency of 255. */
-        idle = 255-LFUDecrAndReturn(o);
-    } else if (g_pserver->maxmemory_policy == MAXMEMORY_VOLATILE_TTL) {
-        /* In this case the sooner the expire the better. */
-        idle = ULLONG_MAX - e->when();
-    } else {
-        serverPanic("Unknown eviction policy in evictionPoolPopulate()");
-    }
+    unsigned long long idle = getIdle(o,e);
 
     /* Insert the element inside the pool.
         * First, find the first empty bucket or the first populated
@@ -600,6 +609,31 @@ static unsigned long evictionTimeLimitUs() {
     return ULONG_MAX;   /* No limit to eviction time */
 }
 
+void evict(redisDb *db, robj *keyobj) {
+    mstime_t eviction_latency;
+    propagateExpire(db,keyobj,g_pserver->lazyfree_lazy_eviction);
+    /* We compute the amount of memory freed by db*Delete() alone.
+    * It is possible that actually the memory needed to propagate
+    * the DEL in AOF and replication link is greater than the one
+    * we are freeing removing the key, but we can't account for
+    * that otherwise we would never exit the loop.
+    *
+    * AOF and Output buffer memory will be freed eventually so
+    * we only care about memory used by the key space. */
+    latencyStartMonitor(eviction_latency);
+    if (g_pserver->lazyfree_lazy_eviction)
+        dbAsyncDelete(db,keyobj);
+    else
+        dbSyncDelete(db,keyobj);
+    latencyEndMonitor(eviction_latency);
+    latencyAddSampleIfNeeded("eviction-del",eviction_latency);
+
+    signalModifiedKey(NULL,db,keyobj);
+    notifyKeyspaceEvent(NOTIFY_EVICTED, "evicted",
+        keyobj, db->id);
+    decrRefCount(keyobj);
+}
+
 static void updateSysAvailableMemory() {
     if (g_pserver->force_eviction_percent) {
         g_pserver->cron_malloc_stats.sys_available = getMemAvailable();
@@ -637,7 +671,7 @@ int performEvictions(bool fPreSnapshot) {
     int keys_freed = 0;
     size_t mem_reported, mem_tofree;
     long long mem_freed; /* May be negative */
-    mstime_t latency, eviction_latency;
+    mstime_t latency;
     long long delta;
     int slaves = listLength(g_pserver->slaves);
     const bool fEvictToStorage = !cserver.delete_on_evict && g_pserver->db[0]->FStorageProvider();
@@ -662,6 +696,43 @@ int performEvictions(bool fPreSnapshot) {
     monotime evictionTimer;
     elapsedStart(&evictionTimer);
 
+    if (g_pserver->maxstorage && g_pserver->m_pstorageFactory != nullptr) {
+        while (g_pserver->m_pstorageFactory->totalDiskspaceUsed() >= g_pserver->maxstorage && elapsedUs(evictionTimer) < eviction_time_limit_us) {
+            redisDb *db;
+            std::vector<std::string> evictionPool;
+            robj *bestkey = nullptr;
+            redisDb *bestdb = nullptr;
+            unsigned long long bestidle = 0;
+            for (int i = 0; i < cserver.dbnum; i++) {
+                db = g_pserver->db[i];
+                evictionPool = db->getStorageCache()->getEvictionCandidates(g_pserver->maxmemory_samples);
+                for (std::string key : evictionPool) {
+                    robj *keyobj = createStringObject(key.c_str(), key.size());
+                    robj *obj = db->find(szFromObj(keyobj));
+                    if (obj != nullptr) {
+                        expireEntry *e = db->getExpire(keyobj);
+                        unsigned long long idle = getIdle(obj, e);
+
+                        if (bestkey == nullptr || bestidle < idle) {
+                            if (bestkey != nullptr)
+                                decrRefCount(bestkey);
+                            incrRefCount(keyobj);
+                            bestkey = keyobj;
+                            bestidle = idle;
+                            bestdb = db;
+                        }
+                    }
+                    decrRefCount(keyobj);
+                }
+            }
+            if (bestkey) {
+                evict(bestdb, bestkey);
+            } else {
+                break; //could not find a key to evict so stop now
+            }
+        }
+    }
+
     if (g_pserver->maxstorage && g_pserver->m_pstorageFactory != nullptr && g_pserver->m_pstorageFactory->totalDiskspaceUsed() >= g_pserver->maxstorage)
         goto cant_free_storage;
 
@@ -776,7 +847,7 @@ int performEvictions(bool fPreSnapshot) {
                 if (db->removeCachedValue(bestkey, &deT)) {
                     mem_freed += splazy->addEntry(db->dictUnsafeKeyOnly(), deT);
                     ckeysFailed = 0;
-		    g_pserver->stat_evictedkeys++;
+                    g_pserver->stat_evictedkeys++;
                 }
                 else {
                     delta = 0;
@@ -788,30 +859,11 @@ int performEvictions(bool fPreSnapshot) {
             else
             {
                 robj *keyobj = createStringObject(bestkey,sdslen(bestkey));
-                propagateExpire(db,keyobj,g_pserver->lazyfree_lazy_eviction);
-                /* We compute the amount of memory freed by db*Delete() alone.
-                * It is possible that actually the memory needed to propagate
-                * the DEL in AOF and replication link is greater than the one
-                * we are freeing removing the key, but we can't account for
-                * that otherwise we would never exit the loop.
-                *
-                * AOF and Output buffer memory will be freed eventually so
-                * we only care about memory used by the key space. */
                 delta = (long long) zmalloc_used_memory();
-                latencyStartMonitor(eviction_latency);
-                if (g_pserver->lazyfree_lazy_eviction)
-                    dbAsyncDelete(db,keyobj);
-                else
-                    dbSyncDelete(db,keyobj);
-                latencyEndMonitor(eviction_latency);
-                latencyAddSampleIfNeeded("eviction-del",eviction_latency);
+                evict(db, keyobj);
                 delta -= (long long) zmalloc_used_memory();
                 mem_freed += delta;
                 g_pserver->stat_evictedkeys++;
-                signalModifiedKey(NULL,db,keyobj);
-                notifyKeyspaceEvent(NOTIFY_EVICTED, "evicted",
-                    keyobj, db->id);
-                decrRefCount(keyobj);
             }
             keys_freed++;
 

src/expire.cpp

@@ -396,126 +396,152 @@ void pexpireMemberAtCommand(client *c)
          * distribute the time evenly across DBs. */
         current_db++;
 
-        /* Continue to expire if at the end of the cycle there are still
-         * a big percentage of keys to expire, compared to the number of keys
-         * we scanned. The percentage, stored in config_cycle_acceptable_stale
-         * is not fixed, but depends on the Redis configured "expire effort". */
-        do {
-            unsigned long num, slots;
-            long long now, ttl_sum;
-            int ttl_samples;
-            iteration++;
+        if (g_pserver->m_pstorageFactory == nullptr) {
+            /* Continue to expire if at the end of the cycle there are still
+            * a big percentage of keys to expire, compared to the number of keys
+            * we scanned. The percentage, stored in config_cycle_acceptable_stale
+            * is not fixed, but depends on the Redis configured "expire effort". */
+            do {
+                unsigned long num, slots;
+                long long now, ttl_sum;
+                int ttl_samples;
+                iteration++;
 
-            /* If there is nothing to expire try next DB ASAP. */
-            if (db->expireSize() == 0) {
-                db->avg_ttl = 0;
-                break;
-            }
-            num = dictSize(db->m_pdict);
-            slots = dictSlots(db->m_pdict);
-            now = mstime();
-
-            /* When there are less than 1% filled slots, sampling the key
-             * space is expensive, so stop here waiting for better times...
-             * The dictionary will be resized asap. */
-            if (slots > DICT_HT_INITIAL_SIZE &&
-                (num*100/slots < 1)) break;
-
-            /* The main collection cycle. Sample random keys among keys
-             * with an expire set, checking for expired ones. */
-            expired = 0;
-            sampled = 0;
-            ttl_sum = 0;
-            ttl_samples = 0;
-
-            if (num > config_keys_per_loop)
-                num = config_keys_per_loop;
-
-            /* Here we access the low level representation of the hash table
-             * for speed concerns: this makes this code coupled with dict.c,
-             * but it hardly changed in ten years.
-             *
-             * Note that certain places of the hash table may be empty,
-             * so we want also a stop condition about the number of
-             * buckets that we scanned. However scanning for free buckets
-             * is very fast: we are in the cache line scanning a sequential
-             * array of NULL pointers, so we can scan a lot more buckets
-             * than keys in the same time. */
-            long max_buckets = num*20;
-            long checked_buckets = 0;
-
-            while (sampled < num && checked_buckets < max_buckets) {
-                for (int table = 0; table < 2; table++) {
-                    if (table == 1 && !dictIsRehashing(db->m_pdict)) break;
-
-                    unsigned long idx = db->expires_cursor;
-                    idx &= db->m_pdict->ht[table].sizemask;
-                    dictEntry *de = db->m_pdict->ht[table].table[idx];
-                    long long ttl;
-
-                    /* Scan the current bucket of the current table. */
-                    checked_buckets++;
-                    while(de) {
-                        /* Get the next entry now since this entry may get
-                         * deleted. */
-                        dictEntry *e = de;
-                        robj *o = (robj*)dictGetVal(de);
-                        de = de->next;
-                        if (!o->FExpires())
-                            continue;
-
-                        expireEntry *exp = &o->expire;
-
-                        serverAssert(exp->when() > 0);
-                        ttl = exp->when()-now;
-                        size_t tried = 0;
-                        if (exp->when() <= now) {
-                            if (activeExpireCycleExpire(db,(const char*)dictGetKey(e),*exp,now,tried)) expired++;
-                            serverAssert(ttl <= 0);
-                        } else {
-                            serverAssert(ttl > 0);
-                        }
-                        if (ttl > 0) {
-                            /* We want the average TTL of keys yet
-                             * not expired. */
-                            ttl_sum += ttl;
-                            ttl_samples++;
-                        }
-                        sampled++;
-                    }
-                }
-                db->expires_cursor++;
-            }
-            total_expired += expired;
-            total_sampled += sampled;
-
-            /* Update the average TTL stats for this database. */
-            if (ttl_samples) {
-                long long avg_ttl = ttl_sum/ttl_samples;
-
-                /* Do a simple running average with a few samples.
-                 * We just use the current estimate with a weight of 2%
-                 * and the previous estimate with a weight of 98%. */
-                if (db->avg_ttl == 0) db->avg_ttl = avg_ttl;
-                db->avg_ttl = (db->avg_ttl/50)*49 + (avg_ttl/50);
-            }
-
-            /* We can't block forever here even if there are many keys to
-             * expire. So after a given amount of milliseconds return to the
-             * caller waiting for the other active expire cycle. */
-            if ((iteration & 0xf) == 0) { /* check once every 16 iterations. */
-                elapsed = ustime()-start;
-                if (elapsed > timelimit) {
-                    timelimit_exit = 1;
-                    g_pserver->stat_expired_time_cap_reached_count++;
+                /* If there is nothing to expire try next DB ASAP. */
+                if (db->expireSize() == 0) {
+                    db->avg_ttl = 0;
                     break;
                 }
+                num = dictSize(db->m_pdict);
+                slots = dictSlots(db->m_pdict);
+                now = mstime();
+
+                /* When there are less than 1% filled slots, sampling the key
+                * space is expensive, so stop here waiting for better times...
+                * The dictionary will be resized asap. */
+                if (slots > DICT_HT_INITIAL_SIZE &&
+                    (num*100/slots < 1)) break;
+
+                /* The main collection cycle. Sample random keys among keys
+                * with an expire set, checking for expired ones. */
+                expired = 0;
+                sampled = 0;
+                ttl_sum = 0;
+                ttl_samples = 0;
+
+                if (num > config_keys_per_loop)
+                    num = config_keys_per_loop;
+
+                /* Here we access the low level representation of the hash table
+                * for speed concerns: this makes this code coupled with dict.c,
+                * but it hardly changed in ten years.
+                *
+                * Note that certain places of the hash table may be empty,
+                * so we want also a stop condition about the number of
+                * buckets that we scanned. However scanning for free buckets
+                * is very fast: we are in the cache line scanning a sequential
+                * array of NULL pointers, so we can scan a lot more buckets
+                * than keys in the same time. */
+                long max_buckets = num*20;
+                long checked_buckets = 0;
+
+                while (sampled < num && checked_buckets < max_buckets) {
+                    for (int table = 0; table < 2; table++) {
+                        if (table == 1 && !dictIsRehashing(db->m_pdict)) break;
+
+                        unsigned long idx = db->expires_cursor;
+                        idx &= db->m_pdict->ht[table].sizemask;
+                        dictEntry *de = db->m_pdict->ht[table].table[idx];
+                        long long ttl;
+
+                        /* Scan the current bucket of the current table. */
+                        checked_buckets++;
+                        while(de) {
+                            /* Get the next entry now since this entry may get
+                            * deleted. */
+                            dictEntry *e = de;
+                            robj *o = (robj*)dictGetVal(de);
+                            de = de->next;
+                            if (!o->FExpires())
+                                continue;
+
+                            expireEntry *exp = &o->expire;
+
+                            serverAssert(exp->when() > 0);
+                            ttl = exp->when()-now;
+                            size_t tried = 0;
+                            if (exp->when() <= now) {
+                                if (activeExpireCycleExpire(db,(const char*)dictGetKey(e),*exp,now,tried)) expired++;
+                                serverAssert(ttl <= 0);
+                            } else {
+                                serverAssert(ttl > 0);
+                            }
+                            if (ttl > 0) {
+                                /* We want the average TTL of keys yet
+                                * not expired. */
+                                ttl_sum += ttl;
+                                ttl_samples++;
+                            }
+                            sampled++;
+                        }
+                    }
+                    db->expires_cursor++;
+                }
+                total_expired += expired;
+                total_sampled += sampled;
+
+                /* Update the average TTL stats for this database. */
+                if (ttl_samples) {
+                    long long avg_ttl = ttl_sum/ttl_samples;
+
+                    /* Do a simple running average with a few samples.
+                    * We just use the current estimate with a weight of 2%
+                    * and the previous estimate with a weight of 98%. */
+                    if (db->avg_ttl == 0) db->avg_ttl = avg_ttl;
+                    db->avg_ttl = (db->avg_ttl/50)*49 + (avg_ttl/50);
+                }
+
+                /* We can't block forever here even if there are many keys to
+                * expire. So after a given amount of milliseconds return to the
+                * caller waiting for the other active expire cycle. */
+                if ((iteration & 0xf) == 0) { /* check once every 16 iterations. */
+                    elapsed = ustime()-start;
+                    if (elapsed > timelimit) {
+                        timelimit_exit = 1;
+                        g_pserver->stat_expired_time_cap_reached_count++;
+                        break;
+                    }
+                }
+                /* We don't repeat the cycle for the current database if there are
+                * an acceptable amount of stale keys (logically expired but yet
+                * not reclaimed). */
+            } while (sampled == 0 ||
+                    (expired*100/sampled) > config_cycle_acceptable_stale);
+        } else {
+            long prev_expired;
+            long long now = mstime();
+            size_t tried = 0;
+            std::vector<std::string> keys;
+            do {
+                prev_expired = total_expired;
+                keys = db->getStorageCache()->getExpirationCandidates(ACTIVE_EXPIRE_CYCLE_LOOKUPS_PER_LOOP);
+                for (std::string key : keys) {
+                    robj* keyobj = createStringObject(key.c_str(), key.size());
+                    db->find(szFromObj(keyobj));
+                    expireEntry *e = db->getExpire(keyobj);
+                    if (e != nullptr && e->when() < now)
+                        total_expired += activeExpireCycleExpire(db, szFromObj(keyobj), *e, now, tried);
+                    decrRefCount(keyobj);
+                }
+                total_sampled += keys.size();
+                elapsed = ustime()-start;
+            } while (keys.size() > 0 && (elapsed < timelimit) && (total_expired - prev_expired) > 0);
+
+            if (ustime()-start > timelimit) {
+                timelimit_exit = 1;
+                g_pserver->stat_expired_time_cap_reached_count++;
             }
-            /* We don't repeat the cycle for the current database if there are
-             * an acceptable amount of stale keys (logically expired but yet
-             * not reclaimed). */
-        } while (sampled == 0 ||
-                 (expired*100/sampled) > config_cycle_acceptable_stale);
+        }
     }
 
     elapsed = ustime()-start;

src/expire.h

@@ -160,6 +160,8 @@ public:
 
     expireEntry &operator=(expireEntry &&e)
     {
+        if (FFat())
+            delete pfatentry();
         s = e.s;
         e.s.m_whenAndPtrUnion = 0;
         e.s.fFat = false;
@@ -167,6 +169,8 @@ 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,6 +19,7 @@ static size_t getMemKey(std::string key) {
     }
     return 0;
 # else
+    (void)key;
     return 0;
 # endif
 }

src/module.cpp

@@ -690,8 +690,9 @@ void moduleHandlePropagationAfterCommandCallback(RedisModuleCtx *ctx) {
 }
 
 /* Free the context after the user function was called. */
-void moduleFreeContext(RedisModuleCtx *ctx) {
-    moduleHandlePropagationAfterCommandCallback(ctx);
+void moduleFreeContext(RedisModuleCtx *ctx, bool propogate) {
+    if (propogate)
+        moduleHandlePropagationAfterCommandCallback(ctx);
     autoMemoryCollect(ctx);
     poolAllocRelease(ctx);
     if (ctx->postponed_arrays) {
@@ -7057,7 +7058,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 comparision is successful the command returns REDISMODULE_OK
+ * If the comparison 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)
+	$(CXX) -o $@ -c $< $(MODULE_FLAGS) -I../../../deps/cpp-statsd-client/include $(MODSNAP_CXX_FLAGS) -g
 
 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;
-utsname sysName;
+std::string nodeName;
 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), true /* prefixOnly */);   
+	g_stats->gauge("cpu_load_perc", ((double)(mscpuCur - s_mscpuLast) / ((curTime - lastTime)*1000))*100, false /* 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) + sysName.nodename);
+            g_stats->increment("node_name" + std::string(g_string_counter_separator) + 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);
+        g_stats->gauge("keys", keys, false /* prefixOnly */);
         RedisModule_Log(ctx, REDISMODULE_LOGLEVEL_DEBUG, "Emitting metric \"keys\": %llu", keys);
         g_stats->timing("emit_keys_metric_time_taken_us", ustime() - commandStartTime);
 
@@ -672,17 +672,20 @@ 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 = std::string(sysName.nodename);
-            unameResult = 1;
-        }
+    else if (unameResult == 0) {
+        m_strPrefix = nodeName;
+        unameResult = 1;
     }
-    std::replace(m_strPrefix.begin(), m_strPrefix.end(), '.', '-');
 
     for (int iarg = 0; iarg < argc; ++iarg) {
         size_t len = 0;

src/networking.cpp

@@ -163,7 +163,6 @@ client *createClient(connection *conn, int iel) {
     c->multibulklen = 0;
     c->bulklen = -1;
     c->sentlen = 0;
-    c->sentlenAsync = 0;
     c->flags = 0;
     c->fPendingAsyncWrite = FALSE;
     c->fPendingAsyncWriteHandler = FALSE;

src/object.cpp

@@ -46,6 +46,7 @@ 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;
@@ -418,6 +419,7 @@ 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);
+            moduleFreeContext(io.ctx, false /* propogate */);
             zfree(io.ctx);
         }
 

src/readwritelock.h

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

src/redis-cli.c

@@ -2524,7 +2524,10 @@ cleanup:
 
 static int clusterManagerNodeConnect(clusterManagerNode *node) {
     if (node->context) redisFree(node->context);
-    node->context = redisConnect(node->ip, node->port);
+    struct timeval tv;
+    tv.tv_sec = config.cluster_manager_command.timeout / 1000;
+    tv.tv_usec = (config.cluster_manager_command.timeout % 1000) * 1000;
+    node->context = redisConnectWithTimeout(node->ip, node->port, tv);
     if (!node->context->err && config.tls) {
         const char *err = NULL;
         if (cliSecureConnection(node->context, config.sslconfig, &err) == REDIS_ERR && err) {
@@ -5626,7 +5629,10 @@ static int clusterManagerCommandImport(int argc, char **argv) {
     char *reply_err = NULL;
     redisReply *src_reply = NULL;
     // Connect to the source node.
-    redisContext *src_ctx = redisConnect(src_ip, src_port);
+    struct timeval tv;
+    tv.tv_sec = config.cluster_manager_command.timeout / 1000;
+    tv.tv_usec = (config.cluster_manager_command.timeout % 1000) * 1000;
+    redisContext *src_ctx = redisConnectWithTimeout(src_ip, src_port, tv);
     if (src_ctx->err) {
         success = 0;
         fprintf(stderr,"Could not connect to KeyDB at %s:%d: %s.\n", src_ip,

src/replication.cpp

@@ -188,6 +188,8 @@ 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__)
@@ -1017,8 +1019,6 @@ public:
     void addReplica(client *replica) {
         replicas.push_back(replica);
         replicationSetupSlaveForFullResync(replica,getPsyncInitialOffset());
-        // Optimize the socket for bulk transfer
-        //connDisableTcpNoDelay(replica->conn);
     }
 
     bool isActive() const { return !replicas.empty(); }
@@ -1091,7 +1091,7 @@ public:
         reply->used += size;
     }
 
-    void addLongWithPrefix(long val, char prefix) {
+    void addLongLongWithPrefix(long long val, char prefix) {
         char buf[128];
         int len;
 
@@ -1103,15 +1103,19 @@ public:
     }
 
     void addArrayLen(int len) {
-        addLongWithPrefix(len, '*');
+        addLongLongWithPrefix(len, '*');
     }
 
     void addLong(long val) {
-        addLongWithPrefix(val, ':');
+        addLongLongWithPrefix(val, ':');
+    }
+
+    void addLongLong(long long val) {
+        addLongLongWithPrefix(val, ':');
     }
 
     void addString(const char *s, unsigned long len) {
-        addLongWithPrefix(len, '$');
+        addLongLongWithPrefix(len, '$');
         addData(s, len);
         addData("\r\n", 2);
     }
@@ -1147,22 +1151,19 @@ int rdbSaveSnapshotForReplication(rdbSaveInfo *rsi) {
     // TODO: This needs to be on a background thread
     int retval = C_OK;
     serverAssert(GlobalLocksAcquired());
-    serverLog(LL_NOTICE, "Starting storage provider fast full sync with target: %s", "disk");
+    serverLog(LL_NOTICE, "Starting fast full sync with target: %s", "disk");
 
     std::shared_ptr<replicationBuffer> spreplBuf = std::make_shared<replicationBuffer>();
     listNode *ln;
     listIter li;
-    client *replica = nullptr;
     listRewind(g_pserver->slaves, &li);
-    while (replica == nullptr && (ln = listNext(&li))) {
+    while ((ln = listNext(&li))) {
         client *replicaCur = (client*)listNodeValue(ln);
-        if ((replicaCur->slave_capa & SLAVE_CAPA_ROCKSDB_SNAPSHOT) && (replicaCur->replstate == SLAVE_STATE_WAIT_BGSAVE_START)) {
-            replica = replicaCur;
-            spreplBuf->addReplica(replica);
+        if ((replicaCur->slave_capa & SLAVE_CAPA_KEYDB_FASTSYNC) && (replicaCur->replstate == SLAVE_STATE_WAIT_BGSAVE_START)) {
+            spreplBuf->addReplica(replicaCur);
             replicaCur->replstate = SLAVE_STATE_FASTSYNC_TX;
         }
     }
-    serverAssert(replica != nullptr);
 
     spreplBuf->addArrayLen(2);   // Two sections: Metadata and databases
 
@@ -1177,7 +1178,7 @@ int rdbSaveSnapshotForReplication(rdbSaveInfo *rsi) {
             spreplBuf->addString(rsi->repl_id, CONFIG_RUN_ID_SIZE);
         spreplBuf->addArrayLen(2);
             spreplBuf->addString("repl-offset", 11);
-            spreplBuf->addLong(rsi->master_repl_offset);
+            spreplBuf->addLongLong(rsi->master_repl_offset);
 
     if (dictSize(g_pserver->lua_scripts)) {
         dictEntry *de;
@@ -1193,13 +1194,13 @@ int rdbSaveSnapshotForReplication(rdbSaveInfo *rsi) {
         di = NULL; /* So that we don't release it again on error. */
     }
 
-    std::shared_ptr<std::vector<std::unique_ptr<const StorageCache>>> spvecspsnapshot = std::make_shared<std::vector<std::unique_ptr<const StorageCache>>>();
+    std::shared_ptr<std::vector<const redisDbPersistentDataSnapshot*>> spvecsnapshot = std::make_shared<std::vector<const redisDbPersistentDataSnapshot*>>();
     for (int idb = 0; idb < cserver.dbnum; ++idb) {
-        spvecspsnapshot->emplace_back(g_pserver->db[idb]->CloneStorageCache());
+        spvecsnapshot->emplace_back(g_pserver->db[idb]->createSnapshot(getMvccTstamp(), false));
     }
     aeReleaseLock();
 
-    g_pserver->asyncworkqueue->AddWorkFunction([spreplBuf = std::move(spreplBuf), spvecspsnapshot = std::move(spvecspsnapshot)]{
+    g_pserver->asyncworkqueue->AddWorkFunction([spreplBuf = std::move(spreplBuf), spvecsnapshot = std::move(spvecsnapshot)]{
         int retval = C_OK;
         auto timeStart = ustime();
         auto lastLogTime = timeStart;
@@ -1209,15 +1210,16 @@ int rdbSaveSnapshotForReplication(rdbSaveInfo *rsi) {
         // Databases
         replBuf.addArrayLen(cserver.dbnum);
         for (int idb = 0; idb < cserver.dbnum; ++idb) {
-            auto &spsnapshot = (*spvecspsnapshot)[idb];
-            size_t snapshotDeclaredCount = spsnapshot->count();
+            auto &spsnapshot = (*spvecsnapshot)[idb];
+            size_t snapshotDeclaredCount = spsnapshot->size();
             replBuf.addArrayLen(snapshotDeclaredCount);
             size_t count = 0;
-            bool result = spsnapshot->enumerate([&replBuf, &count, &cbData, &lastLogTime, &cbLastUpdate](const char *rgchKey, size_t cchKey, const void *rgchVal, size_t cchVal) -> bool{
+            bool result = spsnapshot->iterate_threadsafe([&replBuf, &count, &cbData, &lastLogTime, &cbLastUpdate](const char *strKey, robj_roptr o) -> bool{
                 replBuf.addArrayLen(2);
 
-                replBuf.addString(rgchKey, cchKey);
-                replBuf.addString((const char *)rgchVal, cchVal);
+                replBuf.addString(strKey, sdslen(strKey));
+                sds strT = serializeStoredObjectAndExpire(o);
+                replBuf.addString(strT, sdslen(strT));
                 ++count;
                 if ((count % 8092) == 0) {
                     auto curTime = ustime();
@@ -1228,7 +1230,8 @@ int rdbSaveSnapshotForReplication(rdbSaveInfo *rsi) {
                         lastLogTime = ustime();
                     }
                 }
-                cbData += cchKey + cchVal;
+                cbData += sdslen(strKey) + sdslen(strT);
+                sdsfree(strT);
                 return replBuf.isActive();
             });
 
@@ -1300,7 +1303,7 @@ int startBgsaveForReplication(int mincapa) {
     /* Only do rdbSave* when rsiptr is not NULL,
      * otherwise replica will miss repl-stream-db. */
     if (rsiptr) {
-        if (mincapa & SLAVE_CAPA_ROCKSDB_SNAPSHOT && g_pserver->m_pstorageFactory)
+        if (mincapa & SLAVE_CAPA_KEYDB_FASTSYNC && FFastSyncEnabled())
             retval = rdbSaveSnapshotForReplication(rsiptr);
         else if (socket_target)
             retval = rdbSaveToSlavesSockets(rsiptr);
@@ -1479,7 +1482,7 @@ void syncCommand(client *c) {
     }
 
     /* CASE 0: Fast Sync */
-    if ((c->slave_capa & SLAVE_CAPA_ROCKSDB_SNAPSHOT) && g_pserver->m_pstorageFactory) {
+    if (c->slave_capa & SLAVE_CAPA_KEYDB_FASTSYNC && FFastSyncEnabled()) {
         serverLog(LL_NOTICE,"Fast SYNC on next replication cycle");
     /* CASE 1: BGSAVE is in progress, with disk target. */
     } else if (g_pserver->FRdbSaveInProgress() &&
@@ -1670,8 +1673,8 @@ void replconfCommand(client *c) {
                 c->slave_capa |= SLAVE_CAPA_PSYNC2;
             else if (!strcasecmp((const char*)ptrFromObj(c->argv[j+1]), "activeExpire"))
                 c->slave_capa |= SLAVE_CAPA_ACTIVE_EXPIRE;
-            else if (!strcasecmp((const char*)ptrFromObj(c->argv[j+1]), "rocksdb-snapshot-load"))
-                c->slave_capa |= SLAVE_CAPA_ROCKSDB_SNAPSHOT;
+            else if (!strcasecmp((const char*)ptrFromObj(c->argv[j+1]), "keydb-fastsync"))
+                c->slave_capa |= SLAVE_CAPA_KEYDB_FASTSYNC;
 
             fCapaCommand = true;
         } else if (!strcasecmp((const char*)ptrFromObj(c->argv[j]),"ack")) {
@@ -1736,10 +1739,14 @@ void replconfCommand(client *c) {
 
     if (fCapaCommand) {
         sds reply = sdsnew("+OK");
-        if (g_pserver->fActiveReplica)
+        if (g_pserver->fActiveReplica) {
             reply = sdscat(reply, " active-replica");
-        if (g_pserver->m_pstorageFactory && (c->slave_capa & SLAVE_CAPA_ROCKSDB_SNAPSHOT) && !g_pserver->fActiveReplica)
-            reply = sdscat(reply, " rocksdb-snapshot-save");
+        }
+        if ((c->slave_capa & SLAVE_CAPA_KEYDB_FASTSYNC) && FFastSyncEnabled()) {
+            reply = sdscat(reply, " keydb-fastsync-save");
+        } else {
+            c->slave_capa = (c->slave_capa & (~SLAVE_CAPA_KEYDB_FASTSYNC)); // never try to fast sync for this as they won't expect it
+        }
         reply = sdscat(reply, "\r\n");
         addReplySds(c, reply);
     } else {
@@ -2513,6 +2520,7 @@ bool readSnapshotBulkPayload(connection *conn, redisMaster *mi, rdbSaveInfo &rsi
         }
     }
 
+    serverAssert(mi->parseState != nullptr);
     for (int iter = 0; iter < 10; ++iter) {
         if (mi->parseState->shouldThrottle())
             return false;
@@ -2523,11 +2531,16 @@ bool readSnapshotBulkPayload(connection *conn, redisMaster *mi, rdbSaveInfo &rsi
 
         auto nread = connRead(conn, mi->bulkreadBuffer+qblen, readlen);
         if (nread <= 0) {
-            if (connGetState(conn) != CONN_STATE_CONNECTED)
+            if (connGetState(conn) != CONN_STATE_CONNECTED) {
+                serverLog(LL_WARNING,"I/O error trying to sync with MASTER: %s",
+                    (nread == -1) ? strerror(errno) : "connection lost");
                 cancelReplicationHandshake(mi, true);
+            }
             return false;
         }
+        g_pserver->stat_net_input_bytes += nread;
         mi->repl_transfer_lastio = g_pserver->unixtime;
+        mi->repl_transfer_read += nread;
         sdsIncrLen(mi->bulkreadBuffer,nread);
 
         size_t offset = 0;
@@ -2599,8 +2612,9 @@ bool readSnapshotBulkPayload(connection *conn, redisMaster *mi, rdbSaveInfo &rsi
             rsi.repl_stream_db = mi->parseState->getMetaDataLongLong("repl-stream-db");
             rsi.repl_offset = mi->parseState->getMetaDataLongLong("repl-offset");
             sds str = mi->parseState->getMetaDataStr("repl-id");
-            if (sdslen(str) == CONFIG_RUN_ID_SIZE+1) {
+            if (sdslen(str) == CONFIG_RUN_ID_SIZE) {
                 memcpy(rsi.repl_id, str, CONFIG_RUN_ID_SIZE+1);
+                rsi.repl_id_is_set = 1;
             }
 
             fFinished = true;
@@ -2999,7 +3013,7 @@ void readSyncBulkPayload(connection *conn) {
         return;
     }
 
-    if (mi->isRocksdbSnapshotRepl) {
+    if (mi->isKeydbFastsync) {
         if (!readSnapshotBulkPayload(conn, mi, rsi))
             return;
     } else {
@@ -3012,17 +3026,15 @@ void readSyncBulkPayload(connection *conn) {
 
     /* Final setup of the connected slave <- master link */
     replicationCreateMasterClient(mi,mi->repl_transfer_s,rsi.repl_stream_db);
-    if (mi->isRocksdbSnapshotRepl) {
+    if (mi->isKeydbFastsync) {
         /* We need to handle the case where the initial querybuf data was read by fast sync */
         /* This should match the work readQueryFromClient would do for a master client */
         mi->master->querybuf = sdscatsds(mi->master->querybuf, mi->bulkreadBuffer);
+        mi->master->pending_querybuf = sdscatsds(mi->master->pending_querybuf, mi->bulkreadBuffer);
+        mi->master->read_reploff += sdslen(mi->bulkreadBuffer);
+
         sdsfree(mi->bulkreadBuffer);
         mi->bulkreadBuffer = nullptr;
-
-        mi->master->pending_querybuf = sdscatlen(mi->master->pending_querybuf,
-            mi->master->querybuf,sdslen(mi->master->querybuf));
-
-        mi->master->read_reploff += sdslen(mi->master->querybuf);
     }
     mi->repl_transfer_s = nullptr;
     mi->repl_state = REPL_STATE_CONNECTED;
@@ -3061,13 +3073,13 @@ void readSyncBulkPayload(connection *conn) {
     }
 
     /* Send the initial ACK immediately to put this replica in online state. */
-    if (usemark || mi->isRocksdbSnapshotRepl) replicationSendAck(mi);
+    if (usemark || mi->isKeydbFastsync) replicationSendAck(mi);
 
     /* Restart the AOF subsystem now that we finished the sync. This
      * will trigger an AOF rewrite, and when done will start appending
      * to the new file. */
     if (g_pserver->aof_enabled) restartAOFAfterSYNC();
-    if (mi->isRocksdbSnapshotRepl)
+    if (mi->isKeydbFastsync)
         readQueryFromClient(conn); // There may be querybuf data we just appeneded
     return;
 }
@@ -3393,15 +3405,15 @@ void parseMasterCapa(redisMaster *mi, sds strcapa)
     char *pchEnd = szStart;
 
     mi->isActive = false;
-    mi->isRocksdbSnapshotRepl = false;
+    mi->isKeydbFastsync = false;
     for (;;)
     {
         if (*pchEnd == ' ' || *pchEnd == '\0') {
             // Parse the word
             if (strncmp(szStart, "active-replica", pchEnd - szStart) == 0) {
                 mi->isActive = true;
-            } else if (strncmp(szStart, "rocksdb-snapshot-save", pchEnd - szStart) == 0) {
-                mi->isRocksdbSnapshotRepl = true;
+            } else if (strncmp(szStart, "keydb-fastsync-save", pchEnd - szStart) == 0) {
+                mi->isKeydbFastsync = true;
             }
             szStart = pchEnd + 1;
         }
@@ -3548,9 +3560,9 @@ retry_connect:
             "capa","psync2",
             "capa","activeExpire",
         };
-        if (g_pserver->m_pstorageFactory && !g_pserver->fActiveReplica && g_pserver->repl_diskless_load != REPL_DISKLESS_LOAD_SWAPDB) {
+        if (FFastSyncEnabled() && g_pserver->repl_diskless_load != REPL_DISKLESS_LOAD_SWAPDB) {
             veccapabilities.push_back("capa");
-            veccapabilities.push_back("rocksdb-snapshot-load");
+            veccapabilities.push_back("keydb-fastsync");
         }
 
         err = sendCommandArgv(conn, veccapabilities.size(), veccapabilities.data(), nullptr);
@@ -3733,12 +3745,12 @@ retry_connect:
     }
 
     /* Prepare a suitable temp file for bulk transfer */
-    if (!useDisklessLoad()) {
+    if (!useDisklessLoad() && !mi->isKeydbFastsync) {
         while(maxtries--) {
             auto dt = std::chrono::system_clock::now().time_since_epoch();
             auto dtMillisecond = std::chrono::duration_cast<std::chrono::milliseconds>(dt);
             snprintf(tmpfile,sizeof(tmpfile),
-                "temp-%d.%ld.rdb",(int)dtMillisecond.count(),(long int)getpid());
+                "temp-%lld.%ld.rdb",(long long)dtMillisecond.count(),(long int)getpid());
             dfd = open(tmpfile,O_CREAT|O_WRONLY|O_EXCL,0644);
             if (dfd != -1) break;
             sleep(1);
@@ -3862,6 +3874,10 @@ int cancelReplicationHandshake(redisMaster *mi, int reconnect) {
         delete mi->parseState;
         mi->parseState = nullptr;
     }
+    if (mi->bulkreadBuffer) {
+        sdsfree(mi->bulkreadBuffer);
+        mi->bulkreadBuffer = nullptr;
+    }
 
     if (mi->repl_state == REPL_STATE_TRANSFER) {
         replicationAbortSyncTransfer(mi);
@@ -4327,7 +4343,6 @@ void replicationCacheMaster(redisMaster *mi, client *c) {
     if (c->flags & CLIENT_MULTI) discardTransaction(c);
     listEmpty(c->reply);
     c->sentlen = 0;
-    c->sentlenAsync = 0;
     c->reply_bytes = 0;
     c->bufpos = 0;
     resetClient(c);

src/sds.h

@@ -418,7 +418,10 @@ public:
     sdsstring &operator=(const sdsstring &other)
     {
         sdsfree(m_str);
-        m_str = sdsdup(other.m_str);
+        if (other.m_str != nullptr)
+            m_str = sdsdup(other.m_str);
+        else
+            m_str = nullptr;
         return *this;
     }
 

src/server.cpp

@@ -1870,6 +1870,9 @@ VersionCompareResult compareVersion(SymVer *pver)
     if ((symVerThis.major == 0 && symVerThis.minor == 0 && symVerThis.build == 0)
         || (pver->major == 0 && pver->minor == 0 && pver->build == 0))
         return VersionCompareResult::EqualVersion;
+
+    if (pver->major <= 6 && pver->minor <= 3 && pver->build <= 3)
+        return VersionCompareResult::IncompatibleVersion;
     
     for (int iver = 0; iver < 3; ++iver)
     {
@@ -2134,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 (rehashes_per_ms > 0 && async_rehashes < 131 && !cserver.active_defrag_enabled && cserver.cthreads > 1 && dictSize(dict) > 2048 && dictIsRehashing(dict) && !g_pserver->loading) {
+                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) {
                     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;
@@ -5099,8 +5102,8 @@ int processCommand(client *c, int callFlags) {
 
             sds str = sdsempty();
             for (int j = 0; j < numkeys; j++) {
-                sdscatsds(str, (sds)ptrFromObj(c->argv[keyindex[j]]));
-                sdscat(str, " ");
+                str = sdscatsds(str, (sds)ptrFromObj(c->argv[keyindex[j]]));
+                str = sdscat(str, " ");
             }
         
             if (numkeys > 0)
@@ -5660,7 +5663,8 @@ sds genRedisInfoString(const char *section) {
             "lru_clock:%u\r\n"
             "executable:%s\r\n"
             "config_file:%s\r\n"
-            "availability_zone:%s\r\n",
+            "availability_zone:%s\r\n"
+            "features:%s\r\n",
             KEYDB_SET_VERSION,
             redisGitSHA1(),
             strtol(redisGitDirty(),NULL,10) > 0,
@@ -5687,7 +5691,8 @@ sds genRedisInfoString(const char *section) {
             lruclock,
             cserver.executable ? cserver.executable : "",
             cserver.configfile ? cserver.configfile : "",
-            g_pserver->sdsAvailabilityZone);
+            g_pserver->sdsAvailabilityZone,
+            "cluster_mget");
     }
 
     /* Clients */
@@ -7408,7 +7413,7 @@ static void validateConfiguration()
         exit(EXIT_FAILURE);
     }
 
-    g_pserver->repl_backlog_config_size = g_pserver->repl_backlog_size; // this is normally set in the update logic, but not on initial config
+    g_pserver->repl_backlog_size = g_pserver->repl_backlog_config_size; // this is normally set in the update logic, but not on initial config
 }
 
 int iAmMaster(void) {

src/server.h

@@ -620,7 +620,7 @@ typedef enum {
 #define SLAVE_CAPA_EOF (1<<0)    /* Can parse the RDB EOF streaming format. */
 #define SLAVE_CAPA_PSYNC2 (1<<1) /* Supports PSYNC2 protocol. */
 #define SLAVE_CAPA_ACTIVE_EXPIRE (1<<2) /* Will the slave perform its own expirations? (Don't send delete) */
-#define SLAVE_CAPA_ROCKSDB_SNAPSHOT (1<<3)
+#define SLAVE_CAPA_KEYDB_FASTSYNC (1<<3)
 
 /* Synchronous read timeout - replica side */
 #define CONFIG_REPL_SYNCIO_TIMEOUT 5
@@ -966,6 +966,7 @@ struct redisObjectExtended {
 
 typedef struct redisObject {
     friend redisObject *createEmbeddedStringObject(const char *ptr, size_t len);
+    friend redisObject *createObject(int type, void *ptr);
 protected:
     redisObject() {}
 
@@ -1609,7 +1610,6 @@ struct client {
     unsigned long long reply_bytes; /* Tot bytes of objects in reply list. */
     size_t sentlen;         /* Amount of bytes already sent in the current
                                buffer or object being sent. */
-    size_t sentlenAsync;    /* same as sentlen buf for async buffers (which are a different stream) */
     time_t ctime;           /* Client creation time. */
     long duration;          /* Current command duration. Used for measuring latency of blocking/non-blocking cmds */
     time_t lastinteraction; /* Time of the last interaction, used for timeout */
@@ -1870,7 +1870,7 @@ struct redisMaster {
     long long master_initial_offset;           /* Master PSYNC offset. */
 
     bool isActive = false;
-    bool isRocksdbSnapshotRepl = false;
+    bool isKeydbFastsync = false;
     int repl_state;          /* Replication status if the instance is a replica */
     off_t repl_transfer_size; /* Size of RDB to read from master during sync. */
     off_t repl_transfer_read; /* Amount of RDB read from master during sync. */
@@ -2305,6 +2305,7 @@ 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. */
@@ -2686,6 +2687,7 @@ struct redisServer {
 
     int fActiveReplica;                          /* Can this replica also be a master? */
     int fWriteDuringActiveLoad;                  /* Can this active-replica write during an RDB load? */
+    int fEnableFastSync = false;
 
     // Format:
     //  Lower 20 bits: a counter incrementing for each command executed in the same millisecond
@@ -2917,7 +2919,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);
+void moduleFreeContext(struct RedisModuleCtx *ctx, bool propogate = true);
 void unblockClientFromModule(client *c);
 void moduleHandleBlockedClients(int iel);
 void moduleBlockedClientTimedOut(client *c);
@@ -3180,6 +3182,7 @@ void trimStringObjectIfNeeded(robj *o);
 robj *deserializeStoredObject(const void *data, size_t cb);
 std::unique_ptr<expireEntry> deserializeExpire(const char *str, size_t cch, size_t *poffset);
 sds serializeStoredObject(robj_roptr o, sds sdsPrefix = nullptr);
+sds serializeStoredObjectAndExpire(robj_roptr o);
 
 #define sdsEncodedObject(objptr) (objptr->encoding == OBJ_ENCODING_RAW || objptr->encoding == OBJ_ENCODING_EMBSTR)
 
@@ -3921,7 +3924,7 @@ void updateActiveReplicaMastersFromRsi(rdbSaveInfo *rsi);
 uint64_t getMvccTstamp();
 void incrementMvccTstamp();
 
-#if __GNUC__ >= 7 && !defined(NO_DEPRECATE_FREE)
+#if __GNUC__ >= 7 && !defined(NO_DEPRECATE_FREE) && !defined(ALPINE)
  [[deprecated]]
 void *calloc(size_t count, size_t size) noexcept;
  [[deprecated]]
@@ -3969,6 +3972,10 @@ inline int ielFromEventLoop(const aeEventLoop *eventLoop)
     return iel;
 }
 
+inline bool FFastSyncEnabled() {
+    return g_pserver->fEnableFastSync && !g_pserver->fActiveReplica;
+}
+
 inline int FCorrectThread(client *c)
 {
     return (c->conn == nullptr)

src/snapshot.cpp

@@ -587,8 +587,12 @@ bool redisDbPersistentDataSnapshot::iterate_threadsafe_core(std::function<bool(c
                 if (!fKeyOnly)
                 {
                     size_t offset = 0;
-                    deserializeExpire((const char*)data, cbData, &offset);
+                    std::unique_ptr<expireEntry> spexpire = 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

@@ -7,6 +7,8 @@
 #include "../cluster.h"
 #include "rocksdbfactor_internal.h"
 
+template class std::basic_string<char>;
+
 static const char keyprefix[] = INTERNAL_KEY_PREFIX;
 
 rocksdb::Options DefaultRocksDBOptions();
@@ -24,8 +26,9 @@ bool FInternalKey(const char *key, size_t cch)
 
 std::string getPrefix(unsigned int hashslot)
 {
-    char *hash_char = (char *)&hashslot;
-    return std::string(hash_char + (sizeof(unsigned int) - 2), 2);
+    HASHSLOT_PREFIX_TYPE slot = HASHSLOT_PREFIX_ENDIAN((HASHSLOT_PREFIX_TYPE)hashslot);
+    char *hash_char = (char *)&slot;
+    return std::string(hash_char, HASHSLOT_PREFIX_BYTES);
 }
 
 std::string prefixKey(const char *key, size_t cchKey)
@@ -33,8 +36,8 @@ std::string prefixKey(const char *key, size_t cchKey)
     return FInternalKey(key, cchKey) ? std::string(key, cchKey) : getPrefix(keyHashSlot(key, cchKey)) + std::string(key, cchKey);
 }
 
-RocksDBStorageProvider::RocksDBStorageProvider(RocksDBStorageFactory *pfactory, std::shared_ptr<rocksdb::DB> &spdb, std::shared_ptr<rocksdb::ColumnFamilyHandle> &spcolfam, const rocksdb::Snapshot *psnapshot, size_t count)
-    : m_pfactory(pfactory), m_spdb(spdb), m_psnapshot(psnapshot), m_spcolfamily(spcolfam), m_count(count)
+RocksDBStorageProvider::RocksDBStorageProvider(RocksDBStorageFactory *pfactory, std::shared_ptr<rocksdb::DB> &spdb, std::shared_ptr<rocksdb::ColumnFamilyHandle> &spcolfam, std::shared_ptr<rocksdb::ColumnFamilyHandle> &spexpirecolfam, const rocksdb::Snapshot *psnapshot, size_t count)
+    : m_pfactory(pfactory), m_spdb(spdb), m_psnapshot(psnapshot), m_spcolfamily(spcolfam), m_spexpirecolfamily(spexpirecolfam), m_count(count)
 {
     m_readOptionsTemplate = rocksdb::ReadOptions();
     m_readOptionsTemplate.verify_checksums = false;
@@ -144,16 +147,28 @@ void RocksDBStorageProvider::retrieve(const char *key, size_t cchKey, callbackSi
 size_t RocksDBStorageProvider::clear()
 {
     size_t celem = count();
+    auto options = m_spdb->GetOptions(m_spcolfamily.get());
     auto status = m_spdb->DropColumnFamily(m_spcolfamily.get());
     auto strName = m_spcolfamily->GetName();
 
     rocksdb::ColumnFamilyHandle *handle = nullptr;
-    rocksdb::ColumnFamilyOptions cf_options(m_pfactory->RocksDbOptions());
+    rocksdb::ColumnFamilyOptions cf_options(options);
+
     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;
 }
@@ -172,7 +187,7 @@ bool RocksDBStorageProvider::enumerate(callback fn) const
         if (FInternalKey(it->key().data(), it->key().size()))
             continue;
         ++count;
-        bool fContinue = fn(it->key().data()+2, it->key().size()-2, it->value().data(), it->value().size());
+        bool fContinue = fn(it->key().data()+HASHSLOT_PREFIX_BYTES, it->key().size()-HASHSLOT_PREFIX_BYTES, it->value().data(), it->value().size());
         if (!fContinue)
             break;
     }
@@ -191,17 +206,17 @@ bool RocksDBStorageProvider::enumerate_hashslot(callback fn, unsigned int hashsl
     std::string prefix = getPrefix(hashslot);
     std::unique_ptr<rocksdb::Iterator> it = std::unique_ptr<rocksdb::Iterator>(m_spdb->NewIterator(ReadOptions(), m_spcolfamily.get()));
     size_t count = 0;
-    for (it->Seek(prefix.c_str()); it->Valid(); it->Next()) {
+    for (it->Seek(prefix); it->Valid(); it->Next()) {
         if (FInternalKey(it->key().data(), it->key().size()))
             continue;
-        if (strncmp(it->key().data(),prefix.c_str(),2) != 0)
+        if (HASHSLOT_PREFIX_RECOVER(*(HASHSLOT_PREFIX_TYPE *)it->key().data()) != hashslot)
             break;
         ++count;
-        bool fContinue = fn(it->key().data()+2, it->key().size()-2, it->value().data(), it->value().size());
+        bool fContinue = fn(it->key().data()+HASHSLOT_PREFIX_BYTES, it->key().size()-HASHSLOT_PREFIX_BYTES, it->value().data(), it->value().size());
         if (!fContinue)
             break;
     }
-    bool full_iter = !it->Valid() || (strncmp(it->key().data(),prefix.c_str(),2) != 0);
+    bool full_iter = !it->Valid() || (HASHSLOT_PREFIX_RECOVER(*(HASHSLOT_PREFIX_TYPE *)it->key().data()) != hashslot);
     if (full_iter && count != g_pserver->cluster->slots_keys_count[hashslot])
     {
         printf("WARNING: rocksdb hashslot count mismatch");
@@ -211,11 +226,81 @@ bool RocksDBStorageProvider::enumerate_hashslot(callback fn, unsigned int hashsl
     return full_iter;
 }
 
+void RocksDBStorageProvider::setExpire(const char *key, size_t cchKey, long long expire)
+{
+    rocksdb::Status status;
+    std::unique_lock<fastlock> l(m_lock);
+    long long beExpire = htobe64(expire);
+    std::string prefix((const char *)&beExpire,sizeof(long long));
+    std::string strKey(key, cchKey);
+    if (m_spbatch != nullptr)
+        status = m_spbatch->Put(m_spexpirecolfamily.get(), rocksdb::Slice(prefix + strKey), rocksdb::Slice(strKey));
+    else
+        status = m_spdb->Put(WriteOptions(), m_spexpirecolfamily.get(), rocksdb::Slice(prefix + strKey), rocksdb::Slice(strKey));
+    if (!status.ok())
+        throw status.ToString();
+}
+
+void RocksDBStorageProvider::removeExpire(const char *key, size_t cchKey, long long expire)
+{
+    rocksdb::Status status;
+    std::unique_lock<fastlock> l(m_lock);
+    long long beExpire = htobe64(expire);
+    std::string prefix((const char *)&beExpire,sizeof(long long));
+    std::string strKey(key, cchKey);
+    std::string fullKey = prefix + strKey;
+    if (!FExpireExists(fullKey))
+        return;
+    if (m_spbatch)
+        status = m_spbatch->Delete(m_spexpirecolfamily.get(), rocksdb::Slice(fullKey));
+    else
+        status = m_spdb->Delete(WriteOptions(), m_spexpirecolfamily.get(), rocksdb::Slice(fullKey));
+    if (!status.ok())
+        throw status.ToString();
+}
+
+std::vector<std::string> RocksDBStorageProvider::getExpirationCandidates(unsigned int count)
+{
+    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()) {
+        if (FInternalKey(it->key().data(), it->key().size()))
+            continue;
+        result.emplace_back(it->value().data(), it->value().size());
+    }
+    return result;
+}
+
+std::string randomHashSlot() {
+    return getPrefix(genrand64_int63() % (1 << 16));
+}
+
+std::vector<std::string> RocksDBStorageProvider::getEvictionCandidates(unsigned int count)
+{
+    std::vector<std::string> result;
+    if (g_pserver->maxmemory_policy & MAXMEMORY_FLAG_ALLKEYS) {
+        std::unique_ptr<rocksdb::Iterator> it = std::unique_ptr<rocksdb::Iterator>(m_spdb->NewIterator(ReadOptions(), m_spcolfamily.get()));
+        for (it->Seek(randomHashSlot()); it->Valid() && result.size() < count; it->Next()) {
+            if (FInternalKey(it->key().data(), it->key().size()))
+                continue;
+            result.emplace_back(it->key().data() + HASHSLOT_PREFIX_BYTES, it->key().size() - HASHSLOT_PREFIX_BYTES);
+        }
+    } 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()) {
+            if (FInternalKey(it->key().data(), it->key().size()))
+                continue;
+            result.emplace_back(it->value().data(), it->value().size());
+        }
+    }
+    return result;
+}
+
 const IStorage *RocksDBStorageProvider::clone() const
 {
     std::unique_lock<fastlock> l(m_lock);
     const rocksdb::Snapshot *psnapshot = const_cast<RocksDBStorageProvider*>(this)->m_spdb->GetSnapshot();
-    return new RocksDBStorageProvider(m_pfactory, const_cast<RocksDBStorageProvider*>(this)->m_spdb, const_cast<RocksDBStorageProvider*>(this)->m_spcolfamily, psnapshot, m_count);
+    return new RocksDBStorageProvider(m_pfactory, const_cast<RocksDBStorageProvider*>(this)->m_spdb, const_cast<RocksDBStorageProvider*>(this)->m_spcolfamily, const_cast<RocksDBStorageProvider*>(this)->m_spexpirecolfamily, psnapshot, m_count);
 }
 
 RocksDBStorageProvider::~RocksDBStorageProvider()
@@ -268,6 +353,7 @@ void RocksDBStorageProvider::batch_unlock()
 void RocksDBStorageProvider::flush()
 {
     m_spdb->SyncWAL();
+    m_spdb->Flush(rocksdb::FlushOptions());
 }
 
 bool RocksDBStorageProvider::FKeyExists(std::string& key) const
@@ -276,4 +362,12 @@ bool RocksDBStorageProvider::FKeyExists(std::string& key) const
     if (m_spbatch)
         return m_spbatch->GetFromBatchAndDB(m_spdb.get(), ReadOptions(), m_spcolfamily.get(), rocksdb::Slice(key), &slice).ok();
     return m_spdb->Get(ReadOptions(), m_spcolfamily.get(), rocksdb::Slice(key), &slice).ok();
+}
+
+bool RocksDBStorageProvider::FExpireExists(std::string& key) const
+{
+    rocksdb::PinnableSlice slice;
+    if (m_spbatch)
+        return m_spbatch->GetFromBatchAndDB(m_spdb.get(), ReadOptions(), m_spexpirecolfamily.get(), rocksdb::Slice(key), &slice).ok();
+    return m_spdb->Get(ReadOptions(), m_spexpirecolfamily.get(), rocksdb::Slice(key), &slice).ok();
 }

src/storage/rocksdb.h

@@ -4,12 +4,25 @@
 #include "../IStorage.h"
 #include <rocksdb/db.h>
 #include <rocksdb/utilities/write_batch_with_index.h>
+#ifdef __APPLE__
+#include <libkern/OSByteOrder.h>
+#define htole16 OSSwapHostToLittleInt16
+#define le16toh OSSwapLittleToHostInt16
+#define htobe64 OSSwapHostToBigInt64
+#else
+#include <endian.h>
+#endif
 #include "../fastlock.h"
 
 #define INTERNAL_KEY_PREFIX "\x00\x04\x03\x00\x05\x02\x04"
+#define HASHSLOT_PREFIX_TYPE uint16_t
+#define HASHSLOT_PREFIX_BYTES sizeof(HASHSLOT_PREFIX_TYPE)
+#define HASHSLOT_PREFIX_ENDIAN htole16
+#define HASHSLOT_PREFIX_RECOVER le16toh
 static const char count_key[] = INTERNAL_KEY_PREFIX "__keydb__count\1";
 static const char version_key[] = INTERNAL_KEY_PREFIX "__keydb__version\1";
 static const char meta_key[] = INTERNAL_KEY_PREFIX "__keydb__metadata\1";
+static const char last_expire_key[] = INTERNAL_KEY_PREFIX "__keydb__last_expire_time";
 class RocksDBStorageFactory;
 
 class RocksDBStorageProvider : public IStorage
@@ -19,12 +32,13 @@ class RocksDBStorageProvider : public IStorage
     std::unique_ptr<rocksdb::WriteBatchWithIndex> m_spbatch;
     const rocksdb::Snapshot *m_psnapshot = nullptr;
     std::shared_ptr<rocksdb::ColumnFamilyHandle> m_spcolfamily;
+    std::shared_ptr<rocksdb::ColumnFamilyHandle> m_spexpirecolfamily;
     rocksdb::ReadOptions m_readOptionsTemplate;
     size_t m_count = 0;
     mutable fastlock m_lock {"RocksDBStorageProvider"};
 
 public:
-    RocksDBStorageProvider(RocksDBStorageFactory *pfactory, std::shared_ptr<rocksdb::DB> &spdb, std::shared_ptr<rocksdb::ColumnFamilyHandle> &spcolfam, const rocksdb::Snapshot *psnapshot, size_t count);
+    RocksDBStorageProvider(RocksDBStorageFactory *pfactory, std::shared_ptr<rocksdb::DB> &spdb, std::shared_ptr<rocksdb::ColumnFamilyHandle> &spcolfam, std::shared_ptr<rocksdb::ColumnFamilyHandle> &spexpirecolfam, const rocksdb::Snapshot *psnapshot, size_t count);
     ~RocksDBStorageProvider();
 
     virtual void insert(const char *key, size_t cchKey, void *data, size_t cb, bool fOverwrite) override;
@@ -34,6 +48,11 @@ 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> 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;
+
     virtual const IStorage *clone() const override;
 
     virtual void beginWriteBatch() override;
@@ -50,6 +69,7 @@ public:
 
 protected:
     bool FKeyExists(std::string&) const;
+    bool FExpireExists(std::string&) const;
 
     const rocksdb::ReadOptions &ReadOptions() const { return m_readOptionsTemplate; }
     rocksdb::WriteOptions WriteOptions() const;

src/storage/rocksdbfactor_internal.h

@@ -5,6 +5,7 @@ class RocksDBStorageFactory : public IStorageFactory
 {
     std::shared_ptr<rocksdb::DB> m_spdb;    // Note: This must be first so it is deleted last
     std::vector<std::unique_ptr<rocksdb::ColumnFamilyHandle>> m_vecspcols;
+    std::vector<std::unique_ptr<rocksdb::ColumnFamilyHandle>> m_vecspexpirecols;
     std::shared_ptr<rocksdb::SstFileManager> m_pfilemanager;
     std::string m_path;
     bool m_fCreatedTempFolder = false;

src/storage/rocksdbfactory.cpp

@@ -61,8 +61,8 @@ RocksDBStorageFactory::RocksDBStorageFactory(const char *dbfile, int dbnum, cons
     auto status = rocksdb::DB::ListColumnFamilies(rocksdb::Options(), dbfile, &vecT);
     // RocksDB requires we know the count of col families before opening, if the user only wants to see less
     //  we still have to make room for all column family handles regardless
-    if (status.ok() && (int)vecT.size() > dbnum)
-        dbnum = (int)vecT.size();
+    if (status.ok() && (int)vecT.size()/2 > dbnum)
+        dbnum = (int)vecT.size()/2;
 
     std::vector<rocksdb::ColumnFamilyDescriptor> veccoldesc;
     veccoldesc.push_back(rocksdb::ColumnFamilyDescriptor(rocksdb::kDefaultColumnFamilyName, rocksdb::ColumnFamilyOptions()));  // ignore default col family
@@ -72,13 +72,14 @@ RocksDBStorageFactory::RocksDBStorageFactory(const char *dbfile, int dbnum, cons
     rocksdb::DB *db = nullptr;
 
     auto options = RocksDbOptions();
-    options.prefix_extractor.reset(rocksdb::NewFixedPrefixTransform(2));
+    options.prefix_extractor.reset(rocksdb::NewFixedPrefixTransform(HASHSLOT_PREFIX_BYTES));
 
     for (int idb = 0; idb < dbnum; ++idb)
     {
         rocksdb::ColumnFamilyOptions cf_options(options);
         cf_options.level_compaction_dynamic_level_bytes = true;
         veccoldesc.push_back(rocksdb::ColumnFamilyDescriptor(std::to_string(idb), cf_options));
+        veccoldesc.push_back(rocksdb::ColumnFamilyDescriptor(std::to_string(idb) + "_expires", cf_options));
     }
 
     if (rgchConfig != nullptr)
@@ -100,23 +101,29 @@ RocksDBStorageFactory::RocksDBStorageFactory(const char *dbfile, int dbnum, cons
     m_spdb = std::shared_ptr<rocksdb::DB>(db);
     for (auto handle : handles)
     {
-        std::string strVersion;
-        auto status = m_spdb->Get(rocksdb::ReadOptions(), handle, rocksdb::Slice(version_key, sizeof(version_key)), &strVersion);
-        if (!status.ok())
-        {
-            setVersion(handle);
-        }
-        else
-        {
-            SymVer ver = parseVersion(strVersion.c_str());
-            auto cmp = compareVersion(&ver);
-            if (cmp == NewerVersion)
-                throw "Cannot load FLASH database created by newer version of KeyDB";
-            if (cmp == OlderVersion)
+        if (handle->GetName().size() > 7 && !strncmp(handle->GetName().substr(handle->GetName().size() - 7).c_str(), "expires", 7)) {
+            m_vecspexpirecols.emplace_back(handle);
+        } else {
+            std::string strVersion;
+            auto status = m_spdb->Get(rocksdb::ReadOptions(), handle, rocksdb::Slice(version_key, sizeof(version_key)), &strVersion);
+            if (!status.ok())
+            {
                 setVersion(handle);
+            }
+            else
+            {
+                SymVer ver = parseVersion(strVersion.c_str());
+                auto cmp = compareVersion(&ver);
+                if (cmp == NewerVersion)
+                    throw "Cannot load FLASH database created by newer version of KeyDB";
+                if (cmp == IncompatibleVersion)
+                    throw "Cannot load FLASH database from before 6.3.4";
+                if (cmp == OlderVersion)
+                    setVersion(handle);
+            }
+            m_vecspcols.emplace_back(handle);
         }
-        m_vecspcols.emplace_back(handle);
-    }   
+    }
 }
 
 RocksDBStorageFactory::~RocksDBStorageFactory()
@@ -156,6 +163,7 @@ IStorage *RocksDBStorageFactory::create(int db, key_load_iterator iter, void *pr
 {
     ++db;   // skip default col family
     std::shared_ptr<rocksdb::ColumnFamilyHandle> spcolfamily(m_vecspcols[db].release());
+    std::shared_ptr<rocksdb::ColumnFamilyHandle> spexpirecolfamily(m_vecspexpirecols[db].release());
     size_t count = 0;
     bool fUnclean = false;
     
@@ -188,11 +196,11 @@ IStorage *RocksDBStorageFactory::create(int db, key_load_iterator iter, void *pr
                 printf("\tDatabase %d was not shutdown cleanly, recomputing metrics\n", db);
             fFirstRealKey = false;
             if (iter != nullptr)
-                iter(it->key().data()+2, it->key().size()-2, privdata);
+                iter(it->key().data()+HASHSLOT_PREFIX_BYTES, it->key().size()-HASHSLOT_PREFIX_BYTES, privdata);
             ++count;
         }
     }
-    return new RocksDBStorageProvider(this, m_spdb, spcolfamily, nullptr, count);
+    return new RocksDBStorageProvider(this, m_spdb, spcolfamily, spexpirecolfamily, nullptr, count);
 }
 
 const char *RocksDBStorageFactory::name() const

src/storage/teststorageprovider.h

@@ -1,6 +1,7 @@
 #include "../IStorage.h"
 #include <string>
 #include <unordered_map>
+#include <vector>
 
 class TestStorageFactory : public IStorageFactory
 {
@@ -28,6 +29,11 @@ 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> 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 {}
+
     virtual void flush() override;
 
     /* This is permitted to be a shallow clone */

src/t_zset.cpp

@@ -2036,7 +2036,7 @@ struct zsetopsrc {
                 zset *zs;
                 zskiplistNode *node;
             } sl;
-        } zset;
+        } iterzset;
     } iter;
 };
 
@@ -2085,7 +2085,7 @@ void zuiInitIterator(zsetopsrc *op) {
         /* Sorted sets are traversed in reverse order to optimize for
          * the insertion of the elements in a new list as in
          * ZDIFF/ZINTER/ZUNION */
-        iterzset *it = (iterzset*)&op->iter.zset;
+        iterzset *it = (iterzset*)&op->iter.iterzset;
         if (op->encoding == OBJ_ENCODING_ZIPLIST) {
             it->zl.zl = (unsigned char*)op->subject->m_ptr;
             it->zl.eptr = ziplistIndex(it->zl.zl,-2);
@@ -2118,7 +2118,7 @@ void zuiClearIterator(zsetopsrc *op) {
             serverPanic("Unknown set encoding");
         }
     } else if (op->type == OBJ_ZSET) {
-        iterzset *it = &op->iter.zset;
+        iterzset *it = &op->iter.iterzset;
         if (op->encoding == OBJ_ENCODING_ZIPLIST) {
             UNUSED(it); /* skip */
         } else if (op->encoding == OBJ_ENCODING_SKIPLIST) {
@@ -2194,7 +2194,7 @@ int zuiNext(zsetopsrc *op, zsetopval *val) {
             serverPanic("Unknown set encoding");
         }
     } else if (op->type == OBJ_ZSET) {
-        iterzset *it = &op->iter.zset;
+        iterzset *it = &op->iter.iterzset;
         if (op->encoding == OBJ_ENCODING_ZIPLIST) {
             /* No need to check both, but better be explicit. */
             if (it->zl.eptr == NULL || it->zl.sptr == NULL)

src/version.h

@@ -1,5 +1,5 @@
-#define KEYDB_REAL_VERSION "255.255.255"
-#define KEYDB_VERSION_NUM 0x00ffffff
+#define KEYDB_REAL_VERSION "6.3.4"
+#define KEYDB_VERSION_NUM 0x00060304
 extern const char *KEYDB_SET_VERSION;   // Unlike real version, this can be overriden by the config
 
 enum VersionCompareResult
@@ -7,6 +7,7 @@ enum VersionCompareResult
     EqualVersion,
     OlderVersion,
     NewerVersion,
+    IncompatibleVersion,
 };
 
 struct SymVer

tests/integration/replication.tcl

@@ -584,6 +584,7 @@ start_server {tags {"repl"}} {
     set master [srv 0 client]
     $master config set repl-diskless-sync yes
     $master config set repl-diskless-sync-delay 1
+    $master config set enable-keydb-fastsync no
     set master_host [srv 0 host]
     set master_port [srv 0 port]
     set master_pid [srv 0 pid]
@@ -727,6 +728,8 @@ start_server {tags {"repl"}} {
             }
         }
     }
+
+    $master config set enable-keydb-fastsync yes
 }
 
 if 0 {
@@ -867,8 +870,10 @@ test {replicaof right after disconnection} {
 test {Kill rdb child process if its dumping RDB is not useful} {
     start_server {tags {"repl"}} {
         set slave1 [srv 0 client]
+        $slave1 config set enable-keydb-fastsync no
         start_server {} {
             set slave2 [srv 0 client]
+            $slave2 config set enable-keydb-fastsync no
             start_server {} {
                 set master [srv 0 client]
                 set master_host [srv 0 host]

tests/unit/flash.tcl

@@ -74,8 +74,11 @@ if {$::flash_enabled} {
             r set testkey foo ex 1
             r flushall cache
             assert_equal {1} [r dbsize]
-            after 1500
-            assert_equal {0} [r dbsize]
+            wait_for_condition 50 1000 {
+                [r dbsize] == 0
+            } else {
+                fail "key is not expired"
+            }
         }
 
         test { SUBKEY EXPIRE persists after cache flush } {
@@ -140,17 +143,17 @@ if {$::flash_enabled} {
                     r set $numkeys xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
                     incr numkeys
                     if {[s used_memory]+1024 >= $limit} {
-                    break
+                        break
                     }
                 }
                 # Add additional keys to force eviction
                 # should still be under the limit for maxmemory, however all keys set should still exist between flash and memory
                 # check same number of keys exist in addition to values of first and last keys
                 set err 0
-            set extra_keys [expr floor([expr ($limit * 0.4) / 1024])]
+                set extra_keys [expr floor([expr ($limit * 0.4) / 1024])]
                 for {set j 0} {$j < $extra_keys} {incr j} {
                     catch {
-                    r set p2$j xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
+                        r set p2$j xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
                     } err
                     assert {$err == {OK}}
                 }
@@ -165,7 +168,49 @@ if {$::flash_enabled} {
                 assert {[r get last] == {val}}
                 r flushall
             }
-        }
 
+            test "FLASH - is flash eviction working? (policy $policy)" {
+                # Get the current memory limit and calculate a new limit.
+                # Set limit to 100M.
+                set used [s used_memory]
+                set limit [expr {$used+60*1024*1024}]
+                r config set maxmemory $limit
+                r config set maxmemory-policy $policy
+                # Now add keys equivalent to 1024b until the limit is almost reached.
+                set numkeys 0
+                r set first val
+                while 1 {
+                    r set $numkeys xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
+                    incr numkeys
+                    if {[s used_memory]+1024 >= $limit} {
+                        break
+                    }
+                }
+                # Add additional keys to force eviction
+                # should still be under the limit for maxmemory, however all keys set should still exist between flash and memory
+                # check same number of keys exist in addition to values of first and last keys
+                set err 0
+                set extra_keys [expr floor([expr ($limit * 0.4) / 1024])]
+                for {set j 0} {$j < $extra_keys} {incr j} {
+                    catch {
+                        r set p2$j xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
+                    } err
+                    assert {$err == {OK}}
+                }
+                if {[log_file_matches [srv 0 stdout] "*Failed to evict*"]} {
+                    fail "Server did not evict cleanly (detected full flush)"
+                }
+                r set last val
+                r debug flush-storage
+                r config set maxstorage 1
+                r config set maxmemory 1
+                set dbsize [r dbsize]
+                # after setting maxstorage and memory below used amount we should evict from storage provider
+                assert {$dbsize < $numkeys+$extra_keys+2}
+                r config set maxstorage 0
+                r config set maxmemory 0
+                r flushall
+            }
+        }
     }
 }

tests/unit/maxmemory.tcl

@@ -182,6 +182,8 @@ proc test_slave_buffers {test_name cmd_count payload_len limit_memory pipeline}
                 fail "Replication not started."
             }
 
+            after 100
+
             # measure used memory after the slave connected and set maxmemory
             set orig_used [s -1 used_memory]
             set orig_client_buf [s -1 mem_clients_normal]

tests/unit/moduleapi/load.tcl

@@ -1,7 +1,7 @@
 set testmodule [file normalize tests/modules/load.so]
 
 if {$::flash_enabled} {
-    start_server [list tags [list "modules"] overrides [list storage-provider {flash ./rocks.db.master} databases 256 loadmodule $testmodule]] {
+    start_server [list tags [list "modules"] overrides [list storage-provider {flash ./rocks.db.master.load.test} databases 256 loadmodule $testmodule]] {
         test "Module is notified of keys loaded from flash" {
             r flushall
             r set foo bar

tests/unit/tls-auditlog.tcl

@@ -51,7 +51,7 @@ if {$::tls} {
         }
     }
 
-    test {TLS: Able to connect while with SAN having a comprehensive list} {
+    test {TLS Audit Log: Able to connect while with SAN having a comprehensive list} {
         start_server {tags {"tls"} overrides {tls-auditlog-blocklist {dummy.keydb.dev san2.keydb.dev other.keydb.dev}}} {
             catch {r PING} e
             assert_match {PONG} $e
@@ -65,7 +65,7 @@ if {$::tls} {
         }
     }
 
-    test {TLS Audit LogTLS: Able to connect while with SAN having a comprehensive list with wildcards} {
+    test {TLS Audit Log: Able to connect while with SAN having a comprehensive list with wildcards} {
         start_server {tags {"tls"} overrides {tls-auditlog-blocklist {dummy.* san*.dev other.*}}} {
             catch {r PING} e
             assert_match {PONG} $e