set version to 6.3.4

*fix for rocksdb centos compile fail

* fix alpine compile

* fix hashslot enumerate

* update alpine docker to latest version

* fix mac build, use big endian for expire prefix

* use endian.h for expires too

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 --tls
       cluster-test:
         type: cmd
         parent: make-build
         # https://github.sc-corp.net/Snapchat/img/tree/master/keydb/ubuntu-20-04
         builder_image: us.gcr.io/snapchat-build-artifacts/prod/snapchat/img/keydb/keydb-ubuntu-20-04@sha256:cf869a3f5d1de1e1d976bb906689c37b7031938eb68661b844a38c532f27248c
         command: ./runtest-cluster --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
       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; \

pkg/deb/master_changelog

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

pkg/docker/Dockerfile

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

pkg/docker/Dockerfile_Alpine

@@ -1,4 +1,4 @@
-FROM alpine:3.12
+FROM alpine:3.18
 # add our user and group first to make sure their IDs get assigned consistently, regardless of whatever dependencies get added
 RUN addgroup -S -g 1000 keydb && adduser -S -G keydb -u 999 keydb
 RUN mkdir -p /etc/keydb
@@ -68,6 +68,10 @@ RUN set -eux; \
         echo 'if [ "${1#-}" != "$1" ] || [ "${1%.conf}" != "$1" ]; then' >> /usr/local/bin/docker-entrypoint.sh; \
         echo '        set -- keydb-server "$@"' >> /usr/local/bin/docker-entrypoint.sh; \
         echo 'fi' >> /usr/local/bin/docker-entrypoint.sh; \
+        echo '# if KEYDB_PASSWORD is set, add it to the arguments' >> /usr/local/bin/docker-entrypoint.sh; \
+        echo 'if [ -n "$KEYDB_PASSWORD" ]; then' >> /usr/local/bin/docker-entrypoint.sh; \
+        echo '        set -- "$@" --requirepass "${KEYDB_PASSWORD}"' >> /usr/local/bin/docker-entrypoint.sh; \
+        echo 'fi' >> /usr/local/bin/docker-entrypoint.sh; \
         echo "# allow the container to be started with `--user`" >> /usr/local/bin/docker-entrypoint.sh; \
         echo 'if [ "$1" = "keydb-server" -a "$(id -u)" = "0" ]; then' >> /usr/local/bin/docker-entrypoint.sh; \
         echo "        find . \! -user keydb -exec chown keydb '{}' +" >> /usr/local/bin/docker-entrypoint.sh; \

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

src/SnapshotPayloadParseState.cpp

@@ -233,6 +233,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,9 @@ 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),
+    createBoolConfig("enable-keydb-fastsync", NULL, MODIFIABLE_CONFIG, g_pserver->fEnableFastSync, 0, NULL, NULL),
 
 #ifdef USE_OPENSSL
     createIntConfig("tls-port", NULL, MODIFIABLE_CONFIG, 0, 65535, g_pserver->tls_port, 0, INTEGER_CONFIG, NULL, updateTLSPort), /* TCP port. */

src/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

@@ -2528,12 +2528,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 +2557,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 +2884,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 +2988,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 +3043,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 +3115,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 +3135,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 +3159,6 @@ void redisDbPersistentData::removeAllCachedValues()
     } else {
         dictEmpty(m_pdict, nullptr);
     }
-    m_numexpires = 0;
 }
 
 void redisDbPersistentData::disableKeyCache()
@@ -3346,124 +3310,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(). */
@@ -1174,32 +1173,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 +1204,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 +1236,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 +1593,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 +1606,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 +1837,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 +1868,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 +1938,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 +2759,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 +2773,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 +2788,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/readwritelock.h

@@ -42,9 +42,9 @@ public:
             while (m_readCount > 0)
                 m_cv.wait(rm);
             if (exclusive) {
-                /* Another thread might have the write lock while we have the read lock
-                but won't be able to release it until they can acquire the read lock
-                so release the read lock and try again instead of waiting to avoid deadlock */
+                /* Another thread might have the write lock while we have the internal lock
+                but won't be able to release it until they can acquire the internal lock
+                so release the internal lock and try again instead of waiting to avoid deadlock */
                 while(!m_writeLock.try_lock())
                     m_cv.wait(rm);
             }

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;
@@ -1059,7 +1033,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();
             }
         }
@@ -1157,24 +1131,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() {
@@ -1295,7 +1256,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();
         }
     });
 
@@ -2532,7 +2495,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,10 +2520,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())
@@ -2678,14 +2637,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;
@@ -3062,7 +3014,7 @@ void readSyncBulkPayload(connection *conn) {
     }
 
     if (mi->isKeydbFastsync) {
-        if (!readFastSyncBulkPayload(conn, mi, rsi))
+        if (!readSnapshotBulkPayload(conn, mi, rsi))
             return;
     } else {
         if (!readSyncBulkPayloadRdb(conn, mi, rsi, usemark))
@@ -3798,7 +3750,7 @@ retry_connect:
             auto dt = std::chrono::system_clock::now().time_since_epoch();
             auto dtMillisecond = std::chrono::duration_cast<std::chrono::milliseconds>(dt);
             snprintf(tmpfile,sizeof(tmpfile),
-                "temp-%d.%ld.rdb",(int)dtMillisecond.count(),(long int)getpid());
+                "temp-%lld.%ld.rdb",(long long)dtMillisecond.count(),(long int)getpid());
             dfd = open(tmpfile,O_CREAT|O_WRONLY|O_EXCL,0644);
             if (dfd != -1) break;
             sleep(1);
@@ -4829,10 +4781,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 +4797,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/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
     
 }
@@ -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;
 
@@ -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;
@@ -1769,7 +1744,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,  
@@ -2232,8 +2207,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 +2546,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; 
@@ -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; }
 };
@@ -4044,7 +3924,7 @@ void updateActiveReplicaMastersFromRsi(rdbSaveInfo *rsi);
 uint64_t getMvccTstamp();
 void incrementMvccTstamp();
 
-#if __GNUC__ >= 7 && !defined(NO_DEPRECATE_FREE)
+#if __GNUC__ >= 7 && !defined(NO_DEPRECATE_FREE) && !defined(ALPINE)
  [[deprecated]]
 void *calloc(size_t count, size_t size) noexcept;
  [[deprecated]]

src/storage/rocksdb.cpp

@@ -7,6 +7,8 @@
 #include "../cluster.h"
 #include "rocksdbfactor_internal.h"
 
+template class std::basic_string<char>;
+
 static const char keyprefix[] = INTERNAL_KEY_PREFIX;
 
 rocksdb::Options DefaultRocksDBOptions();
@@ -24,8 +26,9 @@ bool FInternalKey(const char *key, size_t cch)
 
 std::string getPrefix(unsigned int hashslot)
 {
-    char *hash_char = (char *)&hashslot;
-    return std::string(hash_char, 2);
+    HASHSLOT_PREFIX_TYPE slot = HASHSLOT_PREFIX_ENDIAN((HASHSLOT_PREFIX_TYPE)hashslot);
+    char *hash_char = (char *)&slot;
+    return std::string(hash_char, HASHSLOT_PREFIX_BYTES);
 }
 
 std::string prefixKey(const char *key, size_t cchKey)
@@ -144,12 +147,13 @@ void RocksDBStorageProvider::retrieve(const char *key, size_t cchKey, callbackSi
 size_t RocksDBStorageProvider::clear()
 {
     size_t celem = count();
+    auto options = m_spdb->GetOptions(m_spcolfamily.get());
     auto status = m_spdb->DropColumnFamily(m_spcolfamily.get());
     auto strName = m_spcolfamily->GetName();
 
     rocksdb::ColumnFamilyHandle *handle = nullptr;
-    rocksdb::ColumnFamilyOptions cf_options(m_pfactory->RocksDbOptions());
-    cf_options.level_compaction_dynamic_level_bytes = true;
+    rocksdb::ColumnFamilyOptions cf_options(options);
+
     m_spdb->CreateColumnFamily(cf_options, strName, &handle);
     m_spcolfamily = std::shared_ptr<rocksdb::ColumnFamilyHandle>(handle);
 
@@ -183,7 +187,7 @@ bool RocksDBStorageProvider::enumerate(callback fn) const
         if (FInternalKey(it->key().data(), it->key().size()))
             continue;
         ++count;
-        bool fContinue = fn(it->key().data()+2, it->key().size()-2, it->value().data(), it->value().size());
+        bool fContinue = fn(it->key().data()+HASHSLOT_PREFIX_BYTES, it->key().size()-HASHSLOT_PREFIX_BYTES, it->value().data(), it->value().size());
         if (!fContinue)
             break;
     }
@@ -202,26 +206,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); 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 (HASHSLOT_PREFIX_RECOVER(*(HASHSLOT_PREFIX_TYPE *)it->key().data()) != hashslot)
             break;
         ++count;
-        bool fContinue = fn(it->key().data()+2, it->key().size()-2, it->value().data(), it->value().size());
-        if (!fContinue) {
-            full_iter = false;
+        bool fContinue = fn(it->key().data()+HASHSLOT_PREFIX_BYTES, it->key().size()-HASHSLOT_PREFIX_BYTES, it->value().data(), it->value().size());
+        if (!fContinue)
             break;
-        }
     }
+    bool full_iter = !it->Valid() || (HASHSLOT_PREFIX_RECOVER(*(HASHSLOT_PREFIX_TYPE *)it->key().data()) != hashslot);
     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;
 }
 
@@ -229,7 +230,8 @@ void RocksDBStorageProvider::setExpire(const char *key, size_t cchKey, long long
 {
     rocksdb::Status status;
     std::unique_lock<fastlock> l(m_lock);
-    std::string prefix((const char *)&expire,sizeof(long long));
+    long long beExpire = htobe64(expire);
+    std::string prefix((const char *)&beExpire,sizeof(long long));
     std::string strKey(key, cchKey);
     if (m_spbatch != nullptr)
         status = m_spbatch->Put(m_spexpirecolfamily.get(), rocksdb::Slice(prefix + strKey), rocksdb::Slice(strKey));
@@ -243,7 +245,8 @@ void RocksDBStorageProvider::removeExpire(const char *key, size_t cchKey, long l
 {
     rocksdb::Status status;
     std::unique_lock<fastlock> l(m_lock);
-    std::string prefix((const char *)&expire,sizeof(long long));
+    long long beExpire = htobe64(expire);
+    std::string prefix((const char *)&beExpire,sizeof(long long));
     std::string strKey(key, cchKey);
     std::string fullKey = prefix + strKey;
     if (!FExpireExists(fullKey))
@@ -280,7 +283,7 @@ std::vector<std::string> RocksDBStorageProvider::getEvictionCandidates(unsigned
         for (it->Seek(randomHashSlot()); it->Valid() && result.size() < count; it->Next()) {
             if (FInternalKey(it->key().data(), it->key().size()))
                 continue;
-            result.emplace_back(it->key().data() + 2, it->key().size() - 2);
+            result.emplace_back(it->key().data() + HASHSLOT_PREFIX_BYTES, it->key().size() - HASHSLOT_PREFIX_BYTES);
         }
     } else {
         std::unique_ptr<rocksdb::Iterator> it = std::unique_ptr<rocksdb::Iterator>(m_spdb->NewIterator(ReadOptions(), m_spexpirecolfamily.get()));
@@ -337,31 +340,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 +364,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

@@ -4,9 +4,21 @@
 #include "../IStorage.h"
 #include <rocksdb/db.h>
 #include <rocksdb/utilities/write_batch_with_index.h>
+#ifdef __APPLE__
+#include <libkern/OSByteOrder.h>
+#define htole16 OSSwapHostToLittleInt16
+#define le16toh OSSwapLittleToHostInt16
+#define htobe64 OSSwapHostToBigInt64
+#else
+#include <endian.h>
+#endif
 #include "../fastlock.h"
 
 #define INTERNAL_KEY_PREFIX "\x00\x04\x03\x00\x05\x02\x04"
+#define HASHSLOT_PREFIX_TYPE uint16_t
+#define HASHSLOT_PREFIX_BYTES sizeof(HASHSLOT_PREFIX_TYPE)
+#define HASHSLOT_PREFIX_ENDIAN htole16
+#define HASHSLOT_PREFIX_RECOVER le16toh
 static const char count_key[] = INTERNAL_KEY_PREFIX "__keydb__count\1";
 static const char version_key[] = INTERNAL_KEY_PREFIX "__keydb__version\1";
 static const char meta_key[] = INTERNAL_KEY_PREFIX "__keydb__metadata\1";
@@ -32,10 +44,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 +57,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
@@ -72,7 +72,7 @@ RocksDBStorageFactory::RocksDBStorageFactory(const char *dbfile, int dbnum, cons
     rocksdb::DB *db = nullptr;
 
     auto options = RocksDbOptions();
-    options.prefix_extractor.reset(rocksdb::NewFixedPrefixTransform(2));
+    options.prefix_extractor.reset(rocksdb::NewFixedPrefixTransform(HASHSLOT_PREFIX_BYTES));
 
     for (int idb = 0; idb < dbnum; ++idb)
     {
@@ -196,7 +196,7 @@ IStorage *RocksDBStorageFactory::create(int db, key_load_iterator iter, void *pr
                 printf("\tDatabase %d was not shutdown cleanly, recomputing metrics\n", db);
             fFirstRealKey = false;
             if (iter != nullptr)
-                iter(it->key().data()+2, it->key().size()-2, privdata);
+                iter(it->key().data()+HASHSLOT_PREFIX_BYTES, it->key().size()-HASHSLOT_PREFIX_BYTES, privdata);
             ++count;
         }
     }

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

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

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

src/version.h

@@ -1,5 +1,5 @@
-#define KEYDB_REAL_VERSION "255.255.255"
-#define KEYDB_VERSION_NUM 0x00ffffff
+#define KEYDB_REAL_VERSION "6.3.4"
+#define KEYDB_VERSION_NUM 0x00060304
 extern const char *KEYDB_SET_VERSION;   // Unlike real version, this can be overriden by the config
 
 enum VersionCompareResult

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,7 +578,7 @@ 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]
@@ -820,7 +814,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.

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