Merge remote-tracking branch 'internal/main' into merge_internal

build.yaml

@@ -21,25 +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 --tls --skipunit unit/memefficiency
-      tests-with-flash:
-        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 --verbose --skipunit unit/memefficiency --flash
+        command: ./runtest --clients 4 --verbose 
       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 --tls
-      cluster-test-flash:
-        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 --tls --flash
+        command: ./runtest-cluster 
       sentinel-test:
         type: cmd
         parent: make-build
@@ -71,7 +59,7 @@ machamp:
       # to ensure a clearer docker build env
       code-checkout:
         type: cmd
-        command: echo checkout && git submodule init && git submodule update
+        command: echo checkout
         # default machamp builder image does not work for multi arch
         builder_image: us.gcr.io/snapchat-build-artifacts/prod/snapchat/img/ubuntu/ubuntu-23-04@sha256:bd43177a80e6ce1c3583e8ea959b88a9081c0f56b765ec9c5a157c27a637c23b
       docker:

ci.yaml

@@ -26,16 +26,6 @@ on:
           # references a build defined in build.yaml
           build_name: keydb-docker-build
           arch_types: ["amd64", "arm64"]
-  # Doc: go/cool-guide
-  cool:
-    workflows:
-      - workflow_type: backend_workflow
-        build_name: keydb-build
-        arch_types: ["amd64", "arm64"]
-      - workflow_type: backend_workflow
-        # references a build defined in build.yaml
-        build_name: keydb-docker-build
-        arch_types: ["amd64", "arm64"]
 
 # below defines which branch is release branch / release tag
 machamp:

machamp_scripts/Dockerfile

@@ -24,7 +24,7 @@ RUN set -eux; \
         gosu nobody true
 # build KeyDB
 ARG MAKE_JOBS=""
-ARG ENABLE_FLASH="yes"
+ARG ENABLE_FLASH=""
 COPY . /tmp/keydb-internal
 RUN set -eux; \
         cd /tmp/keydb-internal; \
@@ -73,25 +73,7 @@ RUN set -eux; \
                | xargs -r apt-mark manual \
         ; \
         apt-get purge -y --auto-remove -o APT::AutoRemove::RecommendsImportant=false; \
-        rm -rf /var/lib/apt/lists/*
-# build RedisBloom
-RUN set -eux; \
-        savedAptMark="$(apt-mark showmanual)"; \
-        apt-get update; \
-        DEBIAN_FRONTEND=noninteractive apt-get -o Dpkg::Options::="--force-confnew" install -qqy --no-install-recommends \
-                ca-certificates \
-                wget \
-                make \
-                pkg-config \
-                build-essential \
-                git; \
-        git clone --recursive https://github.com/RedisBloom/RedisBloom.git; \
-        cd RedisBloom; \
-        ./sbin/setup; \
-        bash -l; \
-        make; \
-        redisBloomPath=`make run -n | awk '{print $NF}'`; \
-        cp $redisBloomPath /usr/local/lib/
+        rm -rf /var/lib/apt/lists/*; \
 # create working directories and organize files
 RUN \
         mkdir /data && chown keydb:keydb /data; \

src/IStorage.h

@@ -1,25 +1,10 @@
 #pragma once
 #include <functional>
-#include <unordered_set>
 #include "sds.h"
-#include "ae.h"
 #include <string>
 
 #define METADATA_DB_IDENTIFIER "c299fde0-6d42-4ec4-b939-34f680ffe39f"
 
-struct StorageToken {
-    enum class TokenType {
-        SingleRead,
-        SingleWrite,
-        Delete,
-        BatchWrite,
-    };
-    TokenType type;
-    std::unordered_set<struct client *> setc;
-    struct redisDbPersistentData *db;
-    virtual ~StorageToken() {}
-};
-
 class IStorageFactory
 {
 public:
@@ -51,12 +36,6 @@ public:
     virtual bool enumerate_hashslot(callback fn, unsigned int hashslot) const = 0;
     virtual size_t count() const = 0;
 
-    virtual StorageToken *begin_retrieve(struct aeEventLoop *, aePostFunctionTokenProc, sds *, size_t) {return nullptr;};
-    virtual void complete_retrieve(StorageToken * /*tok*/, callbackSingle /*fn*/) {};
-
-    virtual StorageToken* begin_endWriteBatch(struct aeEventLoop *, aePostFunctionTokenProc*) { return nullptr; } // NOP
-    virtual void complete_endWriteBatch(StorageToken * /*tok*/) {};
-
     virtual void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem) {
         beginWriteBatch();
         for (size_t ielem = 0; ielem < celem; ++ielem) {

src/Makefile

@@ -73,10 +73,10 @@ ifneq ($(strip $(SANITIZE)),)
 endif
 
 ifeq ($(ENABLE_FLASH),yes)
-	FINAL_LIBS+= ../deps/rocksdb/librocksdb.a
 	FINAL_LIBS+= -lz -lcrypto -lbz2 -lzstd -llz4 -lsnappy
 	CXXFLAGS+= -I../deps/rocksdb/include/ -DENABLE_ROCKSDB
 	STORAGE_OBJ+= storage/rocksdb.o storage/rocksdbfactory.o
+	FINAL_LIBS+= ../deps/rocksdb/librocksdb.a 
 	DEPENDENCY_TARGETS+= rocksdb
 endif
 

src/SnapshotPayloadParseState.cpp

@@ -136,33 +136,14 @@ 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) {
-        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();
-        }
+        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 {
         // else drop the data
         vecqueuedKeys.clear();
@@ -233,6 +214,11 @@ void SnapshotPayloadParseState::trimState() {
     
     if (stackParse.empty()) {
         flushQueuedKeys();
+        while (*insertsInFlight > 0) {
+            // TODO: ProcessEventsWhileBlocked
+            aeReleaseLock();
+            aeAcquireLock();
+        }
     }
 }
 

src/SnapshotPayloadParseState.h

@@ -63,5 +63,4 @@ public:
     void pushValue(const char *rgch, long long cch);
     void pushValue(long long value);
     bool shouldThrottle() const { return *insertsInFlight > (cserver.cthreads*4); }
-    bool hasIOInFlight() const { return *insertsInFlight > 0; }
 };

src/StorageCache.cpp

@@ -208,30 +208,6 @@ void StorageCache::retrieve(sds key, IStorage::callbackSingle fn) const
     m_spstorage->retrieve(key, sdslen(key), fn);
 }
 
-StorageToken *StorageCache::begin_retrieve(struct aeEventLoop *el, aePostFunctionTokenProc proc, sds *rgkey, size_t ckey) {
-    std::unique_lock<fastlock> ul(m_lock);
-    if (m_pdict != nullptr)
-    {
-        bool fAnyKey = false;
-        for (size_t ik = 0; ik < ckey; ++ik) {
-            uint64_t hash = dictSdsHash(rgkey[ik]);
-            dictEntry *de = dictFind(m_pdict, reinterpret_cast<void*>(hash));
-
-            if (de != nullptr)
-                fAnyKey = true;
-        }
-        if (!fAnyKey)
-            return nullptr; // All keys are missing - skip the io
-    }
-    ul.unlock();
-
-    return m_spstorage->begin_retrieve(el, proc, rgkey, ckey);
-}
-
-void StorageCache::complete_retrieve(StorageToken *tok, IStorage::callbackSingle fn) {
-    m_spstorage->complete_retrieve(tok, fn);
-}
-
 size_t StorageCache::count() const
 {
     std::unique_lock<fastlock> ul(m_lock, std::defer_lock);
@@ -257,5 +233,4 @@ void StorageCache::emergencyFreeCache() {
             dictRelease(d);
         });
     }
-}
-
+}

src/StorageCache.h

@@ -43,10 +43,6 @@ public:
     void insert(sds key, const void *data, size_t cbdata, bool fOverwrite);
     void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem);
     void retrieve(sds key, IStorage::callbackSingle fn) const;
-    StorageToken *begin_retrieve(struct aeEventLoop *el, aePostFunctionTokenProc proc, sds *rgkey, size_t ckey);
-    void complete_retrieve(StorageToken *tok, IStorage::callbackSingle fn);
-    StorageToken* begin_endWriteBatch(struct aeEventLoop *el, aePostFunctionTokenProc* proc) {return m_spstorage->begin_endWriteBatch(el,proc);} // NOP
-    void complete_endWriteBatch(StorageToken *tok) {m_spstorage->complete_endWriteBatch(tok);};
     bool erase(sds key);
     void emergencyFreeCache();
     bool keycacheIsEnabled() const { return m_pdict != nullptr; }

src/ae.cpp

@@ -278,7 +278,7 @@ int aePostFunction(aeEventLoop *eventLoop, aePostFunctionTokenProc *proc, Storag
     cmd.op = AE_ASYNC_OP::PostAsynDBFunction; 
     cmd.tproc = proc;
     cmd.clientData = (void*)token; 
-    cmd.fLock = false;
+    cmd.fLock = true;
     auto size = write(eventLoop->fdCmdWrite, &cmd, sizeof(cmd));
     if (size != sizeof(cmd))
         return AE_ERR;

src/blocked.cpp

@@ -92,8 +92,7 @@ void blockClient(client *c, int btype) {
     /* Master client should never be blocked unless pause or module */
     serverAssert(!(c->flags & CLIENT_MASTER &&
                    btype != BLOCKED_MODULE &&
-                   btype != BLOCKED_PAUSE &&
-                   btype != BLOCKED_STORAGE));
+                   btype != BLOCKED_PAUSE));
 
     c->flags |= CLIENT_BLOCKED;
     c->btype = btype;
@@ -208,8 +207,6 @@ void unblockClient(client *c) {
     } else if (c->btype == BLOCKED_PAUSE) {
         listDelNode(g_pserver->paused_clients,c->paused_list_node);
         c->paused_list_node = NULL;
-    } else if (c->btype == BLOCKED_STORAGE) {
-        serverTL->vecclientsProcess.push_back(c);
     } else {
         serverPanic("Unknown btype in unblockClient().");
     }

src/config.cpp

@@ -361,7 +361,7 @@ bool initializeStorageProvider(const char **err)
             // Create The Storage Factory (if necessary)
             serverLog(LL_NOTICE, "Initializing FLASH storage provider (this may take a long time)");
             adjustOpenFilesLimit();
-            g_pserver->m_pstorageFactory = CreateRocksDBStorageFactory(g_sdsArgs, cserver.dbnum, cserver.storage_conf, cserver.storage_conf ? strlen(cserver.storage_conf) : 0, &g_pserver->asyncreadworkqueue, &g_pserver->asyncwriteworkqueue);
+            g_pserver->m_pstorageFactory = CreateRocksDBStorageFactory(g_sdsArgs, cserver.dbnum, cserver.storage_conf, cserver.storage_conf ? strlen(cserver.storage_conf) : 0);
 #else
             serverLog(LL_WARNING, "To use the flash storage provider please compile KeyDB with ENABLE_FLASH=yes");
             serverLog(LL_WARNING, "Exiting due to the use of an unsupported storage provider");
@@ -474,90 +474,6 @@ void initConfigValues() {
     }
 }
 
-struct in_addr maskFromBitsIPV4(int bits) {
-    struct in_addr mask;
-    memset(&mask, 0, sizeof(mask));
-    mask.s_addr = htonl(ULONG_MAX << (IPV4_BITS - bits));
-    return mask;
-}
-
-bool validateAndAddIPV4(sds string) {
-    struct in_addr ip, mask;
-    int subnetParts;
-    sds* subnetSplit = sdssplitlen(string, sdslen(string), "/", 1, &subnetParts);
-    TCleanup splitCleanup([&](){
-        sdsfreesplitres(subnetSplit, subnetParts);
-    });
-    if (subnetParts > 2) {
-        return false;
-    } else if (subnetParts == 2) {
-        char* endptr;
-        long bits = strtol(subnetSplit[1], &endptr, 10);
-        if (*endptr != '\0' || bits < 0 || bits > IPV4_BITS) {
-            return false;
-        }
-        mask = maskFromBitsIPV4(bits);
-    } else {
-        mask = maskFromBitsIPV4(IPV4_BITS);
-    }
-    if (subnetParts > 0) {
-        if (inet_pton(AF_INET, subnetSplit[0], &ip) == 0) {
-            return false;
-        }
-        g_pserver->overload_ignorelist.emplace(ip, mask);
-        return true;
-    } else {
-        return false;
-    }
-}
-
-struct in6_addr maskFromBitsIPV6(int bits) {
-    struct in6_addr mask;
-    memset(&mask, 0, sizeof(mask));
-    for (unsigned int i = 0; i < sizeof(struct in6_addr); i++) {
-        if (bits >= CHAR_BIT) {
-            mask.s6_addr[i] = UCHAR_MAX;
-            bits -= CHAR_BIT;
-        } else if (bits > 0) {
-            mask.s6_addr[i] = UCHAR_MAX << (CHAR_BIT - bits);
-            bits = 0;
-        } else {
-            break;
-        }
-    }
-    return mask;
-}
-
-bool validateAndAddIPV6(sds string) {
-    struct in6_addr ip, mask;
-    int subnetParts;
-    sds* subnetSplit = sdssplitlen(string, sdslen(string), "/", 1, &subnetParts);
-    TCleanup splitCleanup([&](){
-        sdsfreesplitres(subnetSplit, subnetParts);
-    });
-    if (subnetParts > 2) {
-        return false;
-    } else if (subnetParts == 2) {
-        char* endptr;
-        long bits = strtol(subnetSplit[1], &endptr, 10);
-        if (*endptr != '\0' || bits < 0 || bits > IPV6_BITS) {
-            return false;
-        }
-        mask = maskFromBitsIPV6(bits);
-    } else {
-        mask = maskFromBitsIPV6(IPV6_BITS);
-    }
-    if (subnetParts > 0) {
-        if (inet_pton(AF_INET6, subnetSplit[0], &ip) == 0) {
-            return false;
-        }
-        g_pserver->overload_ignorelist_ipv6.emplace(ip, mask);
-        return true;
-    } else {
-        return false;
-    }
-}
-
 void loadServerConfigFromString(char *config) {
     const char *err = NULL;
     int linenum = 0, totlines, i;
@@ -863,27 +779,6 @@ void loadServerConfigFromString(char *config) {
             }
             for (int i = 1; i < argc; i++)
                 g_pserver->tls_auditlog_blocklist.emplace(argv[i], strlen(argv[i]));
-        } else if (!strcasecmp(argv[0], "tls-overload-ignorelist")) {
-            if (argc < 2) {
-                err = "must supply at least one element in the ignore list"; goto loaderr;
-            }
-            if (!g_pserver->tls_overload_ignorelist.empty()) {
-                err = "tls-overload-ignorelist may only be set once"; goto loaderr;
-            }
-            for (int i = 1; i < argc; i++)
-                g_pserver->tls_overload_ignorelist.emplace(argv[i], strlen(argv[i]));
-        } else if (!strcasecmp(argv[0], "overload-ignorelist")) {
-            if (argc < 2) {
-                err = "must supply at least one element in the ignore list"; goto loaderr;
-            }
-            if (!g_pserver->overload_ignorelist.empty() || !g_pserver->overload_ignorelist_ipv6.empty()) {
-                err = "overload-ignorelist may only be set once"; goto loaderr;
-            }
-            for (int i = 1; i < argc; i++) {
-                if (!validateAndAddIPV4(argv[i]) && !validateAndAddIPV6(argv[i])) {
-                    err = "overload-ignorelist must be a list of valid IP addresses or subnets"; goto loaderr;
-                }
-            }
         } else if (!strcasecmp(argv[0], "version-override") && argc == 2) {
             KEYDB_SET_VERSION = zstrdup(argv[1]);
             serverLog(LL_WARNING, "Warning version is overriden to: %s\n", KEYDB_SET_VERSION);
@@ -1013,8 +908,8 @@ void configSetCommand(client *c) {
 
     if (c->argc < 4 || c->argc > 4) {
         o = nullptr;
-        // Variadic set is only supported for tls-allowlist, tls-auditlog-blocklist, overload-ignorelist and tls-overload-ignorelist
-        if (strcasecmp(szFromObj(c->argv[2]), "tls-allowlist") && strcasecmp(szFromObj(c->argv[2]), "tls-auditlog-blocklist") && strcasecmp(szFromObj(c->argv[2]), "overload-ignorelist") && strcasecmp(szFromObj(c->argv[2]), "tls-overload-ignorelist") ) {
+        // Variadic set is only supported for tls-allowlist
+        if (strcasecmp(szFromObj(c->argv[2]), "tls-allowlist")) {
             addReplySubcommandSyntaxError(c);
             return;
         }
@@ -1189,28 +1084,6 @@ void configSetCommand(client *c) {
             robj *val = c->argv[i];
             g_pserver->tls_allowlist.emplace(szFromObj(val), sdslen(szFromObj(val)));
         }
-    } config_set_special_field("tls-auditlog-blocklist") {
-        g_pserver->tls_auditlog_blocklist.clear();
-        for (int i = 3; i < c->argc; ++i) {
-            robj *val = c->argv[i];
-            g_pserver->tls_auditlog_blocklist.emplace(szFromObj(val), sdslen(szFromObj(val)));
-        }
-    } config_set_special_field("tls-overload-ignorelist") {
-        g_pserver->tls_overload_ignorelist.clear();
-        for (int i = 3; i < c->argc; ++i) {
-            robj *val = c->argv[i];
-            g_pserver->tls_overload_ignorelist.emplace(szFromObj(val), sdslen(szFromObj(val)));
-        }
-    } config_set_special_field("overload-ignorelist") {
-        g_pserver->overload_ignorelist.clear();
-        g_pserver->overload_ignorelist_ipv6.clear();
-        for (int i = 3; i < c->argc; ++i) {
-            robj *val = c->argv[i];
-            if (!validateAndAddIPV4(szFromObj(val)) && !validateAndAddIPV6(szFromObj(val))) {
-                addReplyError(c, "overload-ignorelist must be a list of valid IP addresses or subnets");
-                return;
-            }
-        }
     /* Everything else is an error... */
     } config_set_else {
         addReplyErrorFormat(c,"Unsupported CONFIG parameter: %s",
@@ -1422,37 +1295,6 @@ void configGetCommand(client *c) {
         }
         matches++;
     }
-    if (stringmatch(pattern, "tls-auditlog-blocklist", 1)) {
-        addReplyBulkCString(c,"tls-auditlog-blocklist");
-        addReplyArrayLen(c, (long)g_pserver->tls_auditlog_blocklist.size());
-        for (auto &elem : g_pserver->tls_auditlog_blocklist) {
-            addReplyBulkCBuffer(c, elem.get(), elem.size()); // addReplyBulkSds will free which we absolutely don't want
-        }
-        matches++;
-    }
-    if (stringmatch(pattern, "tls-overload-ignorelist", 1)) {
-        addReplyBulkCString(c,"tls-overload-ignorelist");
-        addReplyArrayLen(c, (long)g_pserver->tls_overload_ignorelist.size());
-        for (auto &elem : g_pserver->tls_overload_ignorelist) {
-            addReplyBulkCBuffer(c, elem.get(), elem.size()); // addReplyBulkSds will free which we absolutely don't want
-        }
-        matches++;
-    }
-    if (stringmatch(pattern, "overload-ignorelist", 1)) {
-        addReplyBulkCString(c,"overload-ignorelist");
-        addReplyArrayLen(c, (long)g_pserver->overload_ignorelist.size() + (long)g_pserver->overload_ignorelist_ipv6.size());
-        for (auto &elem : g_pserver->overload_ignorelist) {
-            sds elem_sds = elem.getString();
-            addReplyBulkCBuffer(c, elem_sds, sdslen(elem_sds));
-            sdsfree(elem_sds);
-        }
-        for (auto &elem : g_pserver->overload_ignorelist_ipv6) {
-            sds elem_sds = elem.getString();
-            addReplyBulkCBuffer(c, elem_sds, sdslen(elem_sds));
-            sdsfree(elem_sds);
-        }
-        matches++;
-    }
 
     setDeferredMapLen(c,replylen,matches);
 }
@@ -2136,56 +1978,6 @@ int rewriteConfig(char *path, int force_all) {
         rewriteConfigMarkAsProcessed(state, "tls-allowlist"); // ensure the line is removed if it existed
     }
 
-    if (!g_pserver->tls_overload_ignorelist.empty()) {
-        sds conf = sdsnew("tls-overload-ignorelist ");
-        for (auto &elem : g_pserver->tls_overload_ignorelist) {
-            conf = sdscatsds(conf, (sds)elem.get());
-            conf = sdscat(conf, " ");
-        }
-        // trim the trailing space
-        sdsrange(conf, 0, -1);
-        rewriteConfigRewriteLine(state,"tls-overload-ignorelist",conf,1 /*force*/);
-        // note: conf is owned by rewriteConfigRewriteLine - no need to free
-    } else {
-        rewriteConfigMarkAsProcessed(state, "tls-overload-ignorelist"); // ensure the line is removed if it existed
-    }
-
-    if (!g_pserver->tls_auditlog_blocklist.empty()) {
-        sds conf = sdsnew("tls-auditlog-blocklist ");
-        for (auto &elem : g_pserver->tls_auditlog_blocklist) {
-            conf = sdscatsds(conf, (sds)elem.get());
-            conf = sdscat(conf, " ");
-        }
-        // trim the trailing space
-        sdsrange(conf, 0, -1);
-        rewriteConfigRewriteLine(state,"tls-auditlog-blocklist",conf,1 /*force*/);
-        // note: conf is owned by rewriteConfigRewriteLine - no need to free
-    } else {
-        rewriteConfigMarkAsProcessed(state, "tls-auditlog-blocklist"); // ensure the line is removed if it existed
-    }
-
-    if (!g_pserver->overload_ignorelist.empty() || !g_pserver->overload_ignorelist_ipv6.empty()) {
-        sds conf = sdsnew("overload-ignorelist ");
-        for (auto &elem : g_pserver->overload_ignorelist) {
-            sds elem_sds = elem.getString();
-            conf = sdscatsds(conf, elem_sds);
-            sdsfree(elem_sds);
-            conf = sdscat(conf, " ");
-        }
-        for (auto &elem : g_pserver->overload_ignorelist_ipv6) {
-            sds elem_sds = elem.getString();
-            conf = sdscatsds(conf, elem_sds);
-            sdsfree(elem_sds);
-            conf = sdscat(conf, " ");
-        }
-        // trim the trailing space
-        sdsrange(conf, 0, -1);
-        rewriteConfigRewriteLine(state,"overload-ignorelist",conf,1 /*force*/);
-        // note: conf is owned by rewriteConfigRewriteLine - no need to free
-    } else {
-        rewriteConfigMarkAsProcessed(state, "overload-ignorelist"); // ensure the line is removed if it existed
-    }
-
     /* Rewrite Sentinel config if in Sentinel mode. */
     if (g_pserver->sentinel_mode) rewriteConfigSentinelOption(state);
 
@@ -2724,14 +2516,6 @@ static int isValidProcTitleTemplate(char *val, const char **err) {
     return 1;
 }
 
-static int isValidDisklessEnum(int e, const char **err) {
-    if (e == REPL_DISKLESS_LOAD_SWAPDB && g_pserver->m_pstorageFactory != nullptr) {
-        *err = "cannot use swapdb with a storage provider";
-        return 0;
-    }
-    return 1;
-}
-
 static int updateProcTitleTemplate(char *val, char *prev, const char **err) {
     UNUSED(val);
     UNUSED(prev);
@@ -3099,7 +2883,7 @@ standardConfig configs[] = {
     /* Enum Configs */
     createEnumConfig("supervised", NULL, IMMUTABLE_CONFIG, supervised_mode_enum, cserver.supervised_mode, SUPERVISED_NONE, NULL, NULL),
     createEnumConfig("syslog-facility", NULL, IMMUTABLE_CONFIG, syslog_facility_enum, g_pserver->syslog_facility, LOG_LOCAL0, NULL, NULL),
-    createEnumConfig("repl-diskless-load", NULL, MODIFIABLE_CONFIG, repl_diskless_load_enum, g_pserver->repl_diskless_load, REPL_DISKLESS_LOAD_DISABLED, isValidDisklessEnum, NULL),
+    createEnumConfig("repl-diskless-load", NULL, MODIFIABLE_CONFIG, repl_diskless_load_enum, g_pserver->repl_diskless_load, REPL_DISKLESS_LOAD_DISABLED, NULL, NULL),
     createEnumConfig("loglevel", NULL, MODIFIABLE_CONFIG, loglevel_enum, cserver.verbosity, LL_NOTICE, NULL, NULL),
     createEnumConfig("maxmemory-policy", NULL, MODIFIABLE_CONFIG, maxmemory_policy_enum, g_pserver->maxmemory_policy, MAXMEMORY_NO_EVICTION, NULL, NULL),
     createEnumConfig("appendfsync", NULL, MODIFIABLE_CONFIG, aof_fsync_enum, g_pserver->aof_fsync, AOF_FSYNC_EVERYSEC, NULL, NULL),
@@ -3166,7 +2950,6 @@ standardConfig configs[] = {
     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),
-    createLongLongConfig("repl-backlog-max-writes-per-event", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, g_pserver->repl_backlog_max_writes_per_event, 8ll*1024*1024, MEMORY_CONFIG, NULL, NULL),
 
     /* Unsigned Long Long configs */
     createULongLongConfig("maxmemory", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, g_pserver->maxmemory, 0, MEMORY_CONFIG, NULL, updateMaxmemory),
@@ -3198,10 +2981,8 @@ standardConfig configs[] = {
     createSizeTConfig("semi-ordered-set-bucket-size", NULL, MODIFIABLE_CONFIG, 0, 1024, g_semiOrderedSetTargetBucketSize, 0, INTEGER_CONFIG, NULL, NULL),
     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("overload-protect-tenacity", NULL, MODIFIABLE_CONFIG, 0, 100, g_pserver->overload_protect_tenacity, 10, 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, 1, NULL, NULL),
 
 #ifdef USE_OPENSSL
     createIntConfig("tls-port", NULL, MODIFIABLE_CONFIG, 0, 65535, g_pserver->tls_port, 0, INTEGER_CONFIG, NULL, updateTLSPort), /* TCP port. */

src/connection.h

@@ -52,7 +52,6 @@ typedef enum {
 #define CONN_FLAG_READ_THREADSAFE        (1<<2)
 #define CONN_FLAG_WRITE_THREADSAFE       (1<<3)
 #define CONN_FLAG_AUDIT_LOGGING_REQUIRED (1<<4)
-#define CONN_FLAG_IGNORE_OVERLOAD (1<<5)
 
 #define CONN_TYPE_SOCKET            1
 #define CONN_TYPE_TLS               2

src/db.cpp

@@ -56,6 +56,7 @@ 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 */
@@ -2528,12 +2529,6 @@ void slotToKeyFlush(int async) {
  * decrement the reference count to release the keys names. */
 unsigned int getKeysInSlot(unsigned int hashslot, robj **keys, unsigned int count) {
     if (g_pserver->m_pstorageFactory != nullptr) {
-        // We must commit so the storage engine agrees on the number of items in the hash slot
-        if (g_pserver->db[0]->FTrackingChanges()) {
-            if (g_pserver->db[0]->processChanges(false))
-                g_pserver->db[0]->commitChanges();
-            g_pserver->db[0]->trackChanges(false);
-        }
         int j = 0;
         g_pserver->db[0]->getStorageCache()->enumerate_hashslot([&](const char *key, size_t cchKey, const void *, size_t )->bool {
             keys[j++] = createStringObject(key, cchKey);
@@ -2563,12 +2558,6 @@ unsigned int getKeysInSlot(unsigned int hashslot, robj **keys, unsigned int coun
 unsigned int delKeysInSlot(unsigned int hashslot) {
     serverAssert(GlobalLocksAcquired());
     if (g_pserver->m_pstorageFactory != nullptr) {
-        // We must commit so the storage engine agrees on the number of items in the hash slot
-        if (g_pserver->db[0]->FTrackingChanges()) {
-            if (g_pserver->db[0]->processChanges(false))
-                g_pserver->db[0]->commitChanges();
-            g_pserver->db[0]->trackChanges(false);
-        }
         int j = 0;
         g_pserver->db[0]->getStorageCache()->enumerate_hashslot([&](const char *key, size_t cchKey, const void *, size_t )->bool {
             robj *keyobj = createStringObject(key, cchKey);
@@ -2896,7 +2885,6 @@ LNotFound:
                 o->SetFExpires(spexpire != nullptr);
                 if (spexpire != nullptr) {
                     o->expire = std::move(*spexpire);
-                    ++m_numexpires;
                 }
                 g_pserver->stat_storage_provider_read_hits++;
             } else {
@@ -3001,7 +2989,6 @@ void redisDbPersistentData::processChangesAsync(std::atomic<int> &pendingJobs)
     dict *dictNew = dictCreate(&dbDictType, nullptr);
     std::swap(dictNew, m_pdict);
     m_cnewKeysPending = 0;
-    m_numexpires = 0;
     g_pserver->asyncworkqueue->AddWorkFunction([dictNew, this, &pendingJobs]{
         dictIterator *di = dictGetIterator(dictNew);
         dictEntry *de;
@@ -3057,18 +3044,7 @@ void redisDbPersistentData::commitChanges(const redisDbPersistentDataSnapshot **
         m_pdbSnapshotStorageFlush = nullptr;
     }
     if (m_spstorage != nullptr)
-    {
-#if 1
         m_spstorage->endWriteBatch();
-#else
-        auto *tok = m_spstorage->begin_endWriteBatch(serverTL->el, storageLoadCallback);
-        if (tok != nullptr)
-        {
-            tok->db = this;
-            tok->type = StorageToken::TokenType::BatchWrite;
-        }
-#endif
-    }
 }
 
 redisDbPersistentData::~redisDbPersistentData()
@@ -3140,14 +3116,7 @@ bool redisDbPersistentData::removeCachedValue(const char *key, dictEntry **ppde)
     // since we write ASAP the database already has a valid copy so safe to delete
     if (ppde != nullptr) {
         *ppde = dictUnlink(m_pdict, key);
-        robj *o = (robj*)dictGetVal(*ppde);
-        if (o->FExpires())
-            --m_numexpires;
     } else {
-        dictEntry *deT = dictFind(m_pdict, key);
-        robj *o = (robj*)dictGetVal(deT);
-        if (o->FExpires())
-            --m_numexpires;
         dictDelete(m_pdict, key);
     }
 
@@ -3167,11 +3136,8 @@ void redisDbPersistentData::trackChanges(bool fBulk, size_t sizeHint)
     if (fBulk)
         m_fAllChanged.fetch_add(1, std::memory_order_acq_rel);
 
-    if (sizeHint > 0) {
-        aeAcquireLock();
+    if (sizeHint > 0)
         dictExpand(m_dictChanged, sizeHint, false);
-        aeReleaseLock();
-    }
 }
 
 void redisDbPersistentData::removeAllCachedValues()
@@ -3194,7 +3160,6 @@ void redisDbPersistentData::removeAllCachedValues()
     } else {
         dictEmpty(m_pdict, nullptr);
     }
-    m_numexpires = 0;
 }
 
 void redisDbPersistentData::disableKeyCache()
@@ -3346,124 +3311,90 @@ void redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
                 }
             }
         }
-#else
-        UNUSED(c);
-        UNUSED(command);
 #endif
         return;
     }
-}
 
-void redisDbPersistentData::prefetchKeysFlash(const std::unordered_set<client*> &setc)
-{
-    serverAssert(GlobalLocksAcquired());
-    std::vector<sds> veckeys;
-    std::unordered_set<client*> setcBlocked;
+    AeLocker lock;
 
-    for (client *c : setc) {
-        for (auto &command : c->vecqueuedcmd) {
-            getKeysResult result = GETKEYS_RESULT_INIT;
-            if (command.argc == 0)  // parse can do this it will be handled by processClient
-                break;
-            auto cmd = lookupCommand(szFromObj(command.argv[0]));
-            if (cmd == nullptr)
-                break; // Bad command? It's not for us to judge, just bail
-            
-            if (command.argc < std::abs(cmd->arity))
-                break; // Invalid number of args
-            
-            int numkeys = getKeysFromCommand(cmd, command.argv, command.argc, &result);
-            bool fQueued = false;
-            for (int ikey = 0; ikey < numkeys; ++ikey)
+    std::vector<robj*> veckeys;
+    lock.arm(c);
+    getKeysResult result = GETKEYS_RESULT_INIT;
+    auto cmd = lookupCommand(szFromObj(command.argv[0]));
+    if (cmd == nullptr)
+        return; // Bad command? It's not for us to judge, just bail
+    
+    if (command.argc < std::abs(cmd->arity))
+        return; // Invalid number of args
+    
+    int numkeys = getKeysFromCommand(cmd, command.argv, command.argc, &result);
+    for (int ikey = 0; ikey < numkeys; ++ikey)
+    {
+        robj *objKey = command.argv[result.keys[ikey]];
+        if (this->find_cached_threadsafe(szFromObj(objKey)) == nullptr)
+            veckeys.push_back(objKey);
+    }
+    lock.disarm();
+
+    getKeysFreeResult(&result);
+
+    std::vector<std::tuple<sds, robj*, std::unique_ptr<expireEntry>>> vecInserts;
+    for (robj *objKey : veckeys)
+    {
+        sds sharedKey = sdsdupshared((sds)szFromObj(objKey));
+        std::unique_ptr<expireEntry> spexpire;
+        robj *o = nullptr;
+        m_spstorage->retrieve((sds)szFromObj(objKey), [&](const char *, size_t, const void *data, size_t cb){
+                size_t offset = 0;
+                spexpire = deserializeExpire((const char*)data, cb, &offset);    
+                o = deserializeStoredObject(reinterpret_cast<const char*>(data) + offset, cb - offset);
+                serverAssert(o != nullptr);
+        });
+
+        if (o != nullptr) {
+            vecInserts.emplace_back(sharedKey, o, std::move(spexpire));
+        } else if (sharedKey != nullptr) {
+            sdsfree(sharedKey);
+        }
+    }
+
+    if (!vecInserts.empty()) {
+        lock.arm(c);
+        for (auto &tuple : vecInserts)
+        {
+            sds sharedKey = std::get<0>(tuple);
+            robj *o = std::get<1>(tuple);
+            std::unique_ptr<expireEntry> spexpire = std::move(std::get<2>(tuple));
+
+            if (o != nullptr)
             {
-                robj *objKey = command.argv[result.keys[ikey]];
-                if (this->find_cached_threadsafe(szFromObj(objKey)) == nullptr) {
-                    veckeys.push_back(szFromObj(objKey));
-                    fQueued = true;
+                if (this->find_cached_threadsafe(sharedKey) != nullptr)
+                {
+                    // While unlocked this was already ensured
+                    decrRefCount(o);
+                    sdsfree(sharedKey);
+                }
+                else
+                {
+                    if (spexpire != nullptr) {
+                        if (spexpire->when() < mstime()) {
+                            break;
+                        }
+                    }
+                    dictAdd(m_pdict, sharedKey, o);
+                    if (spexpire != nullptr)
+                        o->expire = std::move(*spexpire);
+                    o->SetFExpires(spexpire != nullptr);
                 }
             }
-
-            if (fQueued) {
-                setcBlocked.insert(c);
-            }
-            getKeysFreeResult(&result);
-        }
-    }
-
-    if (veckeys.empty())
-        return;
-
-    StorageToken *tok = m_spstorage->begin_retrieve(serverTL->el, storageLoadCallback, veckeys.data(), veckeys.size());
-    if (tok != nullptr) {
-        for (client *c : setcBlocked) {
-            if (!(c->flags & CLIENT_BLOCKED)) {
-                blockClient(c, BLOCKED_STORAGE);
+            else
+            {
+                if (sharedKey != nullptr)
+                    sdsfree(sharedKey); // BUG but don't bother crashing
             }
         }
-        tok->setc = std::move(setcBlocked);
-        tok->type = StorageToken::TokenType::SingleRead;
-        tok->db = this;
+        lock.disarm();
     }
+
     return;
 }
-
-/*static*/ void redisDbPersistentData::storageLoadCallback(aeEventLoop *, StorageToken *tok) {
-    serverTL->setStorageTokensProcess.insert(tok);
-}
-
-void redisDbPersistentData::processStorageToken(StorageToken *tok) {
-    auto setc = std::move(tok->setc);
-    switch (tok->type)
-    {
-
-    case StorageToken::TokenType::SingleRead:
-    {
-        tok->db->m_spstorage->complete_retrieve(tok, [&](const char *szKey, size_t cbKey, const void *data, size_t cb) {
-            auto *db = tok->db;
-            size_t offset = 0;
-            sds key = sdsnewlen(szKey, -((ssize_t)cbKey));
-            auto spexpire = deserializeExpire((const char*)data, cb, &offset);    
-            robj *o = deserializeStoredObject(reinterpret_cast<const char*>(data) + offset, cb - offset);
-            serverAssert(o != nullptr);
-
-            if (db->find_cached_threadsafe(key) != nullptr) {
-            LUnneeded:
-                // While unlocked this was already ensured
-                decrRefCount(o);
-                sdsfree(key);
-            } else {
-                if (spexpire != nullptr) {
-                    if (spexpire->when() < mstime()) {
-                        goto LUnneeded;
-                    }
-                }
-                dictAdd(db->m_pdict, key, o);
-                o->SetFExpires(spexpire != nullptr);
-                if (spexpire != nullptr) {
-                    o->expire = std::move(*spexpire);
-                }
-            } 
-        });
-        break;
-    }
-    case StorageToken::TokenType::BatchWrite:
-    {
-        tok->db->m_spstorage->complete_endWriteBatch(tok);
-        break;
-    }
-    default:
-        serverAssert((tok->type == StorageToken::TokenType::SingleRead) || (tok->type == StorageToken::TokenType::BatchWrite));
-        break;
-    } //switch end
-
-    tok = nullptr; // Invalid past this point
-
-    for (client *c : setc)
-    {
-        std::unique_lock<fastlock> ul(c->lock);
-        if (c->flags & CLIENT_BLOCKED)
-            unblockClient(c);
-        
-        serverTL->vecclientsProcess.push_back(c);
-    }
-}

src/meminfo.cpp

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

src/modules/keydb_modstatsd/modmain.cpp

@@ -67,6 +67,8 @@ public:
 
 /* constants */
 static time_t c_infoUpdateSeconds = 10;
+// the current Redis Cluster setup we configure replication factor as 2, each non-empty master node should have 2 replicas, given that there are 3 zones in each regions
+static const int EXPECTED_NUMBER_OF_REPLICAS = 2;
 
 StatsdClientWrapper *g_stats = nullptr;
 std::string m_strPrefix { "keydb" };
@@ -92,7 +94,6 @@ struct StatsRecord {
 
     /* Dynamic Values */
     long long prevVal = 0;
-    long long currVal = 0;
 };
 
 std::unordered_map<std::string, StatsRecord> g_mapInfoFields = {
@@ -100,8 +101,6 @@ std::unordered_map<std::string, StatsRecord> g_mapInfoFields = {
     { "used_memory", { StatsD_Type::STATSD_GAUGE_BYTES, false /* prefixOnly */}},
     { "used_memory_rss", { StatsD_Type::STATSD_GAUGE_BYTES }},
     { "maxmemory", { StatsD_Type::STATSD_GAUGE_BYTES, false /* prefixOnly */}},
-    { "maxstorage", { StatsD_Type::STATSD_GAUGE_BYTES, false /* prefixOnly */}},
-    { "storage_used", { StatsD_Type::STATSD_GAUGE_BYTES, false /* prefixOnly */}},
     { "used_memory_dataset_perc", { StatsD_Type::STATSD_GAUGE_FLOAT }},
     { "avg_lock_contention", { StatsD_Type::STATSD_GAUGE_LONGLONG }},
     { "repl_backlog_size", { StatsD_Type::STATSD_GAUGE_BYTES }},
@@ -254,7 +253,6 @@ std::unordered_map<std::string, StatsRecord> g_mapInfoFields = {
     { "cluster_size", { StatsD_Type::STATSD_GAUGE_LONGLONG }},
     { "storage_flash_available_bytes", { StatsD_Type::STATSD_GAUGE_BYTES }},
     { "storage_flash_total_bytes", { StatsD_Type::STATSD_GAUGE_BYTES }},
-    { "last_overload_cpu_reading", { StatsD_Type::STATSD_GAUGE_FLOAT }},
 };
 
 /* Globals */
@@ -299,8 +297,6 @@ void handleStatItem(struct RedisModuleCtx *ctx, std::string name, StatsRecord &r
     switch (record.type) {
         case StatsD_Type::STATSD_GAUGE_LONGLONG: {
             long long val = strtoll(pchValue, nullptr, 10);
-            record.prevVal = record.currVal;
-            record.currVal = val;
             g_stats->gauge(name, val, record.prefixOnly);
             RedisModule_Log(ctx, REDISMODULE_LOGLEVEL_DEBUG, "Emitting metric \"%s\": %lld", name.c_str(), val);
             break;
@@ -322,10 +318,9 @@ void handleStatItem(struct RedisModuleCtx *ctx, std::string name, StatsRecord &r
 
         case StatsD_Type::STATSD_DELTA: {
             long long val = strtoll(pchValue, nullptr, 10);
-            record.prevVal = record.currVal;
-            record.currVal = val;
             g_stats->count(name, val - record.prevVal, record.prefixOnly);
             RedisModule_Log(ctx, REDISMODULE_LOGLEVEL_DEBUG, "Emitting metric count for \"%s\": %lld", name.c_str() , val - record.prevVal);
+            record.prevVal = val;
             break;
         }
 
@@ -436,20 +431,6 @@ void handle_info_response(struct RedisModuleCtx *ctx, const char *szReply, size_
     }
 
     #undef SAFETY_CHECK_POINTER
-
-    // Emit keyspace hit rate metric which is computed from info values
-    auto hit = g_mapInfoFields.find("keyspace_hits");
-    auto miss = g_mapInfoFields.find("keyspace_misses");
-    if (hit != g_mapInfoFields.end() && miss != g_mapInfoFields.end()) {
-        long long hitValDelta = hit->second.currVal - hit->second.prevVal;
-        long long missValDelta = miss->second.currVal - miss->second.prevVal;
-        long long total = hitValDelta + missValDelta;
-        if (total > 0) {
-            double hitRate = (double)hitValDelta / total * 1000000;
-            g_stats->gauge("keyspace_hit_rate_per_million", hitRate, false /* prefixOnly */);
-            RedisModule_Log(ctx, REDISMODULE_LOGLEVEL_DEBUG, "Emitting metric \"keyspace_hit_rate_per_million\": %f", hitRate);
-        }
-    }
 }
 
 void handle_cluster_nodes_response(struct RedisModuleCtx *ctx, const char *szReply, size_t len) {
@@ -457,8 +438,6 @@ void handle_cluster_nodes_response(struct RedisModuleCtx *ctx, const char *szRep
     const char *pchLineStart = szReply;
     long long primaries = 0;
     long long replicas = 0;
-    long long handshakes = 0;
-    long long errors = 0;
     while (SAFETY_CHECK_POINTER(pchLineStart) && *pchLineStart != '\0') {
         // Loop Each Line
         const char *pchLineEnd = pchLineStart;
@@ -471,10 +450,6 @@ void handle_cluster_nodes_response(struct RedisModuleCtx *ctx, const char *szRep
             ++primaries;
         } else if (std::string::npos != line.find("slave")) {
             ++replicas;
-        } else if (std::string::npos != line.find("handshake")) {
-            ++handshakes;
-        } else if (std::string::npos != line.find("fail?") || std::string::npos != line.find("fail") || std::string::npos != line.find("noaddr") || std::string::npos != line.find("nofailover") || std::string::npos != line.find("noflags") ) {
-            ++errors;
         } else {
             RedisModule_Log(ctx, REDISMODULE_LOGLEVEL_WARNING, "Unexpected NODE format returned by \"CLUSTER NODES\" command: \"%s\"", line.c_str());
         }
@@ -506,10 +481,6 @@ void handle_cluster_nodes_response(struct RedisModuleCtx *ctx, const char *szRep
     RedisModule_Log(ctx, REDISMODULE_LOGLEVEL_DEBUG, "Emitting metric \"primaries\": %llu", primaries);
     g_stats->gauge("replicas", replicas);
     RedisModule_Log(ctx, REDISMODULE_LOGLEVEL_DEBUG, "Emitting metric \"replicas\": %llu", replicas);
-    g_stats->gauge("handshakes", handshakes);
-    RedisModule_Log(ctx, REDISMODULE_LOGLEVEL_DEBUG, "Emitting metric \"handshakes\": %llu", handshakes);
-    g_stats->gauge("cluster_nodes_error_nodes", errors);
-    RedisModule_Log(ctx, REDISMODULE_LOGLEVEL_DEBUG, "Emitting metric \"cluster_nodes_error_nodes\": %llu", errors);
     #undef SAFETY_CHECK_POINTER
 }
 
@@ -595,8 +566,10 @@ void emit_metrics_for_insufficient_replicas(struct RedisModuleCtx *ctx, long lon
     // check if the current node is a primary
     if (strncmp(role, "master", len) == 0) {
         RedisModuleCallReply *replicasReply = RedisModule_CallReplyArrayElement(reply, 2);
-        size_t numberOfActiveReplicas = RedisModule_CallReplyLength(replicasReply);
-        g_stats->gauge("numberOfActiveReplicas", numberOfActiveReplicas);
+        // check if there are less than 2 connected replicas
+        if (RedisModule_CallReplyLength(replicasReply) < EXPECTED_NUMBER_OF_REPLICAS) {
+            g_stats->increment("lessThanExpectedReplicas_error", 1);
+        }
     }
     RedisModule_FreeCallReply(reply);
 }

src/networking.cpp

@@ -38,7 +38,6 @@
 #include <vector>
 #include <mutex>
 #include "aelocker.h"
-#include <sys/socket.h>
 
 static void setProtocolError(const char *errstr, client *c);
 __thread int ProcessingEventsWhileBlocked = 0; /* See processEventsWhileBlocked(). */
@@ -164,6 +163,7 @@ 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;
@@ -1174,32 +1174,6 @@ int chooseBestThreadForAccept()
     return ielMinLoad;
 }
 
-bool checkOverloadIgnorelist(connection *conn) {
-    if (conn->flags & CONN_FLAG_IGNORE_OVERLOAD) {
-        return true;
-    }
-    struct sockaddr_storage sa;
-    socklen_t salen = sizeof(sa);
-    if (getpeername(conn->fd, (struct sockaddr *)&sa, &salen) == -1) {
-        return false;
-    }
-    if (sa.ss_family == AF_INET) {
-        struct sockaddr_in *s = (struct sockaddr_in *)&sa;
-        for (auto ignore : g_pserver->overload_ignorelist) {
-            if (ignore.match(s->sin_addr))
-                return true;
-        }
-    }
-    if (sa.ss_family == AF_INET6) {
-        struct sockaddr_in6 *s = (struct sockaddr_in6 *)&sa;
-        for (auto ignore : g_pserver->overload_ignorelist_ipv6) {
-            if (ignore.match(s->sin6_addr))
-                return true;
-        }
-    }
-    return false;
-}
-
 void clientAcceptHandler(connection *conn) {
     client *c = (client*)connGetPrivateData(conn);
 
@@ -1231,6 +1205,7 @@ void clientAcceptHandler(connection *conn) {
     {
         char cip[NET_IP_STR_LEN+1] = { 0 };
         connPeerToString(conn, cip, sizeof(cip)-1, NULL);
+
         if (strcmp(cip,"127.0.0.1") && strcmp(cip,"::1")) {
             const char *err =
                 "-DENIED KeyDB is running in protected mode because protected "
@@ -1262,14 +1237,6 @@ void clientAcceptHandler(connection *conn) {
         }
     }
 
-    if (checkOverloadIgnorelist(conn))
-    {
-        c->flags |= CLIENT_IGNORE_OVERLOAD;
-        char cip[NET_IP_STR_LEN+1] = { 0 };
-        connPeerToString(conn, cip, sizeof(cip)-1, NULL);
-        serverLog(LL_NOTICE, "Ignoring client from %s for loadshedding", cip);
-    }
-
     g_pserver->stat_numconnections++;
     moduleFireServerEvent(REDISMODULE_EVENT_CLIENT_CHANGE,
                           REDISMODULE_SUBEVENT_CLIENT_CHANGE_CONNECTED,
@@ -1627,7 +1594,6 @@ void unlinkClient(client *c) {
     }
 
     serverTL->vecclientsProcess.erase(std::remove(serverTL->vecclientsProcess.begin(), serverTL->vecclientsProcess.end(), c), serverTL->vecclientsProcess.end());
-    serverTL->setclientsPrefetch.erase(c);
 
     /* Clear the tracking status. */
     if (c->flags & CLIENT_TRACKING) disableTracking(c);
@@ -1641,7 +1607,7 @@ bool freeClient(client *c) {
 
     /* If a client is protected, yet we need to free it right now, make sure
      * to at least use asynchronous freeing. */
-    if (c->flags & CLIENT_PROTECTED || c->casyncOpsPending || c->replstate == SLAVE_STATE_FASTSYNC_TX || c->btype == BLOCKED_STORAGE) {
+    if (c->flags & CLIENT_PROTECTED || c->casyncOpsPending || c->replstate == SLAVE_STATE_FASTSYNC_TX) {
         freeClientAsync(c);
         return false;
     }
@@ -1872,23 +1838,21 @@ int writeToClient(client *c, int handler_installed) {
 
         while (clientHasPendingReplies(c)) {
             long long repl_end_idx = getReplIndexFromOffset(c->repl_end_off);
-            long long repl_max_writes = g_pserver->repl_backlog_max_writes_per_event > 0 ? g_pserver->repl_backlog_max_writes_per_event : LLONG_MAX;
-
             serverAssert(c->repl_curr_off != -1);
 
-            if (c->repl_curr_off != c->repl_end_off) {
+            if (c->repl_curr_off != c->repl_end_off){
                 long long repl_curr_idx = getReplIndexFromOffset(c->repl_curr_off); 
                 long long nwritten2ndStage = 0; /* How much was written from the start of the replication backlog
                                                 * in the event of a wrap around write */
                 /* normal case with no wrap around */
-                if (repl_end_idx >= repl_curr_idx) {
-                    nwritten = connWrite(c->conn, g_pserver->repl_backlog + repl_curr_idx, std::min(repl_max_writes, repl_end_idx - repl_curr_idx));
+                if (repl_end_idx >= repl_curr_idx){
+                    nwritten = connWrite(c->conn, g_pserver->repl_backlog + repl_curr_idx, repl_end_idx - repl_curr_idx);
                 /* wrap around case */
                 } else {
-                    nwritten = connWrite(c->conn, g_pserver->repl_backlog + repl_curr_idx, std::min(repl_max_writes, g_pserver->repl_backlog_size - repl_curr_idx));
+                    nwritten = connWrite(c->conn, g_pserver->repl_backlog + repl_curr_idx, g_pserver->repl_backlog_size - repl_curr_idx);
                     /* only attempt wrapping if we write the correct number of bytes */
                     if (nwritten == g_pserver->repl_backlog_size - repl_curr_idx){
-                        nwritten2ndStage = connWrite(c->conn, g_pserver->repl_backlog, std::min(repl_max_writes - nwritten, repl_end_idx));
+                        nwritten2ndStage = connWrite(c->conn, g_pserver->repl_backlog, repl_end_idx);
                         if (nwritten2ndStage != -1)
                             nwritten += nwritten2ndStage;
                     }                
@@ -1905,8 +1869,6 @@ int writeToClient(client *c, int handler_installed) {
                 if (nwritten2ndStage == -1) nwritten = -1;
                 if (nwritten == -1)
                     break;
-
-                if (totwritten > g_pserver->repl_backlog_max_writes_per_event) break;
             } else {
                 break;
             }
@@ -1977,7 +1939,7 @@ int writeToClient(client *c, int handler_installed) {
     if (nwritten == -1) {
         if (connGetState(c->conn) != CONN_STATE_CONNECTED) {
             serverLog(LL_VERBOSE,
-                "Error writing to client: %s (totwritten: %zd)", connGetLastError(c->conn), totwritten);
+                "Error writing to client: %s", connGetLastError(c->conn));
             freeClientAsync(c);
             
             return C_ERR;
@@ -2798,9 +2760,9 @@ void readQueryFromClient(connection *conn) {
         return;
     }
 
-    if (cserver.cthreads > 1 || g_pserver->m_pstorageFactory || g_pserver->is_overloaded) {
+    if (cserver.cthreads > 1 || g_pserver->m_pstorageFactory) {
         parseClientCommandBuffer(c);
-        if (g_pserver->enable_async_commands && !serverTL->disable_async_commands && listLength(g_pserver->monitors) == 0 && (aeLockContention() || serverTL->rgdbSnapshot[c->db->id] || g_fTestMode) && !serverTL->in_eval && !serverTL->in_exec && !g_pserver->is_overloaded) {
+        if (g_pserver->enable_async_commands && !serverTL->disable_async_commands && listLength(g_pserver->monitors) == 0 && (aeLockContention() || serverTL->rgdbSnapshot[c->db->id] || g_fTestMode) && !serverTL->in_eval && !serverTL->in_exec) {
             // Frequent writers aren't good candidates for this optimization, they cause us to renew the snapshot too often
             //  so we exclude them unless the snapshot we need already exists.
             // Note: In test mode we want to create snapshots as often as possibl to excercise them - we don't care about perf
@@ -2812,20 +2774,8 @@ void readQueryFromClient(connection *conn) {
                 processInputBuffer(c, false, CMD_CALL_SLOWLOG | CMD_CALL_STATS | CMD_CALL_ASYNC);
             }
         }
-        if (!c->vecqueuedcmd.empty()) {
-            if (g_pserver->m_pstorageFactory != nullptr && g_pserver->prefetch_enabled && !(c->flags & CLIENT_BLOCKED)) {
-                serverTL->setclientsPrefetch.insert(c);
-            } else {
-                if (g_pserver->is_overloaded && !(c->flags & (CLIENT_MASTER | CLIENT_SLAVE | CLIENT_PENDING_WRITE | CLIENT_PUBSUB | CLIENT_BLOCKED | CLIENT_IGNORE_OVERLOAD)) && ((random() % 100) < g_pserver->overload_protect_strength)) {
-                    for (unsigned i = 0; i < c->vecqueuedcmd.size(); i++) {
-                        addReply(c, shared.overloaderr);
-                    }
-                    c->vecqueuedcmd.clear();
-                } else {
-                    serverTL->vecclientsProcess.push_back(c);
-                }
-            }
-        }
+        if (!c->vecqueuedcmd.empty())
+            serverTL->vecclientsProcess.push_back(c);
     } else {
         // If we're single threaded its actually better to just process the command here while the query is hot in the cache
         //  multithreaded lock contention dominates and batching is better
@@ -2839,9 +2789,9 @@ void processClients()
 {
     serverAssert(GlobalLocksAcquired());
 
-    // Note that this function is reentrant and vecclientsProcess may be modified by code called from processInputBuffer
+    // Note that this function is reentrant and vecclients may be modified by code called from processInputBuffer
     while (!serverTL->vecclientsProcess.empty()) {
-        client *c = *serverTL->vecclientsProcess.begin();
+        client *c = serverTL->vecclientsProcess.front();
         serverTL->vecclientsProcess.erase(serverTL->vecclientsProcess.begin());
 
         /* There is more data in the client input buffer, continue parsing it

src/rdb.cpp

@@ -2865,12 +2865,6 @@ public:
 
     size_t endWork() {
         if (!fLaunched) {
-            for (int idb = 0; idb < cserver.dbnum; ++idb) {
-                if (g_pserver->m_pstorageFactory) {
-                    g_pserver->db[idb]->processChangesAsync(cstorageWritesInFlight);
-                }
-            }
-            while (cstorageWritesInFlight > 0);
             return ckeysLoaded;
         }
         if (!vecbatch.empty()) {

src/redis-cli.c

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

src/replication.cpp

@@ -283,7 +283,7 @@ void resizeReplicationBacklog(long long newsize) {
             
             if (cserver.repl_backlog_disk_size != 0) {
                 if (newsize > g_pserver->repl_backlog_config_size || cserver.force_backlog_disk) {
-                    if (g_pserver->repl_backlog_disk == nullptr || g_pserver->repl_backlog == g_pserver->repl_backlog_disk)
+                    if (g_pserver->repl_backlog == g_pserver->repl_backlog_disk)
                         return; // Can't do anything more
                     serverLog(LL_NOTICE, "Switching to disk backed replication backlog due to exceeding memory limits");
                     backlog = g_pserver->repl_backlog_disk;
@@ -308,25 +308,10 @@ void resizeReplicationBacklog(long long newsize) {
                 auto cbActiveBacklog = cbPhase1 + g_pserver->repl_backlog_idx;
                 serverAssert(g_pserver->repl_backlog_histlen == cbActiveBacklog);
             }
-            if (g_pserver->repl_backlog != g_pserver->repl_backlog_disk) {
+            if (g_pserver->repl_backlog != g_pserver->repl_backlog_disk)
                 zfree(g_pserver->repl_backlog);
-            } else {
+            else
                 serverLog(LL_NOTICE, "Returning to memory backed replication backlog");
-                auto repl_backlog = g_pserver->repl_backlog_disk;
-                g_pserver->repl_backlog_disk = nullptr;
-                g_pserver->asyncworkqueue->AddWorkFunction([repl_backlog, size = cserver.repl_backlog_disk_size]{
-                    // The kernel doesn't make promises with how it will manage the memory but users really want to
-                    //  see the RSS go down.  So lets encourage that to happen.
-                    madvise(repl_backlog, size, MADV_DONTNEED); // NOTE: This will block until all pages are released
-                    aeAcquireLock();
-                    if (g_pserver->repl_backlog_disk == nullptr && cserver.repl_backlog_disk_size == size) {
-                        g_pserver->repl_backlog_disk = repl_backlog;
-                    } else {
-                        munmap(g_pserver->repl_backlog_disk, size);
-                    }
-                    aeReleaseLock();
-                }, true /*fHiPri*/);
-            }
             g_pserver->repl_backlog = backlog;
             g_pserver->repl_backlog_idx = g_pserver->repl_backlog_histlen;
             if (g_pserver->repl_batch_idxStart >= 0) {
@@ -346,17 +331,6 @@ void resizeReplicationBacklog(long long newsize) {
             /* Next byte we have is... the next since the buffer is empty. */
             g_pserver->repl_backlog_off = g_pserver->master_repl_offset+1;
             g_pserver->repl_backlog_start = g_pserver->master_repl_offset;
-            listIter li;
-            listNode *ln;
-            listRewind(g_pserver->slaves, &li);
-            while ((ln = listNext(&li))) {
-                client *replica = (client*)listNodeValue(ln);
-
-                std::unique_lock<fastlock> ul(replica->lock);
-
-                replica->repl_curr_off = g_pserver->master_repl_offset;
-                replica->repl_end_off = g_pserver->master_repl_offset;
-            }
         }
     }
     g_pserver->repl_backlog_size = newsize;
@@ -1045,6 +1019,8 @@ 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(); }
@@ -1059,7 +1035,7 @@ public:
             while (checkClientOutputBufferLimits(replica)
               && (replica->flags.load(std::memory_order_relaxed) & CLIENT_CLOSE_ASAP) == 0) {
                 ul.unlock();
-                usleep(1000);   // give 1ms for the I/O before we poll again
+                usleep(0);
                 ul.lock();
             }
         }
@@ -1117,7 +1093,7 @@ public:
         reply->used += size;
     }
 
-    void addLongLongWithPrefix(long long val, char prefix) {
+    void addLongWithPrefix(long val, char prefix) {
         char buf[128];
         int len;
 
@@ -1129,19 +1105,15 @@ public:
     }
 
     void addArrayLen(int len) {
-        addLongLongWithPrefix(len, '*');
+        addLongWithPrefix(len, '*');
     }
 
     void addLong(long val) {
-        addLongLongWithPrefix(val, ':');
-    }
-
-    void addLongLong(long long val) {
-        addLongLongWithPrefix(val, ':');
+        addLongWithPrefix(val, ':');
     }
 
     void addString(const char *s, unsigned long len) {
-        addLongLongWithPrefix(len, '$');
+        addLongWithPrefix(len, '$');
         addData(s, len);
         addData("\r\n", 2);
     }
@@ -1157,24 +1129,11 @@ public:
         }
     }
 
-    void putReplicasOnline() {
-        for (auto replica : replicas) {
-            std::unique_lock<fastlock> ul(replica->lock);
-            //  If we put the replica online before the output is drained then it will get immediately closed
-            while (checkClientOutputBufferLimits(replica)
-              && (replica->flags.load(std::memory_order_relaxed) & CLIENT_CLOSE_ASAP) == 0) {
-                ul.unlock();
-                usleep(1000);   // give 1ms for the I/O before we poll again
-                ul.lock();
-            }
-        }
-
-        aeAcquireLock();
+    void putSlavesOnline() {
         for (auto replica : replicas) {
             replica->replstate = SLAVE_STATE_FASTSYNC_DONE;
             replica->repl_put_online_on_ack = 1;
         }
-        aeReleaseLock();
     }
 
     void abort() {
@@ -1190,19 +1149,22 @@ int rdbSaveSnapshotForReplication(rdbSaveInfo *rsi) {
     // TODO: This needs to be on a background thread
     int retval = C_OK;
     serverAssert(GlobalLocksAcquired());
-    serverLog(LL_NOTICE, "Starting fast full sync with target: %s", "disk");
+    serverLog(LL_NOTICE, "Starting storage provider 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 ((ln = listNext(&li))) {
+    while (replica == nullptr && (ln = listNext(&li))) {
         client *replicaCur = (client*)listNodeValue(ln);
-        if ((replicaCur->slave_capa & SLAVE_CAPA_KEYDB_FASTSYNC) && (replicaCur->replstate == SLAVE_STATE_WAIT_BGSAVE_START)) {
-            spreplBuf->addReplica(replicaCur);
+        if ((replicaCur->slave_capa & SLAVE_CAPA_ROCKSDB_SNAPSHOT) && (replicaCur->replstate == SLAVE_STATE_WAIT_BGSAVE_START)) {
+            replica = replicaCur;
+            spreplBuf->addReplica(replica);
             replicaCur->replstate = SLAVE_STATE_FASTSYNC_TX;
         }
     }
+    serverAssert(replica != nullptr);
 
     spreplBuf->addArrayLen(2);   // Two sections: Metadata and databases
 
@@ -1217,7 +1179,7 @@ int rdbSaveSnapshotForReplication(rdbSaveInfo *rsi) {
             spreplBuf->addString(rsi->repl_id, CONFIG_RUN_ID_SIZE);
         spreplBuf->addArrayLen(2);
             spreplBuf->addString("repl-offset", 11);
-            spreplBuf->addLongLong(rsi->master_repl_offset);
+            spreplBuf->addLong(rsi->master_repl_offset);
 
     if (dictSize(g_pserver->lua_scripts)) {
         dictEntry *de;
@@ -1233,13 +1195,13 @@ int rdbSaveSnapshotForReplication(rdbSaveInfo *rsi) {
         di = NULL; /* So that we don't release it again on error. */
     }
 
-    std::shared_ptr<std::vector<const redisDbPersistentDataSnapshot*>> spvecsnapshot = std::make_shared<std::vector<const redisDbPersistentDataSnapshot*>>();
+    std::shared_ptr<std::vector<std::unique_ptr<const StorageCache>>> spvecspsnapshot = std::make_shared<std::vector<std::unique_ptr<const StorageCache>>>();
     for (int idb = 0; idb < cserver.dbnum; ++idb) {
-        spvecsnapshot->emplace_back(g_pserver->db[idb]->createSnapshot(getMvccTstamp(), false));
+        spvecspsnapshot->emplace_back(g_pserver->db[idb]->CloneStorageCache());
     }
     aeReleaseLock();
 
-    g_pserver->asyncworkqueue->AddWorkFunction([spreplBuf = std::move(spreplBuf), spvecsnapshot = std::move(spvecsnapshot)]{
+    g_pserver->asyncworkqueue->AddWorkFunction([spreplBuf = std::move(spreplBuf), spvecspsnapshot = std::move(spvecspsnapshot)]{
         int retval = C_OK;
         auto timeStart = ustime();
         auto lastLogTime = timeStart;
@@ -1249,16 +1211,15 @@ int rdbSaveSnapshotForReplication(rdbSaveInfo *rsi) {
         // Databases
         replBuf.addArrayLen(cserver.dbnum);
         for (int idb = 0; idb < cserver.dbnum; ++idb) {
-            auto &spsnapshot = (*spvecsnapshot)[idb];
-            size_t snapshotDeclaredCount = spsnapshot->size();
+            auto &spsnapshot = (*spvecspsnapshot)[idb];
+            size_t snapshotDeclaredCount = spsnapshot->count();
             replBuf.addArrayLen(snapshotDeclaredCount);
             size_t count = 0;
-            bool result = spsnapshot->iterate_threadsafe([&replBuf, &count, &cbData, &lastLogTime, &cbLastUpdate](const char *strKey, robj_roptr o) -> bool{
+            bool result = spsnapshot->enumerate([&replBuf, &count, &cbData, &lastLogTime, &cbLastUpdate](const char *rgchKey, size_t cchKey, const void *rgchVal, size_t cchVal) -> bool{
                 replBuf.addArrayLen(2);
 
-                replBuf.addString(strKey, sdslen(strKey));
-                sds strT = serializeStoredObjectAndExpire(o);
-                replBuf.addString(strT, sdslen(strT));
+                replBuf.addString(rgchKey, cchKey);
+                replBuf.addString((const char *)rgchVal, cchVal);
                 ++count;
                 if ((count % 8092) == 0) {
                     auto curTime = ustime();
@@ -1269,8 +1230,7 @@ int rdbSaveSnapshotForReplication(rdbSaveInfo *rsi) {
                         lastLogTime = ustime();
                     }
                 }
-                cbData += sdslen(strKey) + sdslen(strT);
-                sdsfree(strT);
+                cbData += cchKey + cchVal;
                 return replBuf.isActive();
             });
 
@@ -1295,7 +1255,9 @@ int rdbSaveSnapshotForReplication(rdbSaveInfo *rsi) {
         auto usec = ustime() - timeStart;
         serverLog(LL_NOTICE, "Transferred %zuMB total (%zuMB data) in %.2f seconds.  (%.2fGbit/s)", spreplBuf->cbWritten()/1024/1024, cbData/1024/1024, usec/1000000.0, (spreplBuf->cbWritten()*8.0)/(usec/1000000.0)/1000000000.0);
         if (retval == C_OK) {
-            replBuf.putReplicasOnline();
+            aeAcquireLock();
+            replBuf.putSlavesOnline();
+            aeReleaseLock();
         }
     });
 
@@ -1340,7 +1302,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_KEYDB_FASTSYNC && FFastSyncEnabled())
+        if (mincapa & SLAVE_CAPA_ROCKSDB_SNAPSHOT && g_pserver->m_pstorageFactory)
             retval = rdbSaveSnapshotForReplication(rsiptr);
         else if (socket_target)
             retval = rdbSaveToSlavesSockets(rsiptr);
@@ -1519,7 +1481,7 @@ void syncCommand(client *c) {
     }
 
     /* CASE 0: Fast Sync */
-    if (c->slave_capa & SLAVE_CAPA_KEYDB_FASTSYNC && FFastSyncEnabled()) {
+    if ((c->slave_capa & SLAVE_CAPA_ROCKSDB_SNAPSHOT) && g_pserver->m_pstorageFactory) {
         serverLog(LL_NOTICE,"Fast SYNC on next replication cycle");
     /* CASE 1: BGSAVE is in progress, with disk target. */
     } else if (g_pserver->FRdbSaveInProgress() &&
@@ -1710,8 +1672,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]), "keydb-fastsync"))
-                c->slave_capa |= SLAVE_CAPA_KEYDB_FASTSYNC;
+            else if (!strcasecmp((const char*)ptrFromObj(c->argv[j+1]), "rocksdb-snapshot-load"))
+                c->slave_capa |= SLAVE_CAPA_ROCKSDB_SNAPSHOT;
 
             fCapaCommand = true;
         } else if (!strcasecmp((const char*)ptrFromObj(c->argv[j]),"ack")) {
@@ -1776,14 +1738,10 @@ void replconfCommand(client *c) {
 
     if (fCapaCommand) {
         sds reply = sdsnew("+OK");
-        if (g_pserver->fActiveReplica) {
+        if (g_pserver->fActiveReplica)
             reply = sdscat(reply, " active-replica");
-        }
-        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
-        }
+        if (g_pserver->m_pstorageFactory && (c->slave_capa & SLAVE_CAPA_ROCKSDB_SNAPSHOT) && !g_pserver->fActiveReplica)
+            reply = sdscat(reply, " rocksdb-snapshot-save");
         reply = sdscat(reply, "\r\n");
         addReplySds(c, reply);
     } else {
@@ -2532,7 +2490,7 @@ size_t parseCount(const char *rgch, size_t cch, long long *pvalue) {
     return cchNumeral + 3;
 }
 
-bool readFastSyncBulkPayload(connection *conn, redisMaster *mi, rdbSaveInfo &rsi) {
+bool readSnapshotBulkPayload(connection *conn, redisMaster *mi, rdbSaveInfo &rsi) {
     int fUpdate = g_pserver->fActiveReplica || g_pserver->enable_multimaster;
     serverAssert(GlobalLocksAcquired());
     serverAssert(mi->master == nullptr);
@@ -2557,11 +2515,6 @@ bool readFastSyncBulkPayload(connection *conn, redisMaster *mi, rdbSaveInfo &rsi
         }
     }
 
-    if (mi->repl_state == REPL_STATE_WAIT_STORAGE_IO) {
-        goto LWaitIO;
-    }
-
-    serverAssert(mi->parseState != nullptr);
     for (int iter = 0; iter < 10; ++iter) {
         if (mi->parseState->shouldThrottle())
             return false;
@@ -2572,16 +2525,11 @@ bool readFastSyncBulkPayload(connection *conn, redisMaster *mi, rdbSaveInfo &rsi
 
         auto nread = connRead(conn, mi->bulkreadBuffer+qblen, readlen);
         if (nread <= 0) {
-            if (connGetState(conn) != CONN_STATE_CONNECTED) {
-                serverLog(LL_WARNING,"I/O error trying to sync with MASTER: %s",
-                    (nread == -1) ? strerror(errno) : "connection lost");
+            if (connGetState(conn) != CONN_STATE_CONNECTED)
                 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;
@@ -2653,9 +2601,8 @@ bool readFastSyncBulkPayload(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) {
+            if (sdslen(str) == CONFIG_RUN_ID_SIZE+1) {
                 memcpy(rsi.repl_id, str, CONFIG_RUN_ID_SIZE+1);
-                rsi.repl_id_is_set = 1;
             }
 
             fFinished = true;
@@ -2678,14 +2625,7 @@ bool readFastSyncBulkPayload(connection *conn, redisMaster *mi, rdbSaveInfo &rsi
     if (!fFinished)
         return false;
 
-LWaitIO:
-    if (mi->parseState->hasIOInFlight()) {
-        mi->repl_state = REPL_STATE_WAIT_STORAGE_IO;
-        return false;
-    }
-
     serverLog(LL_NOTICE, "Fast sync complete");
-    serverAssert(!mi->parseState->hasIOInFlight());
     delete mi->parseState;
     mi->parseState = nullptr;
     return true;
@@ -3061,8 +3001,8 @@ void readSyncBulkPayload(connection *conn) {
         return;
     }
 
-    if (mi->isKeydbFastsync) {
-        if (!readFastSyncBulkPayload(conn, mi, rsi))
+    if (mi->isRocksdbSnapshotRepl) {
+        if (!readSnapshotBulkPayload(conn, mi, rsi))
             return;
     } else {
         if (!readSyncBulkPayloadRdb(conn, mi, rsi, usemark))
@@ -3074,15 +3014,17 @@ void readSyncBulkPayload(connection *conn) {
 
     /* Final setup of the connected slave <- master link */
     replicationCreateMasterClient(mi,mi->repl_transfer_s,rsi.repl_stream_db);
-    if (mi->isKeydbFastsync) {
+    if (mi->isRocksdbSnapshotRepl) {
         /* 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;
@@ -3121,13 +3063,13 @@ void readSyncBulkPayload(connection *conn) {
     }
 
     /* Send the initial ACK immediately to put this replica in online state. */
-    if (usemark || mi->isKeydbFastsync) replicationSendAck(mi);
+    if (usemark || mi->isRocksdbSnapshotRepl) 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->isKeydbFastsync)
+    if (mi->isRocksdbSnapshotRepl)
         readQueryFromClient(conn); // There may be querybuf data we just appeneded
     return;
 }
@@ -3453,15 +3395,15 @@ void parseMasterCapa(redisMaster *mi, sds strcapa)
     char *pchEnd = szStart;
 
     mi->isActive = false;
-    mi->isKeydbFastsync = false;
+    mi->isRocksdbSnapshotRepl = false;
     for (;;)
     {
         if (*pchEnd == ' ' || *pchEnd == '\0') {
             // Parse the word
             if (strncmp(szStart, "active-replica", pchEnd - szStart) == 0) {
                 mi->isActive = true;
-            } else if (strncmp(szStart, "keydb-fastsync-save", pchEnd - szStart) == 0) {
-                mi->isKeydbFastsync = true;
+            } else if (strncmp(szStart, "rocksdb-snapshot-save", pchEnd - szStart) == 0) {
+                mi->isRocksdbSnapshotRepl = true;
             }
             szStart = pchEnd + 1;
         }
@@ -3608,9 +3550,9 @@ retry_connect:
             "capa","psync2",
             "capa","activeExpire",
         };
-        if (FFastSyncEnabled() && g_pserver->repl_diskless_load != REPL_DISKLESS_LOAD_SWAPDB) {
+        if (g_pserver->m_pstorageFactory && !g_pserver->fActiveReplica && g_pserver->repl_diskless_load != REPL_DISKLESS_LOAD_SWAPDB) {
             veccapabilities.push_back("capa");
-            veccapabilities.push_back("keydb-fastsync");
+            veccapabilities.push_back("rocksdb-snapshot-load");
         }
 
         err = sendCommandArgv(conn, veccapabilities.size(), veccapabilities.data(), nullptr);
@@ -3793,7 +3735,7 @@ retry_connect:
     }
 
     /* Prepare a suitable temp file for bulk transfer */
-    if (!useDisklessLoad() && !mi->isKeydbFastsync) {
+    if (!useDisklessLoad() && !mi->isRocksdbSnapshotRepl) {
         while(maxtries--) {
             auto dt = std::chrono::system_clock::now().time_since_epoch();
             auto dtMillisecond = std::chrono::duration_cast<std::chrono::milliseconds>(dt);
@@ -3922,10 +3864,6 @@ 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);
@@ -4391,6 +4329,7 @@ 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);
@@ -4829,10 +4768,6 @@ void replicationCron(void) {
     {
         redisMaster *mi = (redisMaster*)listNodeValue(lnMaster);
 
-        if (mi->repl_state == REPL_STATE_WAIT_STORAGE_IO && !mi->parseState->hasIOInFlight()) {
-            readSyncBulkPayload(mi->repl_transfer_s);
-        }
-
         std::unique_lock<decltype(mi->master->lock)> ulock;
         if (mi->master != nullptr)
             ulock = decltype(ulock)(mi->master->lock);
@@ -4849,8 +4784,7 @@ void replicationCron(void) {
 
         /* Bulk transfer I/O timeout? */
         if (mi->masterhost && mi->repl_state == REPL_STATE_TRANSFER &&
-            (time(NULL)-mi->repl_transfer_lastio) > g_pserver->repl_timeout &&
-            !(g_pserver->loading & LOADING_REPLICATION))
+            (time(NULL)-mi->repl_transfer_lastio) > g_pserver->repl_timeout)
         {
             serverLog(LL_WARNING,"Timeout receiving bulk data from MASTER... If the problem persists try to set the 'repl-timeout' parameter in keydb.conf to a larger value.");
             cancelReplicationHandshake(mi,true);

src/sds.h

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

src/server.cpp

@@ -1961,6 +1961,15 @@ void getExpansiveClientsInfo(size_t *in_usage, size_t *out_usage) {
     *out_usage = o;
 }
 
+int closeClientOnOverload(client *c) {
+    if (g_pserver->overload_closed_clients > MAX_CLIENTS_SHED_PER_PERIOD) return false;
+    if (!g_pserver->is_overloaded) return false;
+    // Don't close masters, replicas, or pub/sub clients
+    if (c->flags & (CLIENT_MASTER | CLIENT_SLAVE | CLIENT_PENDING_WRITE | CLIENT_PUBSUB | CLIENT_BLOCKED)) return false;
+    freeClient(c);
+    ++g_pserver->overload_closed_clients;
+    return true;
+}
 
 /* This function is called by serverCron() and is used in order to perform
  * operations on clients that are important to perform constantly. For instance
@@ -2032,6 +2041,7 @@ void clientsCron(int iel) {
             if (clientsCronTrackExpansiveClients(c, curr_peak_mem_usage_slot)) goto LContinue;
             if (clientsCronTrackClientsMemUsage(c)) goto LContinue;
             if (closeClientOnOutputBufferLimitReached(c, 0)) continue; // Client also free'd
+            if (closeClientOnOverload(c)) continue;
         LContinue:
             fastlock_unlock(&c->lock);
         }        
@@ -2591,13 +2601,8 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
 
     /* Check for CPU Overload */
     run_with_period(10'000) {
-        if (g_pserver->is_overloaded) {
-            if (g_pserver->overload_protect_strength < 100)
-                g_pserver->overload_protect_strength *= 2;
-        } else {
-            g_pserver->overload_protect_strength = g_pserver->overload_protect_tenacity;
-        }
         g_pserver->is_overloaded = false;
+        g_pserver->overload_closed_clients = 0;
         static clock_t last = 0;
         if (g_pserver->overload_protect_threshold > 0) {
             clock_t cur = clock();
@@ -2605,7 +2610,6 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
             perc /= cserver.cthreads;
             perc *= 100.0;
             serverLog(LL_WARNING, "CPU Used: %.2f", perc);
-            g_pserver->last_overload_cpu_reading = static_cast<float>(perc);
             if (perc > g_pserver->overload_protect_threshold) {
                 serverLog(LL_WARNING, "\tWARNING: CPU overload detected.");
                 g_pserver->is_overloaded = true;
@@ -2838,20 +2842,6 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
 
     locker.arm();
 
-    for (auto *tok : serverTL->setStorageTokensProcess) {
-        tok->db->processStorageToken(tok);
-    }
-    serverTL->setStorageTokensProcess.clear();
-
-    if (g_pserver->m_pstorageFactory != nullptr && !serverTL->setclientsPrefetch.empty()) {
-        g_pserver->db[0]->prefetchKeysFlash(serverTL->setclientsPrefetch);
-        for (client *c : serverTL->setclientsPrefetch) {
-            if (!(c->flags & CLIENT_BLOCKED))
-                serverTL->vecclientsProcess.push_back(c);
-        }
-        serverTL->setclientsPrefetch.clear();
-    }
-
     /* end any snapshots created by fast async commands */
     for (int idb = 0; idb < cserver.dbnum; ++idb) {
         if (serverTL->rgdbSnapshot[idb] != nullptr && serverTL->rgdbSnapshot[idb]->FStale()) {
@@ -3066,8 +3056,10 @@ void afterSleep(struct aeEventLoop *eventLoop) {
         serverAssert(serverTL->gcEpoch.isReset());
         serverTL->gcEpoch = g_pserver->garbageCollector.startEpoch();
 
+        aeAcquireLock();
         for (int idb = 0; idb < cserver.dbnum; ++idb)
             g_pserver->db[idb]->trackChanges(false);
+        aeReleaseLock();
 
         serverTL->disable_async_commands = false;
     }
@@ -3127,8 +3119,7 @@ void createSharedObjects(void) {
         "-NOREPLICAS Not enough good replicas to write.\r\n")));
     shared.busykeyerr = makeObjectShared(createObject(OBJ_STRING,sdsnew(
         "-BUSYKEY Target key name already exists.\r\n")));
-    shared.overloaderr = makeObjectShared(createObject(OBJ_STRING,sdsnew(
-        "-OVERLOAD KeyDB is overloaded.\r\n")));
+    
 
     /* The shared NULL depends on the protocol version. */
     shared.null[0] = NULL;
@@ -3262,7 +3253,6 @@ void initMasterInfo(redisMaster *master)
     master->repl_state = REPL_STATE_NONE;
     master->repl_down_since = 0; /* Never connected, repl is down since EVER. */
     master->mvccLastSync = 0;
-    master->repl_transfer_fd = -1;
 }
 
 void initServerConfig(void) {
@@ -4049,7 +4039,7 @@ void initServer(void) {
     g_pserver->cron_malloc_stats.allocator_active = 0;
     g_pserver->cron_malloc_stats.allocator_resident = 0;
     g_pserver->cron_malloc_stats.sys_available = 0;
-    g_pserver->cron_malloc_stats.sys_total = getMemTotal();
+    g_pserver->cron_malloc_stats.sys_total = g_pserver->force_eviction_percent ? getMemTotal() : 0;
     g_pserver->lastbgsave_status = C_OK;
     g_pserver->aof_last_write_status = C_OK;
     g_pserver->aof_last_write_errno = 0;
@@ -4160,16 +4150,6 @@ void InitServerLast() {
 
     g_pserver->asyncworkqueue = new (MALLOC_LOCAL) AsyncWorkQueue(cserver.cthreads);
 
-    if (g_pserver->m_pstorageFactory != nullptr) {
-        g_pserver->asyncreadworkqueue = new (MALLOC_LOCAL) AsyncWorkQueue(cserver.cthreads);
-
-        //Process one write/commit at a time to ensure consistency
-        g_pserver->asyncwriteworkqueue = new (MALLOC_LOCAL) AsyncWorkQueue(1);
-    } else {
-        g_pserver->asyncreadworkqueue = nullptr;
-        g_pserver->asyncwriteworkqueue = nullptr;
-    }
-
     // Allocate the repl backlog
     
 }
@@ -5122,8 +5102,8 @@ int processCommand(client *c, int callFlags) {
 
             sds str = sdsempty();
             for (int j = 0; j < numkeys; j++) {
-                str = sdscatsds(str, (sds)ptrFromObj(c->argv[keyindex[j]]));
-                str = sdscat(str, " ");
+                sdscatsds(str, (sds)ptrFromObj(c->argv[keyindex[j]]));
+                sdscat(str, " ");
             }
         
             if (numkeys > 0)
@@ -5827,9 +5807,7 @@ sds genRedisInfoString(const char *section) {
             "lazyfree_pending_objects:%zu\r\n"
             "lazyfreed_objects:%zu\r\n"
             "storage_provider:%s\r\n"
-            "available_system_memory:%s\r\n"
-            "maxstorage:%llu\r\n"
-            "storage_used:%lu\r\n",
+            "available_system_memory:%s\r\n",
             zmalloc_used,
             hmem,
             g_pserver->cron_malloc_stats.process_rss,
@@ -5875,9 +5853,7 @@ sds genRedisInfoString(const char *section) {
             lazyfreeGetPendingObjectsCount(),
             lazyfreeGetFreedObjectsCount(),
             g_pserver->m_pstorageFactory ? g_pserver->m_pstorageFactory->name() : "none",
-            available_system_mem,
-            g_pserver->maxstorage,
-            g_pserver->m_pstorageFactory ? g_pserver->m_pstorageFactory->totalDiskspaceUsed() : 0
+            available_system_mem
         );
         freeMemoryOverheadData(mh);
     }
@@ -6295,15 +6271,13 @@ sds genRedisInfoString(const char *section) {
         "used_cpu_sys_children:%ld.%06ld\r\n"
         "used_cpu_user_children:%ld.%06ld\r\n"
         "server_threads:%d\r\n"
-        "long_lock_waits:%" PRIu64 "\r\n"
-        "last_overload_cpu_reading:%.2f\r\n",
+        "long_lock_waits:%" PRIu64 "\r\n",
         (long)self_ru.ru_stime.tv_sec, (long)self_ru.ru_stime.tv_usec,
         (long)self_ru.ru_utime.tv_sec, (long)self_ru.ru_utime.tv_usec,
         (long)c_ru.ru_stime.tv_sec, (long)c_ru.ru_stime.tv_usec,
         (long)c_ru.ru_utime.tv_sec, (long)c_ru.ru_utime.tv_usec,
         cserver.cthreads,
-        fastlock_getlongwaitcount(),
-        g_pserver->last_overload_cpu_reading);
+        fastlock_getlongwaitcount());
 #ifdef RUSAGE_THREAD
         struct rusage m_ru;
         getrusage(RUSAGE_THREAD, &m_ru);
@@ -7439,11 +7413,6 @@ static void validateConfiguration()
         exit(EXIT_FAILURE);
     }
 
-    if (g_pserver->repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB && g_pserver->m_pstorageFactory) {
-        serverLog(LL_WARNING, "SWAPDB is not implemented when using a storage provider.");
-        exit(EXIT_FAILURE);
-    }
-
     g_pserver->repl_backlog_size = g_pserver->repl_backlog_config_size; // this is normally set in the update logic, but not on initial config
 }
 

src/server.h

@@ -64,7 +64,6 @@
 #include <string>
 #include <mutex>
 #include <unordered_set>
-#include <arpa/inet.h>
 #ifdef __cplusplus
 extern "C" {
 #include <lua.h>
@@ -126,25 +125,9 @@ typedef long long ustime_t; /* microsecond time type. */
 #define OVERLOAD_PROTECT_PERIOD_MS 10'000 // 10 seconds
 #define MAX_CLIENTS_SHED_PER_PERIOD (OVERLOAD_PROTECT_PERIOD_MS / 10)  // Restrict to one client per 10ms
 
-#define IPV4_BITS 32
-#define IPV6_BITS 128
-
 extern int g_fTestMode;
 extern struct redisServer *g_pserver;
 
-class TCleanup {
-    std::function<void()> fn;
-
-public:
-    TCleanup(std::function<void()> fn)
-        : fn(fn)
-    {}
-
-    ~TCleanup() {
-        fn();
-    }
-};
-
 struct redisObject;
 class robj_roptr
 {
@@ -563,7 +546,6 @@ extern int configOOMScoreAdjValuesDefaults[CONFIG_OOM_COUNT];
                                           RDB without replication buffer. */
 #define CLIENT_FORCE_REPLY (1ULL<<44) /* Should addReply be forced to write the text? */
 #define CLIENT_AUDIT_LOGGING (1ULL<<45) /* Client commands required audit logging */
-#define CLIENT_IGNORE_OVERLOAD (1ULL<<46) /* Client that should not be disconnected by overload protection */
 
 /* Client block type (btype field in client structure)
  * if CLIENT_BLOCKED flag is set. */
@@ -575,8 +557,7 @@ extern int configOOMScoreAdjValuesDefaults[CONFIG_OOM_COUNT];
 #define BLOCKED_ZSET 5    /* BZPOP et al. */
 #define BLOCKED_PAUSE 6   /* Blocked by CLIENT PAUSE */
 #define BLOCKED_ASYNC 7
-#define BLOCKED_STORAGE 8
-#define BLOCKED_NUM 9     /* Number of blocked states. */
+#define BLOCKED_NUM 8     /* Number of blocked states. */
 
 /* Client request types */
 #define PROTO_REQ_INLINE 1
@@ -612,7 +593,6 @@ typedef enum {
     REPL_STATE_RECEIVE_PSYNC_REPLY, /* Wait for PSYNC reply */
     /* --- End of handshake states --- */
     REPL_STATE_TRANSFER,        /* Receiving .rdb from master */
-    REPL_STATE_WAIT_STORAGE_IO,
     REPL_STATE_CONNECTED,       /* Connected to master */
 } repl_state;
 
@@ -640,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_KEYDB_FASTSYNC (1<<3)
+#define SLAVE_CAPA_ROCKSDB_SNAPSHOT (1<<3)
 
 /* Synchronous read timeout - replica side */
 #define CONFIG_REPL_SYNCIO_TIMEOUT 5
@@ -1231,8 +1211,6 @@ public:
     bool keycacheIsEnabled();
 
     void prefetchKeysAsync(client *c, struct parsed_command &command);
-    void prefetchKeysFlash(const std::unordered_set<client*> &setc);
-    void processStorageToken(StorageToken *tok);
 
     bool FSnapshot() const { return m_spdbSnapshotHOLDER != nullptr; }
 
@@ -1242,7 +1220,6 @@ public:
 
     dict_iter find_cached_threadsafe(const char *key) const;
 
-    static void storageLoadCallback(struct aeEventLoop *el, struct StorageToken *token);
     static void activeExpireCycleCore(int type);
 
 protected:
@@ -1392,8 +1369,6 @@ struct redisDb : public redisDbPersistentDataSnapshot
     using redisDbPersistentData::keycacheIsEnabled;
     using redisDbPersistentData::dictUnsafeKeyOnly;
     using redisDbPersistentData::prefetchKeysAsync;
-    using redisDbPersistentData::prefetchKeysFlash;
-    using redisDbPersistentData::processStorageToken;
     using redisDbPersistentData::prepOverwriteForSnapshot;
     using redisDbPersistentData::FRehashing;
     using redisDbPersistentData::FTrackingChanges;
@@ -1635,6 +1610,7 @@ 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 */
@@ -1769,7 +1745,7 @@ struct sharedObjectsStruct {
     *emptyarray, *wrongtypeerr, *nokeyerr, *syntaxerr, *sameobjecterr,
     *outofrangeerr, *noscripterr, *loadingerr, *slowscripterr, *bgsaveerr,
     *masterdownerr, *roslaveerr, *execaborterr, *noautherr, *noreplicaserr,
-    *busykeyerr, *oomerr, *overloaderr, *plus, *messagebulk, *pmessagebulk, *subscribebulk,
+    *busykeyerr, *oomerr, *plus, *messagebulk, *pmessagebulk, *subscribebulk,
     *unsubscribebulk, *psubscribebulk, *punsubscribebulk, *del, *unlink,
     *rpop, *lpop, *lpush, *rpoplpush, *lmove, *blmove, *zpopmin, *zpopmax,
     *emptyscan, *multi, *exec, *left, *right, *hset, *srem, *xgroup, *xclaim,  
@@ -1895,7 +1871,7 @@ struct redisMaster {
     long long master_initial_offset;           /* Master PSYNC offset. */
 
     bool isActive = false;
-    bool isKeydbFastsync = false;
+    bool isRocksdbSnapshotRepl = 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. */
@@ -2232,8 +2208,6 @@ struct redisServerThreadVars {
     int propagate_in_transaction = 0;  /* Make sure we don't propagate nested MULTI/EXEC */
     int client_pause_in_transaction = 0; /* Was a client pause executed during this Exec? */
     std::vector<client*> vecclientsProcess;
-    std::unordered_set<client*> setclientsPrefetch;
-    std::unordered_set<StorageToken*> setStorageTokensProcess;
     dictAsyncRehashCtl *rehashCtl = nullptr;
 
     int getRdbKeySaveDelay();
@@ -2573,7 +2547,6 @@ struct redisServer {
     int repl_diskless_sync_delay;   /* Delay to start a diskless repl BGSAVE. */
     std::atomic <long long> repl_lowest_off; /* The lowest offset amongst all replicas
                                                 -1 if there are no replicas */
-    long long repl_backlog_max_writes_per_event; /* Caps the output buffer sent to the replica at once */
     /* Replication (replica) */
     list *masters;
     int enable_multimaster; 
@@ -2715,7 +2688,6 @@ 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
@@ -2723,8 +2695,6 @@ struct redisServer {
     uint64_t mvcc_tstamp;
 
     AsyncWorkQueue *asyncworkqueue;
-    AsyncWorkQueue *asyncreadworkqueue;
-    AsyncWorkQueue *asyncwriteworkqueue;
 
     /* System hardware info */
     size_t system_memory_size;  /* Total memory in system as reported by OS */
@@ -2743,95 +2713,7 @@ struct redisServer {
     int tls_rotation;
 
     std::set<sdsstring> tls_auditlog_blocklist; /* Certificates that can be excluded from audit logging */
-    std::set<sdsstring> tls_overload_ignorelist; /* Certificates that are be excluded load shedding */
     std::set<sdsstring> tls_allowlist;
-    class IPV4 {
-        struct in_addr m_ip;
-        struct in_addr m_mask;
-
-        int bitsFromMask() const {
-            uint32_t mask = ntohl(m_mask.s_addr);
-            int bits = 0;
-            while (mask > 0) {
-                bits += mask & 1;
-                mask >>= 1;
-            }
-            return bits;
-        }
-
-    public:
-        IPV4(struct in_addr ip, struct in_addr mask) : m_ip(ip), m_mask(mask) {};
-        bool match(struct in_addr ip) const
-        {
-            return (ip.s_addr & m_mask.s_addr) == m_ip.s_addr;
-        }
-
-        sds getString() const
-        {
-            sds result = sdsempty();
-            char buf[INET_ADDRSTRLEN];
-            inet_ntop(AF_INET, &m_ip, buf, sizeof(buf));
-            result = sdscat(result, buf);
-            int bits = bitsFromMask();
-            if (bits != IPV4_BITS) {
-                result = sdscat(result, "/");
-                result = sdscat(result, std::to_string(bits).c_str());
-            }
-            return result;
-        }
-
-        bool operator<(const IPV4& rhs) const
-        {
-            return memcmp(&m_ip, &rhs.m_ip, sizeof(m_ip)) < 0 || (memcmp(&m_ip, &rhs.m_ip, sizeof(m_ip)) == 0 && memcmp(&m_mask, &rhs.m_mask, sizeof(m_mask)) < 0);
-        }
-    };
-    class IPV6 {
-        struct in6_addr m_ip;
-        struct in6_addr m_mask;
-
-        int bitsFromMask() const {
-            int bits = 0;
-            for (unsigned int i = 0; i < sizeof(struct in6_addr); i++) {
-                uint8_t mask = m_mask.s6_addr[i];
-                while (mask > 0) {
-                    bits += mask & 1;
-                    mask >>= 1;
-                }
-            }
-            return bits;
-        }
-    public:
-        IPV6(struct in6_addr ip, struct in6_addr mask) : m_ip(ip), m_mask(mask) {};
-        bool match(struct in6_addr ip) const
-        {
-            for (unsigned int i = 0; i < sizeof(struct in6_addr); i++) {
-                if ((ip.s6_addr[i] & m_mask.s6_addr[i]) != m_ip.s6_addr[i])
-                    return false;
-            }
-            return true;
-        }
-
-        sds getString() const
-        {
-            sds result = sdsempty();
-            char buf[INET6_ADDRSTRLEN];
-            inet_ntop(AF_INET6, &m_ip, buf, sizeof(buf));
-            result = sdscat(result, buf);
-            int bits = bitsFromMask();
-            if (bits != IPV6_BITS) {
-                result = sdscat(result, "/");
-                result = sdscat(result, std::to_string(bits).c_str());
-            }
-            return result;
-        }
-
-        bool operator<(const IPV6& rhs) const
-        {
-            return memcmp(&m_ip, &rhs.m_ip, sizeof(m_ip)) < 0 || (memcmp(&m_ip, &rhs.m_ip, sizeof(m_ip)) == 0 && memcmp(&m_mask, &rhs.m_mask, sizeof(m_mask)) < 0);
-        }
-    };
-    std::set<IPV4> overload_ignorelist;
-    std::set<IPV6> overload_ignorelist_ipv6;
     redisTLSContextConfig tls_ctx_config;
 
     /* cpu affinity */
@@ -2874,14 +2756,12 @@ struct redisServer {
 
     sds sdsAvailabilityZone;
     int overload_protect_threshold = 0;
-    int overload_protect_tenacity = 0;
-    int overload_protect_strength = 0;
-    float last_overload_cpu_reading = 0.0f;
     int is_overloaded = 0;
+    int overload_closed_clients = 0;
 
-    int module_blocked_pipe[2]; /* Pipe used to awake the event loop if a
-                                client blocked on a module command needs
-                                to be processed. */
+        int module_blocked_pipe[2]; /* Pipe used to awake the event loop if a
+                            client blocked on a module command needs
+                            to be processed. */
 
     bool FRdbSaveInProgress() const { return g_pserver->rdbThreadVars.fRdbThreadActive; }
 };
@@ -3302,7 +3182,6 @@ 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)
 
@@ -4092,10 +3971,6 @@ 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/storage/rocksdb.cpp

@@ -25,7 +25,7 @@ bool FInternalKey(const char *key, size_t cch)
 std::string getPrefix(unsigned int hashslot)
 {
     char *hash_char = (char *)&hashslot;
-    return std::string(hash_char, 2);
+    return std::string(hash_char + (sizeof(unsigned int) - 2), 2);
 }
 
 std::string prefixKey(const char *key, size_t cchKey)
@@ -202,26 +202,23 @@ 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;
-    serverAssert(prefix.size() >= 2);
-    bool full_iter = true;
-    for (it->Seek(rocksdb::Slice(prefix.data(), prefix.size())); it->Valid(); it->Next()) {
+    for (it->Seek(prefix.c_str()); it->Valid(); it->Next()) {
         if (FInternalKey(it->key().data(), it->key().size()))
             continue;
-        if (it->key().size() < 2 || memcmp(it->key().data(),prefix.data(),2) != 0)
+        if (strncmp(it->key().data(),prefix.c_str(),2) != 0)
             break;
         ++count;
         bool fContinue = fn(it->key().data()+2, it->key().size()-2, it->value().data(), it->value().size());
-        if (!fContinue) {
-            full_iter = false;
+        if (!fContinue)
             break;
-        }
     }
+    bool full_iter = !it->Valid() || (strncmp(it->key().data(),prefix.c_str(),2) != 0);
     if (full_iter && count != g_pserver->cluster->slots_keys_count[hashslot])
     {
-        serverLog(LL_WARNING, "WARNING: rocksdb hashslot %d count mismatch %zu vs expected %lu", hashslot, count, g_pserver->cluster->slots_keys_count[hashslot]);
+        printf("WARNING: rocksdb hashslot count mismatch");
     }
-    serverAssert(!full_iter || count == g_pserver->cluster->slots_keys_count[hashslot]);
-    serverAssert(it->status().ok()); // Check for any errors found during the scan
+    assert(!full_iter || count == g_pserver->cluster->slots_keys_count[hashslot]);
+    assert(it->status().ok()); // Check for any errors found during the scan
     return full_iter;
 }
 
@@ -337,31 +334,6 @@ void RocksDBStorageProvider::endWriteBatch()
     m_lock.unlock();
 }
 
-struct BatchStorageToken : public StorageToken {
-    std::shared_ptr<rocksdb::DB> tspdb;    // Note: This must be first so it is deleted last
-    std::unique_ptr<rocksdb::WriteBatchWithIndex> tspbatch;
-};
-
-StorageToken* RocksDBStorageProvider::begin_endWriteBatch(struct aeEventLoop *el, aePostFunctionTokenProc* callback){
-    BatchStorageToken *tok = new BatchStorageToken();
-    tok->tspbatch = std::move(m_spbatch);
-    tok->tspdb = m_spdb;
-    m_spbatch = nullptr;
-    m_lock.unlock();
-    (*m_pfactory->m_wwqueue)->AddWorkFunction([this, el,callback,tok]{
-        tok->tspdb->Write(WriteOptions(),tok->tspbatch.get()->GetWriteBatch());
-        aePostFunction(el,callback,tok);
-    });
-
-    return tok;
-}
-
-void RocksDBStorageProvider::complete_endWriteBatch(StorageToken* tok){
-    delete tok;
-    tok = nullptr;
-}
-
-
 void RocksDBStorageProvider::batch_lock()
 {
     m_lock.lock();
@@ -386,63 +358,10 @@ bool RocksDBStorageProvider::FKeyExists(std::string& key) const
     return m_spdb->Get(ReadOptions(), m_spcolfamily.get(), rocksdb::Slice(key), &slice).ok();
 }
 
-struct RetrievalStorageToken : public StorageToken {
-    std::unordered_map<std::string, std::unique_ptr<rocksdb::PinnableSlice>> mapkeydata;
-};
-
-StorageToken *RocksDBStorageProvider::begin_retrieve(struct aeEventLoop *el, aePostFunctionTokenProc callback, sds *rgkey, size_t ckey) {
-    RetrievalStorageToken *tok = new RetrievalStorageToken();
-
-    for (size_t ikey = 0; ikey < ckey; ++ikey) {
-        tok->mapkeydata.insert(std::make_pair(std::string(rgkey[ikey], sdslen(rgkey[ikey])), nullptr));
-    }
-
-    auto opts = ReadOptions();
-    opts.async_io = true;
-    (*m_pfactory->m_rwqueue)->AddWorkFunction([this, el, callback, tok, opts]{
-        std::vector<std::string> veckeysStr;
-        std::vector<rocksdb::Slice> veckeys;
-        std::vector<rocksdb::PinnableSlice> vecvals;
-        std::vector<rocksdb::Status> vecstatus;
-        veckeys.reserve(tok->mapkeydata.size());
-        veckeysStr.reserve(tok->mapkeydata.size());
-        vecvals.resize(tok->mapkeydata.size());
-        vecstatus.resize(tok->mapkeydata.size());
-        for (auto &pair : tok->mapkeydata) {
-            veckeysStr.emplace_back(prefixKey(pair.first.data(), pair.first.size()));
-            veckeys.emplace_back(veckeysStr.back().data(), veckeysStr.back().size());
-        }
-
-        m_spdb->MultiGet(ReadOptions(),
-            m_spcolfamily.get(),
-            veckeys.size(), const_cast<const rocksdb::Slice*>(veckeys.data()),
-            vecvals.data(), vecstatus.data());
-
-        auto itrDst = tok->mapkeydata.begin();
-        for (size_t ires = 0; ires < veckeys.size(); ++ires) {
-            if (vecstatus[ires].ok()) {
-                itrDst->second = std::make_unique<rocksdb::PinnableSlice>(std::move(vecvals[ires]));
-            }
-            ++itrDst;
-        }
-        aePostFunction(el, callback, tok);
-    });
-    return tok;
-}
-
-void RocksDBStorageProvider::complete_retrieve(StorageToken *tok, callbackSingle fn) {
-    RetrievalStorageToken *rtok = reinterpret_cast<RetrievalStorageToken*>(tok);
-    for (auto &pair : rtok->mapkeydata) {
-        if (pair.second != nullptr) {
-            fn(pair.first.data(), pair.first.size(), pair.second->data(), pair.second->size());
-        }
-    }
-    delete rtok;
-}
 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

@@ -32,10 +32,6 @@ public:
     virtual void insert(const char *key, size_t cchKey, void *data, size_t cb, bool fOverwrite) override;
     virtual bool erase(const char *key, size_t cchKey) override;
     virtual void retrieve(const char *key, size_t cchKey, callbackSingle fn) const override;
-
-    virtual StorageToken *begin_retrieve(struct aeEventLoop *el, aePostFunctionTokenProc callback, sds *rgkey, size_t ckey);
-    virtual void complete_retrieve(StorageToken *tok, callbackSingle fn);
-
     virtual size_t clear() override;
     virtual bool enumerate(callback fn) const override;
     virtual bool enumerate_hashslot(callback fn, unsigned int hashslot) const override;
@@ -49,8 +45,6 @@ public:
 
     virtual void beginWriteBatch() override;
     virtual void endWriteBatch() override;
-    virtual StorageToken* begin_endWriteBatch(struct aeEventLoop *el, aePostFunctionTokenProc* proc);
-    virtual void complete_endWriteBatch(StorageToken *tok);
 
     virtual void bulkInsert(char **rgkeys, size_t *rgcbkeys, char **rgvals, size_t *rgcbvals, size_t celem) override;
 

src/storage/rocksdbfactor_internal.h

@@ -1,6 +1,5 @@
 #pragma once
 #include "rocksdb.h"
-#include "../AsyncWorkQueue.h"
 
 class RocksDBStorageFactory : public IStorageFactory
 {
@@ -12,10 +11,7 @@ class RocksDBStorageFactory : public IStorageFactory
     bool m_fCreatedTempFolder = false;
 
 public:
-    AsyncWorkQueue **m_rwqueue;
-    AsyncWorkQueue **m_wwqueue;
-
-    RocksDBStorageFactory(const char *dbfile, int dbnum, const char *rgchConfig, size_t cchConfig, AsyncWorkQueue **rwqueue, AsyncWorkQueue **wwqueue);
+    RocksDBStorageFactory(const char *dbfile, int dbnum, const char *rgchConfig, size_t cchConfig);
     ~RocksDBStorageFactory();
 
     virtual IStorage *create(int db, key_load_iterator iter, void *privdata) override;

src/storage/rocksdbfactory.cpp

@@ -35,9 +35,9 @@ rocksdb::Options DefaultRocksDBOptions() {
     return options;
 }
 
-IStorageFactory *CreateRocksDBStorageFactory(const char *path, int dbnum, const char *rgchConfig, size_t cchConfig, AsyncWorkQueue **rwqueue, AsyncWorkQueue **wwqueue)
+IStorageFactory *CreateRocksDBStorageFactory(const char *path, int dbnum, const char *rgchConfig, size_t cchConfig)
 {
-    return new RocksDBStorageFactory(path, dbnum, rgchConfig, cchConfig, rwqueue, wwqueue);
+    return new RocksDBStorageFactory(path, dbnum, rgchConfig, cchConfig);
 }
 
 rocksdb::Options RocksDBStorageFactory::RocksDbOptions()
@@ -52,8 +52,8 @@ rocksdb::Options RocksDBStorageFactory::RocksDbOptions()
     return options;
 }
 
-RocksDBStorageFactory::RocksDBStorageFactory(const char *dbfile, int dbnum, const char *rgchConfig, size_t cchConfig, AsyncWorkQueue **rwqueue, AsyncWorkQueue **wwqueue)
-    : m_path(dbfile), m_rwqueue(rwqueue), m_wwqueue(wwqueue)
+RocksDBStorageFactory::RocksDBStorageFactory(const char *dbfile, int dbnum, const char *rgchConfig, size_t cchConfig)
+    : m_path(dbfile)
 {
     dbnum++; // create an extra db for metadata
     // Get the count of column families in the actual database

src/storage/rocksdbfactory.h

@@ -1,3 +1,3 @@
 #pragma once
 
-class IStorageFactory *CreateRocksDBStorageFactory(const char *path, int dbnum, const char *rgchConfig, size_t cchConfig, AsyncWorkQueue **rwqueue, AsyncWorkQueue **wwqueue);
+class IStorageFactory *CreateRocksDBStorageFactory(const char *path, int dbnum, const char *rgchConfig, size_t cchConfig);

src/tls.cpp

@@ -555,6 +555,19 @@ void tlsSetCertificateFingerprint(tls_connection* conn, X509 * cert) {
 #define ASN1_STRING_get0_data ASN1_STRING_data 
 #endif 
 
+class TCleanup {
+    std::function<void()> fn;
+
+public:
+    TCleanup(std::function<void()> fn)
+        : fn(fn)
+    {}
+
+    ~TCleanup() {
+        fn();
+    }
+};
+
 bool tlsCheckCertificateAgainstAllowlist(tls_connection* conn, std::set<sdsstring> allowlist, const char** commonName){
     if (allowlist.empty()){
         // An empty list implies acceptance of all
@@ -627,18 +640,6 @@ bool tlsCheckCertificateAgainstAllowlist(tls_connection* conn, std::set<sdsstrin
     return false;
 }
 
-bool tlsCertificateIgnoreLoadShedding(tls_connection* conn) {
-    const char* cn = "";
-    if (tlsCheckCertificateAgainstAllowlist(conn, g_pserver->tls_overload_ignorelist, &cn)) {
-        // Certificate is in exclusion list, no need to audit log
-        serverLog(LL_NOTICE, "Loadshedding: disabled for %s", conn->c.fprint);
-        return true;
-    } else {
-        serverLog(LL_NOTICE, "Loadshedding: enabled for %s", conn->c.fprint);
-        return false;
-    }
-}
-
 bool tlsCertificateRequiresAuditLogging(tls_connection* conn){
     const char* cn = "";
     if (tlsCheckCertificateAgainstAllowlist(conn, g_pserver->tls_auditlog_blocklist, &cn)) {
@@ -891,9 +892,6 @@ void tlsHandleEvent(tls_connection *conn, int mask) {
                     if (tlsCertificateRequiresAuditLogging(conn)){
                         conn->c.flags |= CONN_FLAG_AUDIT_LOGGING_REQUIRED;
                     }
-                    if (tlsCertificateIgnoreLoadShedding(conn)){
-                        conn->c.flags |= CONN_FLAG_IGNORE_OVERLOAD;
-                    }
                 }
             }
 
@@ -1071,12 +1069,6 @@ static int connTLSWrite(connection *conn_, const void *data, size_t data_len) {
 
     if (conn->c.state != CONN_STATE_CONNECTED) return -1;
     ERR_clear_error();
-
-     if (data_len > std::numeric_limits<int>::max()) {
-        // OpenSSL expects length to be 32-bit int
-        data_len = std::numeric_limits<int>::max();
-    }
-
     ret = SSL_write(conn->ssl, data, data_len);
 
     if (ret <= 0) {

tests/cluster/run.tcl

@@ -10,16 +10,6 @@ source ../../support/cluster.tcl ; # Redis Cluster client.
 set ::instances_count 20 ; # How many instances we use at max.
 set ::tlsdir "../../tls"
 
-# Check if we compiled with flash
-set status [catch {exec ../../../src/keydb-server --is-flash-enabled}]
-if {$status == 0} {
-    puts "KeyDB was built with FLASH, including FLASH tests"
-    set ::flash_enabled 1
-} else {
-    puts "KeyDB was not built with FLASH.  Excluding FLASH tests"
-    set ::flash_enabled 0
-}
-
 proc main {} {
     parse_options
     spawn_instance redis $::redis_base_port $::instances_count {

tests/cluster/tests/17-diskless-load-swapdb.tcl

@@ -14,7 +14,6 @@ test "Cluster is writable" {
     cluster_write_test 0
 }
 
-if {!$::flash_enabled} {
 test "Right to restore backups when fail to diskless load " {
     set master [Rn 0]
     set replica [Rn 1]
@@ -85,4 +84,3 @@ test "Right to restore backups when fail to diskless load " {
     assert_equal {1} [$replica get $slot0_key]
     assert_equal $slot0_key [$replica CLUSTER GETKEYSINSLOT 0 1] "POST RUN"
 }
-}

tests/integration/replication-psync-flash.tcl

@@ -10,7 +10,7 @@
 # checked for consistency.
 if {$::flash_enabled} {
     proc test_psync {descr duration backlog_size backlog_ttl delay cond mdl sdl reconnect} {
-        start_server [list tags {"repl"} overrides [list storage-provider {flash ./rocks.db.m} repl-backlog-size 64m]] {
+        start_server [list tags {"repl"} overrides [list storage-provider {flash .rocks.db.m} repl-backlog-size 64m]] {
             start_server [list tags {flash} overrides [list storage-provider {flash ./rocks.db} delete-on-evict no storage-flush-period 10]] {
 
                 set master [srv -1 client]
@@ -114,10 +114,6 @@ if {$::flash_enabled} {
 
     foreach mdl {no yes} {
         foreach sdl {disabled swapdb} {
-            if {$::flash_enabled && $sdl == "swapdb"} {
-                continue
-            }
-
             test_psync {no reconnection, just sync} 6 1000000 3600 0 {
             } $mdl $sdl 0
 

tests/integration/replication-psync-multimaster.tcl

@@ -128,10 +128,6 @@ proc test_psync {descr duration backlog_size backlog_ttl delay cond mdl sdl reco
 
 foreach mdl {no yes} {
     foreach sdl {disabled swapdb} {
-        if {$::flash_enabled && $sdl == "swapdb"} {
-            continue
-        }
-
         test_psync {no reconnection, just sync} 6 1000000 3600 0 {
         } $mdl $sdl 0
 

tests/integration/replication-psync.tcl

@@ -119,10 +119,6 @@ proc test_psync {descr duration backlog_size backlog_ttl delay cond mdl sdl reco
 
 foreach mdl {no yes} {
     foreach sdl {disabled swapdb} {
-        if {$::flash_enabled && $sdl == "swapdb"} {
-            continue
-        }
-
         test_psync {no reconnection, just sync} 6 1000000 3600 0 {
         } $mdl $sdl 0
 

tests/integration/replication.tcl

@@ -252,10 +252,6 @@ start_server {tags {"repl"}} {
 
 foreach mdl {no yes} {
     foreach sdl {disabled swapdb} {
-        if {$::flash_enabled && $sdl == "swapdb"} {
-            # swapdb not compatible with flash
-            continue
-        }
         start_server {tags {"repl"}} {
             set master [srv 0 client]
             $master config set repl-diskless-sync $mdl
@@ -387,7 +383,6 @@ start_server {tags {"repl"}} {
     }
 }
 
-if {!$::flash_enabled} {
 test {slave fails full sync and diskless load swapdb recovers it} {
     start_server {tags {"repl"}} {
         set slave [srv 0 client]
@@ -552,7 +547,6 @@ test {diskless loading short read} {
         }
     }
 }
-}
 
 # get current stime and utime metrics for a thread (since it's creation)
 proc get_cpu_metrics { statfile } {
@@ -584,13 +578,12 @@ proc compute_cpu_usage {start end} {
     return [ list $pucpu $pscpu ]
 }
 
-if {!$::flash_enabled} {
+
 # test diskless rdb pipe with multiple replicas, which may drop half way
 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]
@@ -734,8 +727,6 @@ start_server {tags {"repl"}} {
             }
         }
     }
-
-    $master config set enable-keydb-fastsync yes
 }
 
 if 0 {
@@ -820,7 +811,6 @@ test "diskless replication read pipe cleanup" {
     }
 }
 }
-}
 
 test {replicaof right after disconnection} {
     # this is a rare race condition that was reproduced sporadically by the psync2 unit.
@@ -877,10 +867,8 @@ 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/introspection.tcl

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

tests/unit/maxmemory.tcl

@@ -182,8 +182,6 @@ 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.load.test} 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 Audit Log: Able to connect while with SAN having a comprehensive list} {
+    test {TLS: 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 Log: Able to connect while with SAN having a comprehensive list with wildcards} {
+    test {TLS Audit LogTLS: 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