don't do if statement on every key, use stack object for module notify

* add s3 config

* missing include

* ignore s3 config in config test

* Update introspection.tcl

* Update config.cpp

.github/workflows/ci.yml

@@ -21,10 +21,10 @@ jobs:
     - name: test-tls
       run: |
         sudo apt-get -y install tcl tcl-tls
-        ./runtest --clients 2 --verbose --tls
+        ./runtest --clients 1 --verbose --tls --config server-threads 3
     - name: cluster-test
       run: |
-        ./runtest-cluster --tls
+        ./runtest-cluster --tls --config server-threads 3
     - name: sentinel test
       run: |
           ./runtest-sentinel

README.md

@@ -4,7 +4,7 @@
 
 ##### KeyDB is now a part of Snap Inc! Check out the announcement [here](https://docs.keydb.dev/news/2022/05/12/keydb-joins-snap) 
 
-##### [Release v6.3.0](https://github.com/EQ-Alpha/KeyDB/releases/tag/v6.3.0) is here with major improvements as we consolodate our Open Source and Enterprise offerings into a single BSD-3 licensed project. See our [roadmap](https://docs.keydb.dev/docs/coming-soon) for details. 
+##### [Release v6.3.0](https://github.com/EQ-Alpha/KeyDB/releases/tag/v6.3.0) is here with major improvements as we consolidate our Open Source and Enterprise offerings into a single BSD-3 licensed project. See our [roadmap](https://docs.keydb.dev/docs/coming-soon) for details. 
 
 ##### Want to extend KeyDB with Javascript?  Try [ModJS](https://github.com/JohnSully/ModJS)
 
@@ -37,7 +37,7 @@ Because of this difference of opinion features which are right for KeyDB may not
 Project Support
 -------------------
 
-The KeyDB team maintains this project as part of Snap Inc. KeyDB is used by Snap as part of its caching infrastructure and is fully open sourced. There is no separate commercial product and no paid support options available. We really value collaborating with the open source community and welcome PRs, bug reports, and open discussion. For community support or to get involved further with the project check out our community support options [here](https://docs.keydb.dev/docs/support) (slack, forum, meetup, github issues). Our team monitors these channlels regularly.
+The KeyDB team maintains this project as part of Snap Inc. KeyDB is used by Snap as part of its caching infrastructure and is fully open sourced. There is no separate commercial product and no paid support options available. We really value collaborating with the open source community and welcome PRs, bug reports, and open discussion. For community support or to get involved further with the project check out our community support options [here](https://docs.keydb.dev/docs/support) (slack, forum, meetup, github issues). Our team monitors these channels regularly.
 
 
 Additional Resources
@@ -104,6 +104,12 @@ Avoid forwarding RREPLAY messages to other masters? WARNING: This setting is dan
 If you would like KeyDB to dump and load directly to AWS S3 this option specifies the bucket.  Using this option with the traditional RDB options will result in KeyDB backing up twice to both locations.  If both are specified KeyDB will first attempt to load from the local dump file and if that fails load from S3.  This requires the AWS CLI tools to be installed and configured which are used under the hood to transfer the data.
 
 
+```
+storage-provider flash /path/to/flash
+```
+If you would like to use KeyDB FLASH storage, specify the storage medium followed by the directory path on your local SSD volume. Note that this feature is still considered experimental and should be used with discretion. See [FLASH Documentation](https://docs.keydb.dev/docs/flash) for more details on configuration and setting up your FLASH volume. 
+
+
 Building KeyDB
 --------------
 
@@ -135,9 +141,9 @@ To append a suffix to KeyDB program names, use:
 ***Note that the following dependencies may be needed: 
     % sudo apt-get install autoconf autotools-dev libnuma-dev libtool
 
-To buik=ld with TLS support, use:
+KeyDB by default is built with TLS enabled. To build without TLS support, use:
 
-    % make BUILD_TLS=yes
+    % make BUILD_TLS=no
 
 Running the tests with TLS enabled (you will need `tcl-tls`
 installed):
@@ -145,6 +151,12 @@ installed):
     % ./utils/gen-test-certs.sh
     % ./runtest --tls
 
+To build with KeyDB FLASH support, use:
+
+    % make ENABLE_FLASH=yes
+
+***Note that the KeyDB FLASH feature is considered experimental (beta) and should used with discretion
+
 
 Fixing build problems with dependencies or cached build options
 ---------

deps/Makefile

@@ -95,7 +95,7 @@ JEMALLOC_LDFLAGS= $(LDFLAGS)
 
 jemalloc: .make-prerequisites
 	@printf '%b %b\n' $(MAKECOLOR)MAKE$(ENDCOLOR) $(BINCOLOR)$@$(ENDCOLOR)
-	cd jemalloc && ./configure --with-version=5.2.1-0-g0 --with-lg-quantum=3 --with-jemalloc-prefix=je_ --disable-cxx CFLAGS="$(JEMALLOC_CFLAGS)" LDFLAGS="$(JEMALLOC_LDFLAGS)"
+	cd jemalloc && ./configure --with-version=5.2.1-0-g0 --with-lg-quantum=3 --disable-cxx CFLAGS="$(JEMALLOC_CFLAGS)" LDFLAGS="$(JEMALLOC_LDFLAGS)"
 	cd jemalloc && $(MAKE) CFLAGS="$(JEMALLOC_CFLAGS)" LDFLAGS="$(JEMALLOC_LDFLAGS)" lib/libjemalloc.a
 
 .PHONY: jemalloc

keydb.conf

@@ -2073,4 +2073,18 @@ replica-weighting-factor 2
 #
 # By default this is enabled
 #
-active-client-balancing yes
+active-client-balancing yes
+
+# Enable FLASH support (Experimental Feature)
+# storage-provider flash /path/to/flash/db
+
+# Blob support is a way to store very large objects (>200MB) on disk
+# The files are automatically cleaned up when KeyDB exits and are only
+# for temporary use.  This helps reduce memory pressure for very large
+# data items at the cost of some performance.
+#
+# By default this config is disable.  When enabled the disk associated
+# with KeyDB's working directory will be used.  If there is insufficient
+# disk space or any other I/O error KeyDB will instead use memory.
+#
+# blob-support false

pkg/deb/conf/keydb.conf

@@ -2063,3 +2063,17 @@ server-threads 2
 #
 # By default KeyDB sets this to 2.
 replica-weighting-factor 2
+
+# Enable FLASH support (Experimental Feature)
+# storage-provider flash /path/to/flash/db
+
+# Blob support is a way to store very large objects (>200MB) on disk
+# The files are automatically cleaned up when KeyDB exits and are only
+# for temporary use.  This helps reduce memory pressure for very large
+# data items at the cost of some performance.
+# 
+# By default this config is disable.  When enabled the disk associated
+# with KeyDB's working directory will be used.  If there is insufficient
+# disk space or any other I/O error KeyDB will instead use memory.
+# 
+# blob-support false

pkg/deb/deb-buildsource.sh

@@ -19,6 +19,11 @@ elif [ "$distributor" == "Ubuntu" ]; then
 fi
 codename=$(lsb_release --codename --short)
 date=$(date +%a," "%d" "%b" "%Y" "%T)
+
+# overwrite debian bookworm version until updated
+if [ $codename == "bookworm" ]; then
+    distname=+deb12u1
+fi
 pkg_name=keydb-$majorv:$version$distname
 
 # create build tree

pkg/deb/debian/rules

@@ -4,6 +4,7 @@ include /usr/share/dpkg/buildflags.mk
 
 export BUILD_TLS=yes
 export USE_SYSTEMD=yes
+export ENABLE_FLASH=yes
 export CFLAGS CPPFLAGS LDFLAGS
 export DEB_BUILD_MAINT_OPTIONS = hardening=+all
 export DEB_LDFLAGS_MAINT_APPEND = -ldl -latomic $(LUA_LDFLAGS)

pkg/deb/debian_dh9/rules

@@ -10,6 +10,7 @@ include /usr/share/dpkg/buildflags.mk
 
 export BUILD_TLS=yes
 export USE_SYSTEMD=yes
+export ENABLE_FLASH=yes
 export CFLAGS CPPFLAGS LDFLAGS
 export DEB_BUILD_MAINT_OPTIONS = hardening=+all
 export DEB_LDFLAGS_MAINT_APPEND = -ldl -latomic $(LUA_LDFLAGS)

pkg/rpm/keydb_build/keydb_rpm/etc/keydb/keydb.conf

@@ -2063,3 +2063,17 @@ server-threads 2
 #
 # By default KeyDB sets this to 2.
 replica-weighting-factor 2
+
+# Enable FLASH support (Experimental Feature)
+# storage-provider flash /path/to/flash/db
+
+# Blob support is a way to store very large objects (>200MB) on disk
+# The files are automatically cleaned up when KeyDB exits and are only
+# for temporary use.  This helps reduce memory pressure for very large
+# data items at the cost of some performance.
+# 
+# By default this config is disable.  When enabled the disk associated
+# with KeyDB's working directory will be used.  If there is insufficient
+# disk space or any other I/O error KeyDB will instead use memory.
+# 
+# blob-support false

runtest-moduleapi

@@ -37,5 +37,6 @@ $TCLSH tests/test_helper.tcl \
 --single unit/moduleapi/hash \
 --single unit/moduleapi/zset \
 --single unit/moduleapi/stream \
+--single unit/moduleapi/load \
 --config server-threads 3 \
 "${@}"

src/Makefile

@@ -20,12 +20,12 @@ release_hdr := $(shell sh -c './mkreleasehdr.sh')
 uname_S := $(shell sh -c 'uname -s 2>/dev/null || echo not')
 uname_M := $(shell sh -c 'uname -m 2>/dev/null || echo not')
 OPTIMIZATION?=-O2 -flto
-DEPENDENCY_TARGETS=hiredis linenoise lua hdr_histogram rocksdb
+DEPENDENCY_TARGETS=hiredis linenoise lua hdr_histogram
 NODEPS:=clean distclean
 
 # Default settings
 STD=-pedantic -DREDIS_STATIC=''
-CXX_STD=-std=c++14 -pedantic -fno-rtti -D__STDC_FORMAT_MACROS  
+CXX_STD=-std=c++17 -pedantic -fno-rtti -D__STDC_FORMAT_MACROS  
 ifneq (,$(findstring clang,$(CC)))
   STD+=-Wno-c11-extensions
 else
@@ -376,7 +376,7 @@ endif
 
 REDIS_SERVER_NAME=keydb-server$(PROG_SUFFIX)
 REDIS_SENTINEL_NAME=keydb-sentinel$(PROG_SUFFIX)
-REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o t_nhash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crcspeed.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o acl.o storage.o rdb-s3.o fastlock.o new.o tracking.o cron.o connection.o tls.o sha256.o motd_server.o timeout.o setcpuaffinity.o AsyncWorkQueue.o snapshot.o storage/rocksdb.o storage/rocksdbfactory.o storage/teststorageprovider.o keydbutils.o StorageCache.o monotonic.o cli_common.o mt19937-64.o $(ASM_OBJ)
+REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o t_nhash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o scripting.o bio.o rio.o rand.o memtest.o crcspeed.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o acl.o storage.o rdb-s3.o fastlock.o new.o tracking.o cron.o connection.o tls.o sha256.o motd_server.o timeout.o setcpuaffinity.o AsyncWorkQueue.o snapshot.o storage/teststorageprovider.o keydbutils.o StorageCache.o monotonic.o cli_common.o mt19937-64.o $(ASM_OBJ) $(STORAGE_OBJ)
 KEYDB_SERVER_OBJ=SnapshotPayloadParseState.o
 REDIS_CLI_NAME=keydb-cli$(PROG_SUFFIX)
 REDIS_CLI_OBJ=anet.o adlist.o dict.o redis-cli.o redis-cli-cpphelper.o zmalloc.o release.o anet.o ae.o crcspeed.o crc64.o siphash.o crc16.o storage-lite.o fastlock.o motd_client.o monotonic.o cli_common.o mt19937-64.o $(ASM_OBJ)
@@ -437,7 +437,7 @@ endif
 
 # keydb-server
 $(REDIS_SERVER_NAME): $(REDIS_SERVER_OBJ) $(KEYDB_SERVER_OBJ)
-	$(REDIS_LD) -o $@ $^ ../deps/hiredis/libhiredis.a ../deps/lua/src/liblua.a ../deps/rocksdb/librocksdb.a $(FINAL_LIBS)
+	$(REDIS_LD) -o $@ $^ ../deps/hiredis/libhiredis.a ../deps/lua/src/liblua.a $(FINAL_LIBS)
 
 # keydb-sentinel
 $(REDIS_SENTINEL_NAME): $(REDIS_SERVER_NAME)

src/SnapshotPayloadParseState.cpp

@@ -140,7 +140,7 @@ void SnapshotPayloadParseState::flushQueuedKeys() {
     auto &insertsInFlightTmp = insertsInFlight; // C++ GRRRRRRRRRRRRRRRR, we don't want to capute "this" because that's dangerous
     if (current_database < cserver.dbnum) {
         g_pserver->asyncworkqueue->AddWorkFunction([idb, vecqueuedKeys = std::move(this->vecqueuedKeys), vecqueuedKeysCb = std::move(this->vecqueuedKeysCb), vecqueuedVals = std::move(this->vecqueuedVals), vecqueuedValsCb = std::move(this->vecqueuedValsCb), &insertsInFlightTmp, pallocator = m_spallocator.release()]() mutable {
-            g_pserver->db[idb]->bulkStorageInsert(vecqueuedKeys.data(), vecqueuedKeysCb.data(), vecqueuedVals.data(), vecqueuedValsCb.data(), vecqueuedKeys.size());
+            g_pserver->db[idb]->bulkDirectStorageInsert(vecqueuedKeys.data(), vecqueuedKeysCb.data(), vecqueuedVals.data(), vecqueuedValsCb.data(), vecqueuedKeys.size());
             --insertsInFlightTmp;
             delete pallocator;
         });

src/ae.cpp

@@ -716,11 +716,6 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags)
             }
         }
 
-        if (eventLoop->flags & AE_DONT_WAIT) {
-            tv.tv_sec = tv.tv_usec = 0;
-            tvp = &tv;
-        }
-
         if (eventLoop->beforesleep != NULL && flags & AE_CALL_BEFORE_SLEEP) {
             std::unique_lock<decltype(g_lock)> ulock(g_lock, std::defer_lock);
             if (!(eventLoop->beforesleepFlags & AE_SLEEP_THREADSAFE)) {
@@ -731,6 +726,11 @@ int aeProcessEvents(aeEventLoop *eventLoop, int flags)
             eventLoop->beforesleep(eventLoop);
         }
 
+        if (eventLoop->flags & AE_DONT_WAIT) {
+            tv.tv_sec = tv.tv_usec = 0;
+            tvp = &tv;
+        }
+
         /* Call the multiplexing API, will return only on timeout or when
          * some event fires. */
         numevents = aeApiPoll(eventLoop, tvp);
@@ -859,6 +859,11 @@ void aeReleaseForkLock()
     g_forkLock.downgradeWrite();
 }
 
+void aeForkLockInChild()
+{
+    g_forkLock.setNotify(false);
+}
+
 int aeThreadOwnsLock()
 {
     return fOwnLockOverride || g_lock.fOwnLock();

src/ae.h

@@ -171,6 +171,7 @@ int aeTryAcquireLock(int fWeak);
 void aeThreadOffline();
 void aeReleaseLock();
 void aeReleaseForkLock();
+void aeForkLockInChild();
 int aeThreadOwnsLock();
 void aeSetThreadOwnsLockOverride(int fOverride);
 int aeLockContested(int threshold);

src/anet.c

@@ -441,7 +441,7 @@ static int _anetTcpServer(char *err, int port, const char *bindaddr, int af, int
     char _port[6];  /* strlen("65535") */
     struct addrinfo hints, *servinfo, *p;
 
-    snprintf(_port,6,"%d",port);
+    snprintf(_port,sizeof(_port),"%d",port);
     memset(&hints,0,sizeof(hints));
     hints.ai_family = af;
     hints.ai_socktype = SOCK_STREAM;

src/aof.cpp

@@ -285,7 +285,7 @@ int startAppendOnly(void) {
             strerror(errno));
         return C_ERR;
     }
-    if (hasActiveChildProcess() && g_pserver->child_type != CHILD_TYPE_AOF) {
+    if (hasActiveChildProcessOrBGSave() && g_pserver->child_type != CHILD_TYPE_AOF) {
         g_pserver->aof_rewrite_scheduled = 1;
         serverLog(LL_WARNING,"AOF was enabled but there is already another background operation. An AOF background was scheduled to start when possible.");
     } else {
@@ -438,7 +438,7 @@ void flushAppendOnlyFile(int force) {
      * useful for graphing / monitoring purposes. */
     if (sync_in_progress) {
         latencyAddSampleIfNeeded("aof-write-pending-fsync",latency);
-    } else if (hasActiveChildProcess()) {
+    } else if (hasActiveChildProcessOrBGSave()) {
         latencyAddSampleIfNeeded("aof-write-active-child",latency);
     } else {
         latencyAddSampleIfNeeded("aof-write-alone",latency);
@@ -535,7 +535,7 @@ void flushAppendOnlyFile(int force) {
 try_fsync:
     /* Don't fsync if no-appendfsync-on-rewrite is set to yes and there are
      * children doing I/O in the background. */
-    if (g_pserver->aof_no_fsync_on_rewrite && hasActiveChildProcess())
+    if (g_pserver->aof_no_fsync_on_rewrite && hasActiveChildProcessOrBGSave())
         return;
 
     /* Perform the fsync if needed. */
@@ -752,7 +752,7 @@ void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int a
      * accumulate the differences between the child DB and the current one
      * in a buffer, so that when the child process will do its work we
      * can append the differences to the new append only file. */
-    if (g_pserver->child_type == CHILD_TYPE_AOF)
+    if (hasActiveChildProcess() && g_pserver->child_type == CHILD_TYPE_AOF)
         aofRewriteBufferAppend((unsigned char*)buf,sdslen(buf));
 
     sdsfree(buf);
@@ -1658,7 +1658,7 @@ int rewriteAppendOnlyFile(char *filename) {
 { // BEGIN GOTO SCOPED VARIABLES
     /* Note that we have to use a different temp name here compared to the
      * one used by rewriteAppendOnlyFileBackground() function. */
-    snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) getpid());
+    snprintf(tmpfile,sizeof(tmpfile),"temp-rewriteaof-%d.aof", (int) getpid());
     fp = fopen(tmpfile,"w");
     if (!fp) {
         serverLog(LL_WARNING, "Opening the temp file for AOF rewrite in rewriteAppendOnlyFile(): %s", strerror(errno));
@@ -1887,7 +1887,7 @@ void aofClosePipes(void) {
 int rewriteAppendOnlyFileBackground(void) {
     pid_t childpid;
 
-    if (hasActiveChildProcess()) return C_ERR;
+    if (hasActiveChildProcessOrBGSave()) return C_ERR;
     if (aofCreatePipes() != C_OK) return C_ERR;
     if ((childpid = redisFork(CHILD_TYPE_AOF)) == 0) {
         char tmpfile[256];
@@ -1895,7 +1895,7 @@ int rewriteAppendOnlyFileBackground(void) {
         /* Child */
         redisSetProcTitle("keydb-aof-rewrite");
         redisSetCpuAffinity(g_pserver->aof_rewrite_cpulist);
-        snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) getpid());
+        snprintf(tmpfile,sizeof(tmpfile),"temp-rewriteaof-bg-%d.aof", (int) getpid());
         if (rewriteAppendOnlyFile(tmpfile) == C_OK) {
             sendChildCowInfo(CHILD_INFO_TYPE_AOF_COW_SIZE, "AOF rewrite");
             exitFromChild(0);
@@ -1930,7 +1930,7 @@ int rewriteAppendOnlyFileBackground(void) {
 void bgrewriteaofCommand(client *c) {
     if (g_pserver->child_type == CHILD_TYPE_AOF) {
         addReplyError(c,"Background append only file rewriting already in progress");
-    } else if (hasActiveChildProcess()) {
+    } else if (hasActiveChildProcessOrBGSave()) {
         g_pserver->aof_rewrite_scheduled = 1;
         addReplyStatus(c,"Background append only file rewriting scheduled");
     } else if (rewriteAppendOnlyFileBackground() == C_OK) {
@@ -1944,10 +1944,10 @@ void bgrewriteaofCommand(client *c) {
 void aofRemoveTempFile(pid_t childpid) {
     char tmpfile[256];
 
-    snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof", (int) childpid);
+    snprintf(tmpfile,sizeof(tmpfile),"temp-rewriteaof-bg-%d.aof", (int) childpid);
     bg_unlink(tmpfile);
 
-    snprintf(tmpfile,256,"temp-rewriteaof-%d.aof", (int) childpid);
+    snprintf(tmpfile,sizeof(tmpfile),"temp-rewriteaof-%d.aof", (int) childpid);
     bg_unlink(tmpfile);
 }
 
@@ -1985,7 +1985,7 @@ void backgroundRewriteDoneHandler(int exitcode, int bysignal) {
         /* Flush the differences accumulated by the parent to the
          * rewritten AOF. */
         latencyStartMonitor(latency);
-        snprintf(tmpfile,256,"temp-rewriteaof-bg-%d.aof",
+        snprintf(tmpfile,sizeof(tmpfile),"temp-rewriteaof-bg-%d.aof",
             (int)g_pserver->child_pid);
         newfd = open(tmpfile,O_WRONLY|O_APPEND);
         if (newfd == -1) {

src/bitops.cpp

@@ -322,7 +322,7 @@ handle_wrap:
     return 1;
 }
 
-int checkSignedBitfieldOverflow(int64_t value, int64_t incr, uint64_t bits, int owtype, int64_t *limit) {
+int checkSignedBitfieldOverflow(int64_t value, int64_t incr, int bits, int owtype, int64_t *limit) {
     int64_t max = (bits == 64) ? INT64_MAX : (((int64_t)1<<(bits-1))-1);
     int64_t min = (-max)-1;
 

src/childinfo.cpp

@@ -42,6 +42,7 @@ typedef struct {
  * RDB / AOF saving process from the child to the parent (for instance
  * the amount of copy on write memory used) */
 void openChildInfoPipe(void) {
+    serverAssert(g_pserver->child_info_pipe[0] == -1);
     if (pipe(g_pserver->child_info_pipe) == -1) {
         /* On error our two file descriptors should be still set to -1,
          * but we call anyway closeChildInfoPipe() since can't hurt. */

src/cluster.cpp

@@ -5107,7 +5107,7 @@ void createDumpPayload(rio *payload, robj_roptr o, robj *key) {
     serverAssert(rdbSaveObject(payload,o,key));
     char szT[32];
     uint64_t mvcc = mvccFromObj(o);
-    snprintf(szT, 32, "%" PRIu64, mvcc);
+    snprintf(szT, sizeof(szT), "%" PRIu64, mvcc);
     serverAssert(rdbSaveAuxFieldStrStr(payload,"mvcc-tstamp", szT) != -1);
 
     /* Write the footer, this is how it looks like:

src/config.cpp

@@ -35,11 +35,13 @@
 
 #include <fcntl.h>
 #include <sys/stat.h>
+#include <sys/wait.h>
 #ifdef __linux__
 #include <sys/sysinfo.h>
 #endif
 
 const char *KEYDB_SET_VERSION = KEYDB_REAL_VERSION;
+size_t g_semiOrderedSetTargetBucketSize = 0;    // Its a header only class so nowhere else for this to go
 
 /*-----------------------------------------------------------------------------
  * Config file name-value maps.
@@ -2455,6 +2457,32 @@ static int isValidAOFfilename(char *val, const char **err) {
     return 1;
 }
 
+static int isValidS3Bucket(char *s3bucket, const char **err) {
+    int status = EXIT_FAILURE;
+    pid_t pid = fork();
+    if (pid < 0)
+    {
+        *err = "couldn't fork to call aws cli";
+        return 0;
+    }
+
+    if (pid == 0)
+    {
+        execlp("aws", "aws", "s3", "ls", s3bucket);
+        exit(EXIT_FAILURE);
+    }
+    else 
+    {
+        waitpid(pid, &status, 0);
+    }
+
+    if (status != EXIT_SUCCESS) {
+        *err = "could not access s3 bucket";
+        return 0;
+    }
+    return 1;
+}
+
 /* Validate specified string is a valid proc-title-template */
 static int isValidProcTitleTemplate(char *val, const char **err) {
     if (!validateProcTitleTemplate(val)) {
@@ -2587,7 +2615,7 @@ static int updateMaxclients(long long val, long long prev, const char **err) {
         adjustOpenFilesLimit();
         if (g_pserver->maxclients != val) {
             static char msg[128];
-            snprintf(msg, 128, "The operating system is not able to handle the specified number of clients, try with %d", g_pserver->maxclients);
+            snprintf(msg, sizeof(msg), "The operating system is not able to handle the specified number of clients, try with %d", g_pserver->maxclients);
             *err = msg;
             if (g_pserver->maxclients > prev) {
                 g_pserver->maxclients = prev;
@@ -2626,7 +2654,7 @@ static int updateMaxclients(long long val, long long prev, const char **err) {
 
                 if (res != AE_OK){
                     static char msg[128];
-                    snprintf(msg, 128,"Failed to post the request to change setsize for Thread %d", iel);
+                    snprintf(msg, sizeof(msg),"Failed to post the request to change setsize for Thread %d", iel);
                     *err = msg;
                     return 0;
                 }
@@ -2802,6 +2830,7 @@ standardConfig configs[] = {
     createStringConfig("cluster-announce-ip", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->cluster_announce_ip, NULL, NULL, NULL),
     createStringConfig("syslog-ident", NULL, IMMUTABLE_CONFIG, ALLOW_EMPTY_STRING, g_pserver->syslog_ident, "redis", NULL, NULL),
     createStringConfig("dbfilename", NULL, MODIFIABLE_CONFIG, ALLOW_EMPTY_STRING, g_pserver->rdb_filename, CONFIG_DEFAULT_RDB_FILENAME, isValidDBfilename, NULL),
+    createStringConfig("db-s3-object", NULL, MODIFIABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->rdb_s3bucketpath, NULL, isValidS3Bucket, NULL),
     createStringConfig("appendfilename", NULL, IMMUTABLE_CONFIG, ALLOW_EMPTY_STRING, g_pserver->aof_filename, "appendonly.aof", isValidAOFfilename, NULL),
     createStringConfig("server_cpulist", NULL, IMMUTABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->server_cpulist, NULL, NULL, NULL),
     createStringConfig("bio_cpulist", NULL, IMMUTABLE_CONFIG, EMPTY_STRING_IS_NULL, g_pserver->bio_cpulist, NULL, NULL, NULL),
@@ -2910,6 +2939,7 @@ standardConfig configs[] = {
     createBoolConfig("allow-write-during-load", NULL, MODIFIABLE_CONFIG, g_pserver->fWriteDuringActiveLoad, 0, NULL, NULL),
     createBoolConfig("force-backlog-disk-reserve", NULL, MODIFIABLE_CONFIG, cserver.force_backlog_disk, 0, NULL, NULL),
     createBoolConfig("soft-shutdown", NULL, MODIFIABLE_CONFIG, g_pserver->config_soft_shutdown, 0, NULL, NULL),
+    createSizeTConfig("semi-ordered-set-bucket-size", NULL, MODIFIABLE_CONFIG, 0, 1024, g_semiOrderedSetTargetBucketSize, 0, INTEGER_CONFIG, NULL, NULL),
 
 #ifdef USE_OPENSSL
     createIntConfig("tls-port", NULL, MODIFIABLE_CONFIG, 0, 65535, g_pserver->tls_port, 0, INTEGER_CONFIG, NULL, updateTLSPort), /* TCP port. */

src/db.cpp

@@ -93,7 +93,7 @@ static void lookupKeyUpdateObj(robj *val, int flags)
     /* Update the access time for the ageing algorithm.
      * Don't do it if we have a saving child, as this will trigger
      * a copy on write madness. */
-    if (!hasActiveChildProcess() && !(flags & LOOKUP_NOTOUCH))
+    if (!hasActiveChildProcessOrBGSave() && !(flags & LOOKUP_NOTOUCH))
     {
         if (g_pserver->maxmemory_policy & MAXMEMORY_FLAG_LFU) {
             updateLFU(val);
@@ -1059,7 +1059,7 @@ void keysCommand(client *c) {
     sds pattern = szFromObj(c->argv[1]);
 
     const redisDbPersistentDataSnapshot *snapshot = nullptr;
-    if (!(c->flags & (CLIENT_MULTI | CLIENT_BLOCKED)))
+    if (!(c->flags & (CLIENT_MULTI | CLIENT_BLOCKED | CLIENT_DENY_BLOCKING)))
         snapshot = c->db->createSnapshot(c->mvccCheckpoint, true /* fOptional */);
     if (snapshot != nullptr)
     {
@@ -2607,6 +2607,17 @@ void clusterStorageLoadCallback(const char *rgchkey, size_t cch, void *)
     slotToKeyUpdateKeyCore(rgchkey, cch, true /*add*/);
 }
 
+void moduleLoadCallback(const char * rgchKey, size_t, void *data) {
+    redisObjectStack keyobj;
+    initStaticStringObject(keyobj, const_cast<char *>(rgchKey));
+    moduleNotifyKeyspaceEvent(NOTIFY_LOADED, "loaded", &keyobj, *(int *)data);
+}
+
+void moduleClusterLoadCallback(const char * rgchKey, size_t cchKey, void *data) {
+    clusterStorageLoadCallback(rgchKey, cchKey, data);
+    moduleLoadCallback(rgchKey, cchKey, data);
+}
+
 void redisDb::initialize(int id)
 {
     redisDbPersistentData::initialize();
@@ -2625,8 +2636,8 @@ void redisDb::storageProviderInitialize()
 {
     if (g_pserver->m_pstorageFactory != nullptr)
     {
-        IStorageFactory::key_load_iterator itr = (g_pserver->cluster_enabled) ? clusterStorageLoadCallback : nullptr;
-        this->setStorageProvider(StorageCache::create(g_pserver->m_pstorageFactory, id, itr, nullptr));
+        IStorageFactory::key_load_iterator itr = g_pserver->cluster_enabled ? moduleClusterLoadCallback : moduleLoadCallback;
+        this->setStorageProvider(StorageCache::create(g_pserver->m_pstorageFactory, id, itr, &id));
     }
 }
 
@@ -2994,8 +3005,16 @@ void redisDbPersistentData::processChangesAsync(std::atomic<int> &pendingJobs)
     });
 }
 
-void redisDbPersistentData::bulkStorageInsert(char **rgKeys, size_t *rgcbKeys, char **rgVals, size_t *rgcbVals, size_t celem)
+/* This function is to bulk insert directly to storage provider bypassing in memory, assumes rgKeys and rgVals are not sds strings */
+void redisDbPersistentData::bulkDirectStorageInsert(char **rgKeys, size_t *rgcbKeys, char **rgVals, size_t *rgcbVals, size_t celem)
 {
+    if (g_pserver->cluster_enabled) {
+        aeAcquireLock();
+        for (size_t i = 0; i < celem; i++) {
+            slotToKeyUpdateKeyCore(rgKeys[i], rgcbKeys[i], 1);
+        }
+        aeReleaseLock();
+    }
     m_spstorage->bulkInsert(rgKeys, rgcbKeys, rgVals, rgcbVals, celem);
 }
 
@@ -3257,7 +3276,7 @@ int dbnumFromDb(redisDb *db)
     serverPanic("invalid database pointer");
 }
 
-bool redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command, bool fExecOK)
+void redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command)
 {
     if (m_spstorage == nullptr) {
 #if defined(__x86_64__) || defined(__i386__)
@@ -3269,7 +3288,7 @@ bool redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
             const char *cmd = szFromObj(command.argv[0]);
             if (!strcasecmp(cmd, "set") || !strcasecmp(cmd, "get")) {
                 if (c->db->m_spdbSnapshotHOLDER != nullptr)
-                    return false; // this is dangerous enough without a snapshot around
+                    return; // this is dangerous enough without a snapshot around
                 auto h = dictSdsHash(szFromObj(command.argv[1]));
                 for (int iht = 0; iht < 2; ++iht) {
                     auto hT = h & c->db->m_pdict->ht[iht].sizemask;
@@ -3289,7 +3308,7 @@ bool redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
             }
         }
 #endif
-        return false;
+        return;
     }
 
     AeLocker lock;
@@ -3299,10 +3318,10 @@ bool redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
     getKeysResult result = GETKEYS_RESULT_INIT;
     auto cmd = lookupCommand(szFromObj(command.argv[0]));
     if (cmd == nullptr)
-        return false; // Bad command? It's not for us to judge, just bail
+        return; // Bad command? It's not for us to judge, just bail
     
     if (command.argc < std::abs(cmd->arity))
-        return false; // Invalid number of args
+        return; // Invalid number of args
     
     int numkeys = getKeysFromCommand(cmd, command.argv, command.argc, &result);
     for (int ikey = 0; ikey < numkeys; ++ikey)
@@ -3335,7 +3354,6 @@ bool redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
         }
     }
 
-    bool fNoInsert = false;
     if (!vecInserts.empty()) {
         lock.arm(c);
         for (auto &tuple : vecInserts)
@@ -3351,13 +3369,11 @@ bool redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
                     // While unlocked this was already ensured
                     decrRefCount(o);
                     sdsfree(sharedKey);
-                    fNoInsert = true;
                 }
                 else
                 {
                     if (spexpire != nullptr) {
                         if (spexpire->when() < mstime()) {
-                            fNoInsert = true;
                             break;
                         }
                     }
@@ -3384,12 +3400,5 @@ bool redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
         lock.disarm();
     }
 
-    if (fExecOK && !fNoInsert && cmd->proc == getCommand && !vecInserts.empty()) {
-        robj *o = std::get<1>(vecInserts[0]);
-        if (o != nullptr) {
-            addReplyBulk(c, o);
-            return true;
-        }
-    }
-    return false;
-}
+    return;
+}

src/debug.cpp

@@ -341,10 +341,10 @@ void mallctl_int(client *c, robj **argv, int argc) {
     }
     size_t sz = sizeof(old);
     while (sz > 0) {
-        if ((ret=je_mallctl(szFromObj(argv[0]), &old, &sz, argc > 1? &val: NULL, argc > 1?sz: 0))) {
+        if ((ret=mallctl(szFromObj(argv[0]), &old, &sz, argc > 1? &val: NULL, argc > 1?sz: 0))) {
             if (ret == EPERM && argc > 1) {
                 /* if this option is write only, try just writing to it. */
-                if (!(ret=je_mallctl(szFromObj(argv[0]), NULL, 0, &val, sz))) {
+                if (!(ret=mallctl(szFromObj(argv[0]), NULL, 0, &val, sz))) {
                     addReply(c, shared.ok);
                     return;
                 }
@@ -375,7 +375,7 @@ void mallctl_string(client *c, robj **argv, int argc) {
     char *old;
     size_t sz = sizeof(old);
     /* for strings, it seems we need to first get the old value, before overriding it. */
-    if ((rret=je_mallctl(szFromObj(argv[0]), &old, &sz, NULL, 0))) {
+    if ((rret=mallctl(szFromObj(argv[0]), &old, &sz, NULL, 0))) {
         /* return error unless this option is write only. */
         if (!(rret == EPERM && argc > 1)) {
             addReplyErrorFormat(c,"%s", strerror(rret));
@@ -387,7 +387,7 @@ void mallctl_string(client *c, robj **argv, int argc) {
         char **valref = &val;
         if ((!strcmp(val,"VOID")))
             valref = NULL, sz = 0;
-        wret = je_mallctl(szFromObj(argv[0]), NULL, 0, valref, sz);
+        wret = mallctl(szFromObj(argv[0]), NULL, 0, valref, sz);
     }
     if (!rret)
         addReplyBulkCString(c, old);

src/dict.cpp

@@ -1577,7 +1577,7 @@ char *stringFromLongLong(long long value) {
     int len;
     char *s;
 
-    len = snprintf(buf,32,"%lld",value);
+    len = snprintf(buf,sizeof(buf),"%lld",value);
     s = zmalloc(len+1);
     memcpy(s, buf, len);
     s[len] = '\0';

src/endianconv.c

@@ -112,15 +112,15 @@ int endianconvTest(int argc, char *argv[], int accurate) {
     UNUSED(argv);
     UNUSED(accurate);
 
-    snprintf(buf,32,"ciaoroma");
+    snprintf(buf,sizeof(buf),"ciaoroma");
     memrev16(buf);
     printf("%s\n", buf);
 
-    snprintf(buf,32,"ciaoroma");
+    snprintf(buf,sizeof(buf),"ciaoroma");
     memrev32(buf);
     printf("%s\n", buf);
 
-    snprintf(buf,32,"ciaoroma");
+    snprintf(buf,sizeof(buf),"ciaoroma");
     memrev64(buf);
     printf("%s\n", buf);
 

src/keydb-diagnostic-tool.cpp

@@ -904,7 +904,7 @@ int main(int argc, const char **argv) {
 
     while (self_threads < config.max_threads) {
         for (int i = 0; i < config.numclients; i++) {
-            snprintf(command, 63, "SET %d %s\r\n", self_threads * config.numclients + i, set_value);
+            snprintf(command, sizeof(command), "SET %d %s\r\n", self_threads * config.numclients + i, set_value);
             createClient(command, strlen(command), NULL,self_threads);
         }
 

src/modules/hellotimer.c

@@ -51,8 +51,9 @@ int TimerCommand_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int
 
     for (int j = 0; j < 10; j++) {
         int delay = rand() % 5000;
-        char *buf = RedisModule_Alloc(256);
-        snprintf(buf,256,"After %d", delay);
+        int bufsize = 256;
+        char *buf = RedisModule_Alloc(bufsize);
+        snprintf(buf,bufsize,"After %d", delay);
         RedisModuleTimerID tid = RedisModule_CreateTimer(ctx,delay,timerHandler,buf);
         REDISMODULE_NOT_USED(tid);
     }

src/networking.cpp

@@ -145,7 +145,7 @@ client *createClient(connection *conn, int iel) {
     client_id = g_pserver->next_client_id.fetch_add(1);
     c->iel = iel;
     c->id = client_id;
-    snprintf(c->lock.szName, 56, "client %" PRIu64, client_id);
+    snprintf(c->lock.szName, sizeof(c->lock.szName), "client %" PRIu64, client_id);
     c->resp = 2;
     c->conn = conn;
     c->name = NULL;
@@ -777,11 +777,11 @@ void setDeferredAggregateLen(client *c, void *node, long length, char prefix) {
          * we return NULL in addReplyDeferredLen() */
         if (node == NULL) return;
         char lenstr[128];
-        size_t lenstr_len = snprintf(lenstr, 128, "%c%ld\r\n", prefix, length);
+        size_t lenstr_len = snprintf(lenstr, sizeof(lenstr), "%c%ld\r\n", prefix, length);
         setDeferredReply(c, node, lenstr, lenstr_len);
     } else {
         char lenstr[128];
-        int lenstr_len = snprintf(lenstr, 128, "%c%ld\r\n", prefix, length);
+        int lenstr_len = snprintf(lenstr, sizeof(lenstr), "%c%ld\r\n", prefix, length);
 
         size_t idxSplice = (size_t)node;
         serverAssert(idxSplice <= c->replyAsync->used);
@@ -2595,9 +2595,7 @@ void parseClientCommandBuffer(client *c) {
             (g_pserver->m_pstorageFactory || aeLockContested(cserver.cthreads/2) || cserver.cthreads == 1) && !GlobalLocksAcquired()) {
             auto &query = c->vecqueuedcmd.back();
             if (query.argc > 0 && query.argc == query.argcMax) {
-                if (c->db->prefetchKeysAsync(c, query, c->vecqueuedcmd.size() == 1)) {
-                    c->vecqueuedcmd.erase(c->vecqueuedcmd.begin());
-                }
+                c->db->prefetchKeysAsync(c, query);
             }
         }
         c->reqtype = 0;

src/new.cpp

@@ -41,10 +41,3 @@ void operator delete(void *p, std::size_t) noexcept
 }
 
 #endif
-
-#if defined(USE_JEMALLOC)
-extern "C" size_t malloc_usable_size(void *ptr)
-{
-    return zmalloc_usable_size(ptr);
-}
-#endif

src/object.cpp

@@ -1541,7 +1541,7 @@ void memoryCommand(client *c) {
     } else if (!strcasecmp(szFromObj(c->argv[1]),"malloc-stats") && c->argc == 2) {
 #if defined(USE_JEMALLOC)
         sds info = sdsempty();
-        je_malloc_stats_print(inputCatSds, &info, NULL);
+        malloc_stats_print(inputCatSds, &info, NULL);
         addReplyVerbatim(c,info,sdslen(info),"txt");
         sdsfree(info);
 #else

src/rdb.cpp

@@ -1163,7 +1163,7 @@ int rdbSaveKeyValuePair(rio *rdb, robj_roptr key, robj_roptr val, const expireEn
 
     char szT[32];
     if (g_pserver->fActiveReplica) {
-        snprintf(szT, 32, "%" PRIu64, mvccFromObj(val));
+        snprintf(szT, sizeof(szT), "%" PRIu64, mvccFromObj(val));
         if (rdbSaveAuxFieldStrStr(rdb,"mvcc-tstamp", szT) == -1) return -1;
     }
 
@@ -1190,7 +1190,7 @@ int rdbSaveKeyValuePair(rio *rdb, robj_roptr key, robj_roptr val, const expireEn
         {
             if (itr.subkey() == nullptr)
                 continue;   // already saved
-            snprintf(szT, 32, "%lld", itr.when());
+            snprintf(szT, sizeof(szT), "%lld", itr.when());
             rdbSaveAuxFieldStrStr(rdb,"keydb-subexpire-key",itr.subkey());
             rdbSaveAuxFieldStrStr(rdb,"keydb-subexpire-when",szT);
         }
@@ -1490,7 +1490,7 @@ int rdbSaveFile(char *filename, const redisDbPersistentDataSnapshot **rgpdb, rdb
     rio rdb;
     int error = 0;
 
-    snprintf(tmpfile,256,"temp-%d-%d.rdb", getpid(), g_pserver->rdbThreadVars.tmpfileNum);
+    getTempFileName(tmpfile, g_pserver->rdbThreadVars.tmpfileNum);
     fp = fopen(tmpfile,"w");
     if (!fp) {
         char *cwdp = getcwd(cwd,MAXPATHLEN);
@@ -1660,12 +1660,14 @@ int launchRdbSaveThread(pthread_t &child, rdbSaveInfo *rsi)
         pthread_attr_t tattr;
         pthread_attr_init(&tattr);
         pthread_attr_setstacksize(&tattr, 1 << 23); // 8 MB
+        openChildInfoPipe();
         if (pthread_create(&child, &tattr, rdbSaveThread, args)) {
             pthread_attr_destroy(&tattr);
             for (int idb = 0; idb < cserver.dbnum; ++idb)
                 g_pserver->db[idb]->endSnapshot(args->rgpdb[idb]);
             args->~rdbSaveThreadArgs();
             zfree(args);
+            closeChildInfoPipe();
             return C_ERR;
         }
         pthread_attr_destroy(&tattr);
@@ -1679,24 +1681,23 @@ int rdbSaveBackground(rdbSaveInfo *rsi) {
     pthread_t child;
     long long start;
 
-    if (hasActiveChildProcess()) return C_ERR;
+    if (hasActiveChildProcessOrBGSave()) return C_ERR;
 
     g_pserver->dirty_before_bgsave = g_pserver->dirty;
     g_pserver->lastbgsave_try = time(NULL);
-    openChildInfoPipe();
 
     start = ustime();
+    latencyStartMonitor(g_pserver->rdb_save_latency);
 
-    
-    g_pserver->stat_fork_time = ustime()-start;
-    g_pserver->stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / g_pserver->stat_fork_time / (1024*1024*1024); /* GB per second. */
     if (launchRdbSaveThread(child, rsi) != C_OK) {
-        closeChildInfoPipe();
         g_pserver->lastbgsave_status = C_ERR;
         serverLog(LL_WARNING,"Can't save in background: fork: %s",
             strerror(errno));
         return C_ERR;
     }
+
+    g_pserver->stat_fork_time = ustime()-start;
+    g_pserver->stat_fork_rate = (double) zmalloc_used_memory() * 1000000 / g_pserver->stat_fork_time / (1024*1024*1024); /* GB per second. */
     latencyAddSampleIfNeeded("fork",g_pserver->stat_fork_time/1000);
     serverLog(LL_NOTICE,"Background saving started");
     g_pserver->rdb_save_time_start = time(NULL);
@@ -3601,6 +3602,8 @@ static void backgroundSaveDoneHandlerDisk(int exitcode, bool fCancelled) {
         g_pserver->dirty = g_pserver->dirty - g_pserver->dirty_before_bgsave;
         g_pserver->lastsave = time(NULL);
         g_pserver->lastbgsave_status = C_OK;
+        latencyEndMonitor(g_pserver->rdb_save_latency);
+        latencyAddSampleIfNeeded("rdb-save",g_pserver->rdb_save_latency);
     } else if (!fCancelled && exitcode != 0) {
         serverLog(LL_WARNING, "Background saving error");
         g_pserver->lastbgsave_status = C_ERR;
@@ -3691,7 +3694,7 @@ void killRDBChild(bool fSynchronous) {
     serverAssert(GlobalLocksAcquired());
 
     if (cserver.fForkBgSave) {
-        kill(g_pserver->rdb_child_pid,SIGUSR1);
+        kill(g_pserver->child_pid,SIGUSR1);
     } else { 
         g_pserver->rdbThreadVars.fRdbThreadCancel = true;
         if (g_pserver->rdb_child_type == RDB_CHILD_TYPE_SOCKET) {
@@ -3782,7 +3785,7 @@ int rdbSaveToSlavesSockets(rdbSaveInfo *rsi) {
     int pipefds[2];
     rdbSaveSocketThreadArgs *args = nullptr;
 
-    if (hasActiveChildProcess()) return C_ERR;
+    if (hasActiveChildProcessOrBGSave()) return C_ERR;
 
     /* Even if the previous fork child exited, don't start a new one until we
      * drained the pipe. */

src/rdb.h

@@ -151,6 +151,7 @@ int rdbLoad(rdbSaveInfo *rsi, int rdbflags);
 int rdbLoadFile(const char *filename, rdbSaveInfo *rsi, int rdbflags);
 int rdbSaveBackground(rdbSaveInfo *rsi);
 int rdbSaveToSlavesSockets(rdbSaveInfo *rsi);
+void getTempFileName(char tmpfile[], int tmpfileNum);
 void rdbRemoveTempFile(pid_t childpid, int from_signal);
 int rdbSave(const redisDbPersistentDataSnapshot **rgpdb, rdbSaveInfo *rsi);
 int rdbSaveFile(char *filename, const redisDbPersistentDataSnapshot **rgpdb, rdbSaveInfo *rsi);

src/readwritelock.h

@@ -8,6 +8,7 @@ class readWriteLock {
     int m_readCount = 0;
     int m_writeCount = 0;
     bool m_writeWaiting = false;
+    bool m_notify = true;
 public:
     readWriteLock(const char *name) : m_readLock(name), m_writeLock(name) {}
 
@@ -65,7 +66,8 @@ public:
     void releaseRead() {
         std::unique_lock<fastlock> rm(m_readLock);
         m_readCount--;
-        m_cv.notify_all();
+        if (m_notify)
+            m_cv.notify_all();
     }
 
     void releaseWrite(bool exclusive = true) {
@@ -74,7 +76,8 @@ public:
         if (exclusive)
             m_writeLock.unlock();
         m_writeCount--;
-        m_cv.notify_all();
+        if (m_notify)
+            m_cv.notify_all();
     }
 
     void downgradeWrite(bool exclusive = true) {
@@ -82,6 +85,10 @@ public:
         acquireRead();
     }
 
+    void setNotify(bool notify) {
+        m_notify = notify;
+    }
+
     bool hasReader() {
         return m_readCount > 0;
     }

src/redis-cli.c

@@ -3293,8 +3293,8 @@ static redisReply *clusterManagerMigrateKeysInReply(clusterManagerNode *source,
     argv_len = zcalloc(argc * sizeof(size_t), MALLOC_LOCAL);
     char portstr[255];
     char timeoutstr[255];
-    snprintf(portstr, 10, "%d", target->port);
-    snprintf(timeoutstr, 10, "%d", timeout);
+    snprintf(portstr, sizeof(portstr), "%d", target->port);
+    snprintf(timeoutstr, sizeof(timeoutstr), "%d", timeout);
     argv[0] = "MIGRATE";
     argv_len[0] = 7;
     argv[1] = target->ip;
@@ -6825,7 +6825,7 @@ static long getLongInfoField(char *info, char *field) {
 
 /* Convert number of bytes into a human readable string of the form:
  * 100B, 2G, 100M, 4K, and so forth. */
-void bytesToHuman(char *s, long long n) {
+void bytesToHuman(char *s, long long n, size_t bufsize) {
     double d;
 
     if (n < 0) {
@@ -6835,17 +6835,17 @@ void bytesToHuman(char *s, long long n) {
     }
     if (n < 1024) {
         /* Bytes */
-        snprintf(s,256,"%lldB",n);
+        snprintf(s,bufsize,"%lldB",n);
         return;
     } else if (n < (1024*1024)) {
         d = (double)n/(1024);
-        snprintf(s,256,"%.2fK",d);
+        snprintf(s,bufsize,"%.2fK",d);
     } else if (n < (1024LL*1024*1024)) {
         d = (double)n/(1024*1024);
-        snprintf(s,256,"%.2fM",d);
+        snprintf(s,bufsize,"%.2fM",d);
     } else if (n < (1024LL*1024*1024*1024)) {
         d = (double)n/(1024LL*1024*1024);
-        snprintf(s,256,"%.2fG",d);
+        snprintf(s,bufsize,"%.2fG",d);
     }
 }
 
@@ -6875,38 +6875,38 @@ static void statMode(void) {
         for (j = 0; j < 20; j++) {
             long k;
 
-            snprintf(buf,64,"db%d:keys",j);
+            snprintf(buf,sizeof(buf),"db%d:keys",j);
             k = getLongInfoField(reply->str,buf);
             if (k == LONG_MIN) continue;
             aux += k;
         }
-        snprintf(buf,64,"%ld",aux);
+        snprintf(buf,sizeof(buf),"%ld",aux);
         printf("%-11s",buf);
 
         /* Used memory */
         aux = getLongInfoField(reply->str,"used_memory");
-        bytesToHuman(buf,aux);
+        bytesToHuman(buf,aux,sizeof(buf));
         printf("%-8s",buf);
 
         /* Clients */
         aux = getLongInfoField(reply->str,"connected_clients");
-        snprintf(buf,64,"%ld",aux);
+        snprintf(buf,sizeof(buf),"%ld",aux);
         printf(" %-8s",buf);
 
         /* Blocked (BLPOPPING) Clients */
         aux = getLongInfoField(reply->str,"blocked_clients");
-        snprintf(buf,64,"%ld",aux);
+        snprintf(buf,sizeof(buf),"%ld",aux);
         printf("%-8s",buf);
 
         /* Requests */
         aux = getLongInfoField(reply->str,"total_commands_processed");
-        snprintf(buf,64,"%ld (+%ld)",aux,requests == 0 ? 0 : aux-requests);
+        snprintf(buf,sizeof(buf),"%ld (+%ld)",aux,requests == 0 ? 0 : aux-requests);
         printf("%-19s",buf);
         requests = aux;
 
         /* Connections */
         aux = getLongInfoField(reply->str,"total_connections_received");
-        snprintf(buf,64,"%ld",aux);
+        snprintf(buf,sizeof(buf),"%ld",aux);
         printf(" %-12s",buf);
 
         /* Children */

src/redismodule.h

@@ -326,6 +326,7 @@ static const RedisModuleEvent
 #define REDISMODULE_SUBEVENT_LOADING_ENDED 3
 #define REDISMODULE_SUBEVENT_LOADING_FAILED 4
 #define _REDISMODULE_SUBEVENT_LOADING_NEXT 5
+#define REDISMODULE_SUBEVENT_LOADING_FLASH_START 6
 
 #define REDISMODULE_SUBEVENT_CLIENT_CHANGE_CONNECTED 0
 #define REDISMODULE_SUBEVENT_CLIENT_CHANGE_DISCONNECTED 1

src/replication.cpp

@@ -1538,7 +1538,7 @@ void syncCommand(client *c) {
         } else {
             /* We don't have a BGSAVE in progress, let's start one. Diskless
              * or disk-based mode is determined by replica's capacity. */
-            if (!hasActiveChildProcess()) {
+            if (!hasActiveChildProcessOrBGSave()) {
                 startBgsaveForReplication(c->slave_capa);
             } else {
                 serverLog(LL_NOTICE,
@@ -3736,7 +3736,7 @@ retry_connect:
         while(maxtries--) {
             auto dt = std::chrono::system_clock::now().time_since_epoch();
             auto dtMillisecond = std::chrono::duration_cast<std::chrono::milliseconds>(dt);
-            snprintf(tmpfile,256,
+            snprintf(tmpfile,sizeof(tmpfile),
                 "temp-%d.%ld.rdb",(int)dtMillisecond.count(),(long int)getpid());
             dfd = open(tmpfile,O_CREAT|O_WRONLY|O_EXCL,0644);
             if (dfd != -1) break;
@@ -3797,9 +3797,11 @@ int connectWithMaster(redisMaster *mi) {
     connSetPrivateData(mi->repl_transfer_s, mi);
     if (connConnect(mi->repl_transfer_s, mi->masterhost, mi->masterport,
                 NET_FIRST_BIND_ADDR, syncWithMaster) == C_ERR) {
-        int sev = g_pserver->enable_multimaster ? LL_NOTICE : LL_WARNING;   // with multimaster its not unheard of to intentiallionall have downed masters
-        serverLog(sev,"Unable to connect to MASTER: %s",
-                connGetLastError(mi->repl_transfer_s));
+        const char *err = "Unknown Error";
+        if (mi->repl_transfer_s->last_errno != 0)
+            err = connGetLastError(mi->repl_transfer_s);
+        int sev = g_pserver->enable_multimaster ? LL_NOTICE : LL_WARNING;   // with multimaster its not unheard of to intentionally have downed masters
+        serverLog(sev, "Unable to connect to MASTER: %s", err);
         connClose(mi->repl_transfer_s);
         mi->repl_transfer_s = NULL;
         return C_ERR;
@@ -4984,7 +4986,7 @@ void replicationStartPendingFork(void) {
      * In case of diskless replication, we make sure to wait the specified
      * number of seconds (according to configuration) so that other slaves
      * have the time to arrive before we start streaming. */
-    if (!hasActiveChildProcess()) {
+    if (!hasActiveChildProcessOrBGSave()) {
         time_t idle, max_idle = 0;
         int slaves_waiting = 0;
         int mincapa = -1;

src/semiorderedset.h

@@ -27,6 +27,7 @@ namespace keydbutils
     template<>
     size_t hash(const sdsview &);
 }
+extern size_t g_semiOrderedSetTargetBucketSize;
 
 template<typename T, typename T_KEY = T, bool MEMMOVE_SAFE = false>
 class semiorderedset
@@ -41,11 +42,14 @@ class semiorderedset
     size_t idxRehash = (1ULL << bits_min); 
     int cfPauseRehash = 0;
 
-    constexpr size_t targetElementsPerBucket()
+    inline size_t targetElementsPerBucket()
     {
         // Aim for roughly 4 cache lines per bucket (determined by imperical testing)
         //  lower values are faster but use more memory
-        return std::max((64/sizeof(T))*8, (size_t)2);
+        if (g_semiOrderedSetTargetBucketSize == 0)
+            return std::max((64/sizeof(T))*8, (size_t)2);
+        else
+            return g_semiOrderedSetTargetBucketSize;
     }
 
 public:

src/server.cpp

@@ -1706,7 +1706,7 @@ int redisDbPersistentData::incrementallyRehash() {
  * for dict.c to resize the hash tables accordingly to the fact we have an
  * active fork child running. */
 void updateDictResizePolicy(void) {
-    if (!hasActiveChildProcess() || (g_pserver->FRdbSaveInProgress() && !cserver.fForkBgSave))
+    if (!hasActiveChildProcess())
         dictEnableResize();
     else
         dictDisableResize();
@@ -1725,7 +1725,11 @@ const char *strChildType(int type) {
 /* Return true if there are active children processes doing RDB saving,
  * AOF rewriting, or some side process spawned by a loaded module. */
 int hasActiveChildProcess() {
-    return g_pserver->FRdbSaveInProgress() || g_pserver->child_pid != -1;
+    return g_pserver->child_pid != -1;
+}
+
+int hasActiveChildProcessOrBGSave() {
+    return g_pserver->FRdbSaveInProgress() || hasActiveChildProcess();
 }
 
 void resetChildState() {
@@ -2074,7 +2078,7 @@ void databasesCron(bool fMainThread) {
     /* Perform hash tables rehashing if needed, but only if there are no
      * other processes saving the DB on disk. Otherwise rehashing is bad
      * as will cause a lot of copy-on-write of memory pages. */
-    if (!hasActiveChildProcess() || g_pserver->FRdbSaveInProgress()) {
+    if (!hasActiveChildProcess()) {
         /* We use global counters so if we stop the computation at a given
          * DB we'll be able to start from the successive in the next
          * cron loop iteration. */
@@ -2215,6 +2219,7 @@ void checkChildrenDone(void) {
             g_pserver->rdbThreadVars.fRdbThreadCancel = false;
             g_pserver->rdbThreadVars.fDone = false;
             if (exitcode == 0) receiveChildInfo();
+            closeChildInfoPipe();
         }
     }
     else if ((pid = waitpid(-1, &statloc, WNOHANG)) != 0) {
@@ -2468,14 +2473,14 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
 
     /* Start a scheduled AOF rewrite if this was requested by the user while
      * a BGSAVE was in progress. */
-    if (!hasActiveChildProcess() &&
+    if (!hasActiveChildProcessOrBGSave() &&
         g_pserver->aof_rewrite_scheduled)
     {
         rewriteAppendOnlyFileBackground();
     }
 
     /* Check if a background saving or AOF rewrite in progress terminated. */
-    if (hasActiveChildProcess() || ldbPendingChildren())
+    if (hasActiveChildProcessOrBGSave() || ldbPendingChildren())
     {
         run_with_period(1000) receiveChildInfo();
         checkChildrenDone();
@@ -2517,7 +2522,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
 
         /* Trigger an AOF rewrite if needed. */
         if (g_pserver->aof_state == AOF_ON &&
-            !hasActiveChildProcess() &&
+            !hasActiveChildProcessOrBGSave() &&
             g_pserver->aof_rewrite_perc &&
             g_pserver->aof_current_size > g_pserver->aof_rewrite_min_size)
         {
@@ -2601,7 +2606,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
      * Note: this code must be after the replicationCron() call above so
      * make sure when refactoring this file to keep this order. This is useful
      * because we want to give priority to RDB savings for replication. */
-    if (!hasActiveChildProcess() &&
+    if (!hasActiveChildProcessOrBGSave() &&
         g_pserver->rdb_bgsave_scheduled &&
         (g_pserver->unixtime-g_pserver->lastbgsave_try > CONFIG_BGSAVE_RETRY_DELAY ||
          g_pserver->lastbgsave_status == C_OK))
@@ -2815,9 +2820,6 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
 
     runAndPropogateToReplicas(processClients);
 
-    /* Handle precise timeouts of blocked clients. */
-    handleBlockedClientsTimeout();
-
     /* Just call a subset of vital functions in case we are re-entering
      * the event loop from processEventsWhileBlocked(). Note that in this
      * case we keep track of the number of events we are processing, since
@@ -4067,12 +4069,6 @@ void initServer(void) {
         }
     }
 
-    /* We have to initialize storage providers after the cluster has been initialized */
-    for (int idb = 0; idb < cserver.dbnum; ++idb)
-    {
-        g_pserver->db[idb]->storageProviderInitialize();
-    }
-
     saveMasterStatusToStorage(false); // eliminate the repl-offset field
     
     /* Initialize ACL default password if it exists */
@@ -4085,6 +4081,15 @@ void initServer(void) {
  * Thread Local Storage initialization collides with dlopen call.
  * see: https://sourceware.org/bugzilla/show_bug.cgi?id=19329 */
 void InitServerLast() {
+
+    /* We have to initialize storage providers after the cluster has been initialized */
+    moduleFireServerEvent(REDISMODULE_EVENT_LOADING, REDISMODULE_SUBEVENT_LOADING_FLASH_START, NULL);
+    for (int idb = 0; idb < cserver.dbnum; ++idb)
+    {
+        g_pserver->db[idb]->storageProviderInitialize();
+    }
+    moduleFireServerEvent(REDISMODULE_EVENT_LOADING, REDISMODULE_SUBEVENT_LOADING_ENDED, NULL);
+
     bioInit();
     set_jemalloc_bg_thread(cserver.jemalloc_bg_thread);
     g_pserver->initial_memory_usage = zmalloc_used_memory();
@@ -5467,30 +5472,30 @@ NULL
 
 /* Convert an amount of bytes into a human readable string in the form
  * of 100B, 2G, 100M, 4K, and so forth. */
-void bytesToHuman(char *s, unsigned long long n) {
+void bytesToHuman(char *s, unsigned long long n, size_t bufsize) {
     double d;
 
     if (n < 1024) {
         /* Bytes */
-        snprintf(s,256,"%lluB",n);
+        snprintf(s,bufsize,"%lluB",n);
     } else if (n < (1024*1024)) {
         d = (double)n/(1024);
-        snprintf(s,256,"%.2fK",d);
+        snprintf(s,bufsize,"%.2fK",d);
     } else if (n < (1024LL*1024*1024)) {
         d = (double)n/(1024*1024);
-        snprintf(s,256,"%.2fM",d);
+        snprintf(s,bufsize,"%.2fM",d);
     } else if (n < (1024LL*1024*1024*1024)) {
         d = (double)n/(1024LL*1024*1024);
-        snprintf(s,256,"%.2fG",d);
+        snprintf(s,bufsize,"%.2fG",d);
     } else if (n < (1024LL*1024*1024*1024*1024)) {
         d = (double)n/(1024LL*1024*1024*1024);
-        snprintf(s,256,"%.2fT",d);
+        snprintf(s,bufsize,"%.2fT",d);
     } else if (n < (1024LL*1024*1024*1024*1024*1024)) {
         d = (double)n/(1024LL*1024*1024*1024*1024);
-        snprintf(s,256,"%.2fP",d);
+        snprintf(s,bufsize,"%.2fP",d);
     } else {
         /* Let's hope we never need this */
-        snprintf(s,256,"%lluB",n);
+        snprintf(s,bufsize,"%lluB",n);
     }
 }
 
@@ -5666,13 +5671,13 @@ sds genRedisInfoString(const char *section) {
         if (zmalloc_used > g_pserver->stat_peak_memory)
             g_pserver->stat_peak_memory = zmalloc_used;
 
-        bytesToHuman(hmem,zmalloc_used);
-        bytesToHuman(peak_hmem,g_pserver->stat_peak_memory);
-        bytesToHuman(total_system_hmem,total_system_mem);
-        bytesToHuman(used_memory_lua_hmem,memory_lua);
-        bytesToHuman(used_memory_scripts_hmem,mh->lua_caches);
-        bytesToHuman(used_memory_rss_hmem,g_pserver->cron_malloc_stats.process_rss);
-        bytesToHuman(maxmemory_hmem,g_pserver->maxmemory);
+        bytesToHuman(hmem,zmalloc_used,sizeof(hmem));
+        bytesToHuman(peak_hmem,g_pserver->stat_peak_memory,sizeof(peak_hmem));
+        bytesToHuman(total_system_hmem,total_system_mem,sizeof(total_system_hmem));
+        bytesToHuman(used_memory_lua_hmem,memory_lua,sizeof(used_memory_lua_hmem));
+        bytesToHuman(used_memory_scripts_hmem,mh->lua_caches,sizeof(used_memory_scripts_hmem));
+        bytesToHuman(used_memory_rss_hmem,g_pserver->cron_malloc_stats.process_rss,sizeof(used_memory_rss_hmem));
+        bytesToHuman(maxmemory_hmem,g_pserver->maxmemory,sizeof(maxmemory_hmem));
 
         if (sections++) info = sdscat(info,"\r\n");
         info = sdscatprintf(info,
@@ -6559,7 +6564,8 @@ void usage(void) {
 
 void redisAsciiArt(void) {
 #include "asciilogo.h"
-    char *buf = (char*)zmalloc(1024*16, MALLOC_LOCAL);
+    size_t bufsize = 1024*16;
+    char *buf = (char*)zmalloc(bufsize, MALLOC_LOCAL);
     const char *mode;
 
     if (g_pserver->cluster_enabled) mode = "cluster";
@@ -6581,7 +6587,7 @@ void redisAsciiArt(void) {
         );
     } else {
         sds motd = fetchMOTD(true, cserver.enable_motd);
-        snprintf(buf,1024*16,ascii_logo,
+        snprintf(buf,bufsize,ascii_logo,
             KEYDB_REAL_VERSION,
             redisGitSHA1(),
             strtol(redisGitDirty(),NULL,10) > 0,
@@ -6884,6 +6890,7 @@ int redisFork(int purpose) {
     latencyAddSampleIfNeeded("fork-lock",(ustime()-startWriteLock)/1000);
     if ((childpid = fork()) == 0) {
         /* Child */
+        aeForkLockInChild();
         aeReleaseForkLock();
         g_pserver->in_fork_child = purpose;
         setOOMScoreAdj(CONFIG_OOM_BGCHILD);
@@ -7566,6 +7573,29 @@ int main(int argc, char **argv) {
 
     validateConfiguration();
 
+    if (!g_pserver->sentinel_mode) {
+    #ifdef __linux__
+        linuxMemoryWarnings();
+    #if defined (__arm64__)
+        int ret;
+        if ((ret = linuxMadvFreeForkBugCheck())) {
+            if (ret == 1)
+                serverLog(LL_WARNING,"WARNING Your kernel has a bug that could lead to data corruption during background save. "
+                                        "Please upgrade to the latest stable kernel.");
+            else
+                serverLog(LL_WARNING, "Failed to test the kernel for a bug that could lead to data corruption during background save. "
+                                        "Your system could be affected, please report this error.");
+            if (!checkIgnoreWarning("ARM64-COW-BUG")) {
+                serverLog(LL_WARNING,"KeyDB will now exit to prevent data corruption. "
+                                        "Note that it is possible to suppress this warning by setting the following config: ignore-warnings ARM64-COW-BUG");
+                exit(1);
+            }
+        }
+    #endif /* __arm64__ */
+    #endif /* __linux__ */
+    }
+
+
     const char *err;
     if (!initializeStorageProvider(&err))
     {
@@ -7591,25 +7621,6 @@ int main(int argc, char **argv) {
     if (!g_pserver->sentinel_mode) {
         /* Things not needed when running in Sentinel mode. */
         serverLog(LL_WARNING,"Server initialized");
-    #ifdef __linux__
-        linuxMemoryWarnings();
-    #if defined (__arm64__)
-        int ret;
-        if ((ret = linuxMadvFreeForkBugCheck())) {
-            if (ret == 1)
-                serverLog(LL_WARNING,"WARNING Your kernel has a bug that could lead to data corruption during background save. "
-                                     "Please upgrade to the latest stable kernel.");
-            else
-                serverLog(LL_WARNING, "Failed to test the kernel for a bug that could lead to data corruption during background save. "
-                                      "Your system could be affected, please report this error.");
-            if (!checkIgnoreWarning("ARM64-COW-BUG")) {
-                serverLog(LL_WARNING,"KeyDB will now exit to prevent data corruption. "
-                                     "Note that it is possible to suppress this warning by setting the following config: ignore-warnings ARM64-COW-BUG");
-                exit(1);
-            }
-        }
-    #endif /* __arm64__ */
-    #endif /* __linux__ */
         moduleInitModulesSystemLast();
         moduleLoadFromQueue();
         ACLLoadUsersAtStartup();

src/server.h

@@ -1204,12 +1204,12 @@ public:
     void disableKeyCache();
     bool keycacheIsEnabled();
 
-    bool prefetchKeysAsync(client *c, struct parsed_command &command, bool fExecOK);
+    void prefetchKeysAsync(client *c, struct parsed_command &command);
 
     bool FSnapshot() const { return m_spdbSnapshotHOLDER != nullptr; }
 
     std::unique_ptr<const StorageCache> CloneStorageCache() { return std::unique_ptr<const StorageCache>(m_spstorage->clone()); }
-    void bulkStorageInsert(char **rgKeys, size_t *rgcbKeys, char **rgVals, size_t *rgcbVals, size_t celem);
+    void bulkDirectStorageInsert(char **rgKeys, size_t *rgcbKeys, char **rgVals, size_t *rgcbVals, size_t celem);
 
     dict_iter find_cached_threadsafe(const char *key) const;
 
@@ -1370,7 +1370,7 @@ struct redisDb : public redisDbPersistentDataSnapshot
     using redisDbPersistentData::FRehashing;
     using redisDbPersistentData::FTrackingChanges;
     using redisDbPersistentData::CloneStorageCache;
-    using redisDbPersistentData::bulkStorageInsert;
+    using redisDbPersistentData::bulkDirectStorageInsert;
 
 public:
     const redisDbPersistentDataSnapshot *createSnapshot(uint64_t mvccCheckpoint, bool fOptional) {
@@ -2470,7 +2470,8 @@ struct redisServer {
     time_t lastbgsave_try;          /* Unix time of last attempted bgsave */
     time_t rdb_save_time_last;      /* Time used by last RDB save run. */
     time_t rdb_save_time_start;     /* Current RDB save start time. */
-    pid_t rdb_child_pid = -1;            /* Used only during fork bgsave */
+    mstime_t rdb_save_latency;      /* Used to track end to end latency of rdb save*/
+    pid_t rdb_child_pid = -1;       /* Used only during fork bgsave */
     int rdb_bgsave_scheduled;       /* BGSAVE when possible if true. */
     int rdb_child_type;             /* Type of save by active child. */
     int lastbgsave_status;          /* C_OK or C_ERR */
@@ -3254,6 +3255,7 @@ void receiveChildInfo(void);
 void executeWithoutGlobalLock(std::function<void()> func);
 int redisFork(int type);
 int hasActiveChildProcess();
+int hasActiveChildProcessOrBGSave();
 void resetChildState();
 int isMutuallyExclusiveChildType(int type);
 
@@ -3892,13 +3894,13 @@ void incrementMvccTstamp();
 
 #if __GNUC__ >= 7 && !defined(NO_DEPRECATE_FREE)
  [[deprecated]]
-void *calloc(size_t count, size_t size);
+void *calloc(size_t count, size_t size) noexcept;
  [[deprecated]]
-void free(void *ptr);
+void free(void *ptr) noexcept;
  [[deprecated]]
-void *malloc(size_t size);
+void *malloc(size_t size) noexcept;
  [[deprecated]]
-void *realloc(void *ptr, size_t size);
+void *realloc(void *ptr, size_t size) noexcept;
 #endif
 
 /* Debugging stuff */

src/sort.cpp

@@ -194,8 +194,8 @@ void sortCommand(client *c) {
     list *operations;
     unsigned int outputlen = 0;
     int desc = 0, alpha = 0;
-    long limit_start = 0, limit_count = -1, start, end;
-    int j, dontsort = 0, vectorlen;
+    long limit_start = 0, limit_count = -1, start, end, vectorlen;
+    int j, dontsort = 0;
     int getop = 0; /* GET operation counter */
     int int_conversion_error = 0;
     int syntax_error = 0;
@@ -321,8 +321,10 @@ void sortCommand(client *c) {
     default: vectorlen = 0; serverPanic("Bad SORT type"); /* Avoid GCC warning */
     }
 
-    /* Perform LIMIT start,count sanity checking. */
-    start = (limit_start < 0) ? 0 : limit_start;
+    /* Perform LIMIT start,count sanity checking.
+     * And avoid integer overflow by limiting inputs to object sizes. */
+    start = std::min(std::max(limit_start, (long)0), vectorlen);
+    limit_count = std::min(std::max(limit_count, (long)-1), vectorlen);
     end = (limit_count < 0) ? vectorlen-1 : start+limit_count-1;
     if (start >= vectorlen) {
         start = vectorlen-1;

src/storage/rocksdb.cpp

@@ -100,14 +100,14 @@ bool RocksDBStorageProvider::erase(const char *key, size_t cchKey)
 {
     rocksdb::Status status;
     std::unique_lock<fastlock> l(m_lock);
+    if (!FKeyExists(key, cchKey))
+        return false;
     if (m_spbatch != nullptr)
     {
         status = m_spbatch->Delete(m_spcolfamily.get(), rocksdb::Slice(key, cchKey));
     }
     else
     {
-        if (!FKeyExists(key, cchKey))
-            return false;
         status = m_spdb->Delete(WriteOptions(), m_spcolfamily.get(), rocksdb::Slice(key, cchKey));
     }
     if (status.ok())
@@ -202,12 +202,12 @@ rocksdb::WriteOptions RocksDBStorageProvider::WriteOptions() const
 void RocksDBStorageProvider::beginWriteBatch()
 {
     m_lock.lock();
-    m_spbatch = std::make_unique<rocksdb::WriteBatch>();
+    m_spbatch = std::make_unique<rocksdb::WriteBatchWithIndex>();
 }
 
 void RocksDBStorageProvider::endWriteBatch()
 {
-    m_spdb->Write(WriteOptions(), m_spbatch.get());
+    m_spdb->Write(WriteOptions(), m_spbatch.get()->GetWriteBatch());
     m_spbatch = nullptr;
     m_lock.unlock();
 }
@@ -230,5 +230,7 @@ void RocksDBStorageProvider::flush()
 bool RocksDBStorageProvider::FKeyExists(const char *key, size_t cch) const
 {
     rocksdb::PinnableSlice slice;
+    if (m_spbatch)
+        return m_spbatch->GetFromBatchAndDB(m_spdb.get(), ReadOptions(), m_spcolfamily.get(), rocksdb::Slice(key, cch), &slice).ok();
     return m_spdb->Get(ReadOptions(), m_spcolfamily.get(), rocksdb::Slice(key, cch), &slice).ok();
 }

src/storage/rocksdb.h

@@ -3,6 +3,7 @@
 #include <memory>
 #include "../IStorage.h"
 #include <rocksdb/db.h>
+#include <rocksdb/utilities/write_batch_with_index.h>
 #include "../fastlock.h"
 
 #define INTERNAL_KEY_PREFIX "\x00\x04\x03\x00\x05\x02\x04"
@@ -15,7 +16,7 @@ class RocksDBStorageProvider : public IStorage
 {
     RocksDBStorageFactory *m_pfactory;
     std::shared_ptr<rocksdb::DB> m_spdb;    // Note: This must be first so it is deleted last
-    std::unique_ptr<rocksdb::WriteBatch> m_spbatch;
+    std::unique_ptr<rocksdb::WriteBatchWithIndex> m_spbatch;
     const rocksdb::Snapshot *m_psnapshot = nullptr;
     std::shared_ptr<rocksdb::ColumnFamilyHandle> m_spcolfamily;
     rocksdb::ReadOptions m_readOptionsTemplate;

src/t_hash.cpp

@@ -1225,8 +1225,13 @@ void hrandfieldCommand(client *c) {
         if (c->argc > 4 || (c->argc == 4 && strcasecmp(szFromObj(c->argv[3]),"withvalues"))) {
             addReplyErrorObject(c,shared.syntaxerr);
             return;
-        } else if (c->argc == 4)
+        } else if (c->argc == 4) {
             withvalues = 1;
+            if (l < LONG_MIN/2 || l > LONG_MAX/2) {
+                addReplyError(c,"value is out of range");
+                return;
+            }
+        }
         hrandfieldWithCountCommand(c, l, withvalues);
         return;
     }

src/t_string.cpp

@@ -38,8 +38,14 @@ int getGenericCommand(client *c);
  * String Commands
  *----------------------------------------------------------------------------*/
 
-static int checkStringLength(client *c, long long size) {
-    if (!(c->flags & CLIENT_MASTER) && size > g_pserver->proto_max_bulk_len) {
+static int checkStringLength(client *c, long long size, long long append) {
+    if (c->flags & CLIENT_MASTER)
+        return C_OK;
+    /* 'uint64_t' cast is there just to prevent undefined behavior on overflow */
+    long long total = (uint64_t)size + append;
+    /* Test configured max-bulk-len represending a limit of the biggest string object,
+     * and also test for overflow. */
+    if (total > g_pserver->proto_max_bulk_len || total < size || total < append) {
         addReplyError(c,"string exceeds maximum allowed size (proto-max-bulk-len)");
         return C_ERR;
     }
@@ -445,7 +451,7 @@ void setrangeCommand(client *c) {
         }
 
         /* Return when the resulting string exceeds allowed size */
-        if (checkStringLength(c,offset+sdslen(value)) != C_OK)
+        if (checkStringLength(c,offset,sdslen(value)) != C_OK)
             return;
 
         o = createObject(OBJ_STRING,sdsnewlen(NULL, offset+sdslen(value)));
@@ -465,7 +471,7 @@ void setrangeCommand(client *c) {
         }
 
         /* Return when the resulting string exceeds allowed size */
-        if (checkStringLength(c,offset+sdslen(value)) != C_OK)
+        if (checkStringLength(c,offset,sdslen(value)) != C_OK)
             return;
 
         /* Create a copy when the object is shared or encoded. */
@@ -685,8 +691,7 @@ void appendCommand(client *c) {
 
         /* "append" is an argument, so always an sds */
         append = c->argv[2];
-        totlen = stringObjectLen(o)+sdslen((sds)ptrFromObj(append));
-        if (checkStringLength(c,totlen) != C_OK)
+        if (checkStringLength(c,stringObjectLen(o),sdslen((sds)ptrFromObj(append))) != C_OK)
             return;
 
         /* Append the value */

src/t_zset.cpp

@@ -4214,8 +4214,13 @@ void zrandmemberCommand(client *c) {
         if (c->argc > 4 || (c->argc == 4 && strcasecmp(szFromObj(c->argv[3]),"withscores"))) {
             addReplyErrorObject(c,shared.syntaxerr);
             return;
-        } else if (c->argc == 4)
+        } else if (c->argc == 4) {
             withscores = 1;
+            if (l < LONG_MIN/2 || l > LONG_MAX/2) {
+                addReplyError(c,"value is out of range");
+                return;
+            }
+        }
         zrandmemberWithCountCommand(c, l, withscores);
         return;
     }

src/tls.cpp

@@ -616,8 +616,9 @@ bool tlsValidateCertificateName(tls_connection* conn){
     /* If neither the CN nor the SANs match, update the SSL error and return false */
     conn->c.last_errno = 0;
     if (conn->ssl_error) zfree(conn->ssl_error);
-    conn->ssl_error = (char*)zmalloc(512);
-    snprintf(conn->ssl_error, 512, "Client CN (%s) and SANs not found in allowlist.", commonName);
+    size_t bufsize = 512;
+    conn->ssl_error = (char*)zmalloc(bufsize);
+    snprintf(conn->ssl_error, bufsize, "Client CN (%s) and SANs not found in allowlist.", commonName);
     return false;
 }
 

src/ziplist.c

@@ -1704,17 +1704,17 @@ static unsigned char *createIntList() {
     unsigned char *zl = ziplistNew();
     char buf[32];
 
-    snprintf(buf, 32, "100");
+    snprintf(buf, sizeof(buf), "100");
     zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_TAIL);
-    snprintf(buf, 32, "128000");
+    snprintf(buf, sizeof(buf), "128000");
     zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_TAIL);
-    snprintf(buf, 32, "-100");
+    snprintf(buf, sizeof(buf), "-100");
     zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_HEAD);
-    snprintf(buf, 32, "4294967296");
+    snprintf(buf, sizeof(buf), "4294967296");
     zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_HEAD);
-    snprintf(buf, 32, "non integer");
+    snprintf(buf, sizeof(buf), "non integer");
     zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_TAIL);
-    snprintf(buf, 32, "much much longer non integer");
+    snprintf(buf, sizeof(buf), "much much longer non integer");
     zl = ziplistPush(zl, (unsigned char*)buf, strlen(buf), ZIPLIST_TAIL);
     return zl;
 }
@@ -2228,7 +2228,7 @@ int ziplistTest(int argc, char **argv, int accurate) {
         char buf[32];
         int i,len;
         for (i = 0; i < 1000; i++) {
-            len = snprintf(buf,32,"%d",i);
+            len = snprintf(buf,sizeof(buf),"%d",i);
             zl = ziplistPush(zl,(unsigned char*)buf,len,ZIPLIST_TAIL);
         }
         for (i = 0; i < 1000; i++) {
@@ -2375,13 +2375,13 @@ int ziplistTest(int argc, char **argv, int accurate) {
                 } else {
                     switch(rand() % 3) {
                     case 0:
-                        buflen = snprintf(buf,1024,"%lld",(0LL + rand()) >> 20);
+                        buflen = snprintf(buf,sizeof(buf),"%lld",(0LL + rand()) >> 20);
                         break;
                     case 1:
-                        buflen = snprintf(buf,1024,"%lld",(0LL + rand()));
+                        buflen = snprintf(buf,sizeof(buf),"%lld",(0LL + rand()));
                         break;
                     case 2:
-                        buflen = snprintf(buf,1024,"%lld",(0LL + rand()) << 20);
+                        buflen = snprintf(buf,sizeof(buf),"%lld",(0LL + rand()) << 20);
                         break;
                     default:
                         assert(NULL);
@@ -2410,7 +2410,7 @@ int ziplistTest(int argc, char **argv, int accurate) {
 
                 assert(ziplistGet(p,&sstr,&slen,&sval));
                 if (sstr == NULL) {
-                    buflen = snprintf(buf,1024,"%lld",sval);
+                    buflen = snprintf(buf,sizeof(buf),"%lld",sval);
                 } else {
                     buflen = slen;
                     memcpy(buf,sstr,buflen);

src/zmalloc.cpp

@@ -80,12 +80,12 @@ static_assert((PREFIX_SIZE % 16) == 0, "Our prefix must be modulo 16-bytes or ou
 #define realloc(ptr,size, type) tc_realloc(ptr,size)
 #define free(ptr) tc_free(ptr)
 #elif defined(USE_JEMALLOC)
-#define malloc(size, type) je_malloc(size)
-#define calloc(count,size,type) je_calloc(count,size)
-#define realloc(ptr,size,type) je_realloc(ptr,size)
-#define free(ptr) je_free(ptr)
-#define mallocx(size,flags) je_mallocx(size,flags)
-#define dallocx(ptr,flags) je_dallocx(ptr,flags)
+#define malloc(size, type) malloc(size)
+#define calloc(count,size,type) calloc(count,size)
+#define realloc(ptr,size,type) realloc(ptr,size)
+#define free(ptr) free(ptr)
+#define mallocx(size,flags) mallocx(size,flags)
+#define dallocx(ptr,flags) dallocx(ptr,flags)
 #else
 #define malloc(size, type) malloc(size)
 #define calloc(count,size,type) calloc(count,size)
@@ -370,7 +370,7 @@ size_t zmalloc_get_rss(void) {
     int fd, count;
     char *p, *x;
 
-    snprintf(filename,256,"/proc/%ld/stat",(long) getpid());
+    snprintf(filename,sizeof(filename),"/proc/%ld/stat",(long) getpid());
     if ((fd = open(filename,O_RDONLY)) == -1) return 0;
     if (read(fd,buf,4096) <= 0) {
         close(fd);
@@ -462,7 +462,7 @@ size_t zmalloc_get_rss(void) {
     char filename[256];
     int fd;
 
-    snprintf(filename,256,"/proc/%ld/psinfo",(long) getpid());
+    snprintf(filename,sizeof(filename),"/proc/%ld/psinfo",(long) getpid());
 
     if ((fd = open(filename,O_RDONLY)) == -1) return 0;
     if (ioctl(fd, PIOCPSINFO, &info) == -1) {
@@ -494,17 +494,17 @@ int zmalloc_get_allocator_info(size_t *allocated,
     *allocated = *resident = *active = 0;
     /* Update the statistics cached by mallctl. */
     sz = sizeof(epoch);
-    je_mallctl("epoch", &epoch, &sz, &epoch, sz);
+    mallctl("epoch", &epoch, &sz, &epoch, sz);
     sz = sizeof(size_t);
     /* Unlike RSS, this does not include RSS from shared libraries and other non
      * heap mappings. */
-    je_mallctl("stats.resident", resident, &sz, NULL, 0);
+    mallctl("stats.resident", resident, &sz, NULL, 0);
     /* Unlike resident, this doesn't not include the pages jemalloc reserves
      * for re-use (purge will clean that). */
-    je_mallctl("stats.active", active, &sz, NULL, 0);
+    mallctl("stats.active", active, &sz, NULL, 0);
     /* Unlike zmalloc_used_memory, this matches the stats.resident by taking
      * into account all allocations done by this process (not only zmalloc). */
-    je_mallctl("stats.allocated", allocated, &sz, NULL, 0);
+    mallctl("stats.allocated", allocated, &sz, NULL, 0);
     return 1;
 }
 
@@ -512,7 +512,7 @@ void set_jemalloc_bg_thread(int enable) {
     /* let jemalloc do purging asynchronously, required when there's no traffic 
      * after flushdb */
     char val = !!enable;
-    je_mallctl("background_thread", NULL, 0, &val, 1);
+    mallctl("background_thread", NULL, 0, &val, 1);
 }
 
 int jemalloc_purge() {
@@ -520,9 +520,9 @@ int jemalloc_purge() {
     char tmp[32];
     unsigned narenas = 0;
     size_t sz = sizeof(unsigned);
-    if (!je_mallctl("arenas.narenas", &narenas, &sz, NULL, 0)) {
-        snprintf(tmp, 32, "arena.%d.purge", narenas);
-        if (!je_mallctl(tmp, NULL, 0, NULL, 0))
+    if (!mallctl("arenas.narenas", &narenas, &sz, NULL, 0)) {
+        snprintf(tmp, sizeof(tmp), "arena.%d.purge", narenas);
+        if (!mallctl(tmp, NULL, 0, NULL, 0))
             return 0;
     }
     return -1;

src/zmalloc.h

@@ -55,7 +55,7 @@
 #include <jemalloc/jemalloc.h>
 #if (JEMALLOC_VERSION_MAJOR == 2 && JEMALLOC_VERSION_MINOR >= 1) || (JEMALLOC_VERSION_MAJOR > 2)
 #define HAVE_MALLOC_SIZE 1
-#define zmalloc_size(p) je_malloc_usable_size(p)
+#define zmalloc_size(p) malloc_usable_size(p)
 #else
 #error "Newer version of jemalloc required"
 #endif

tests/modules/Makefile

@@ -39,7 +39,8 @@ TEST_MODULES = \
     defragtest.so \
     hash.so \
     zset.so \
-    stream.so
+    stream.so \
+    load.so
 
 
 .PHONY: all

tests/modules/hooks.c

@@ -212,6 +212,7 @@ void loadingCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void
         case REDISMODULE_SUBEVENT_LOADING_RDB_START: keyname = "loading-rdb-start"; break;
         case REDISMODULE_SUBEVENT_LOADING_AOF_START: keyname = "loading-aof-start"; break;
         case REDISMODULE_SUBEVENT_LOADING_REPL_START: keyname = "loading-repl-start"; break;
+        case REDISMODULE_SUBEVENT_LOADING_FLASH_START: keyname = "loading-flash-start"; break;
         case REDISMODULE_SUBEVENT_LOADING_ENDED: keyname = "loading-end"; break;
         case REDISMODULE_SUBEVENT_LOADING_FAILED: keyname = "loading-failed"; break;
     }

tests/modules/load.c

@@ -0,0 +1,94 @@
+/* Server hooks API example
+ *
+ * -----------------------------------------------------------------------------
+ *
+ * Copyright (c) 2019, Salvatore Sanfilippo <antirez at gmail dot com>
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *
+ *   * Redistributions of source code must retain the above copyright notice,
+ *     this list of conditions and the following disclaimer.
+ *   * Redistributions in binary form must reproduce the above copyright
+ *     notice, this list of conditions and the following disclaimer in the
+ *     documentation and/or other materials provided with the distribution.
+ *   * Neither the name of Redis nor the names of its contributors may be used
+ *     to endorse or promote products derived from this software without
+ *     specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+ * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
+ * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+ * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+ * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+ * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+ * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+ * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+ * POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#define REDISMODULE_EXPERIMENTAL_API
+#include "redismodule.h"
+
+size_t count, finalCount;
+
+/* Client state change callback. */
+void loadCallback(RedisModuleCtx *ctx, RedisModuleEvent e, uint64_t sub, void *data) {
+    REDISMODULE_NOT_USED(ctx);
+    REDISMODULE_NOT_USED(e);
+    REDISMODULE_NOT_USED(data);
+
+    if (sub == REDISMODULE_SUBEVENT_LOADING_FLASH_START || sub == REDISMODULE_SUBEVENT_LOADING_RDB_START || sub == REDISMODULE_SUBEVENT_LOADING_AOF_START || sub == REDISMODULE_SUBEVENT_LOADING_REPL_START) {
+        count = 0;
+        finalCount = 0;
+    } else if (sub == REDISMODULE_SUBEVENT_LOADING_ENDED) {
+        finalCount = count;
+    }
+}
+
+int loadKeyCallback(RedisModuleCtx *ctx, int type,  const char *event, RedisModuleString *key) {
+    REDISMODULE_NOT_USED(ctx);
+    REDISMODULE_NOT_USED(type);
+    REDISMODULE_NOT_USED(event);
+
+    const char *keyname = RedisModule_StringPtrLen(key, NULL);
+
+    RedisModule_Log(ctx, REDISMODULE_LOGLEVEL_NOTICE, "Loaded key: %s", keyname);
+
+    count++;
+    return 0;
+}
+
+int LoadCount_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    REDISMODULE_NOT_USED(argv);
+    RedisModule_AutoMemory(ctx); /* Use automatic memory management. */
+
+    if (argc != 1) return RedisModule_WrongArity(ctx);
+
+    RedisModule_ReplyWithLongLong(ctx, finalCount);
+
+    return REDISMODULE_OK;
+}
+
+/* This function must be present on each Redis module. It is used in order to
+ * register the commands into the Redis server. */
+int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    REDISMODULE_NOT_USED(argv);
+    REDISMODULE_NOT_USED(argc);
+
+    if (RedisModule_Init(ctx,"load",1,REDISMODULE_APIVER_1)
+        == REDISMODULE_ERR) return REDISMODULE_ERR;
+
+    RedisModule_SubscribeToServerEvent(ctx,
+        RedisModuleEvent_Loading, loadCallback);
+    RedisModule_SubscribeToKeyspaceEvents(ctx,
+        REDISMODULE_NOTIFY_LOADED, loadKeyCallback);
+
+    if (RedisModule_CreateCommand(ctx, "load.count",
+        LoadCount_RedisCommand,"readonly",1,1,1) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+    return REDISMODULE_OK;
+}

tests/support/keydb.tcl

@@ -66,6 +66,33 @@ proc redis {{server 127.0.0.1} {port 6379} {defer 0} {tls 0} {tlsoptions {}} {re
     interp alias {} ::redis::redisHandle$id {} ::redis::__dispatch__ $id
 }
 
+# On recent versions of tcl-tls/OpenSSL, reading from a dropped connection
+# results with an error we need to catch and mimic the old behavior.
+proc ::redis::redis_safe_read {fd len} {
+    if {$len == -1} {
+        set err [catch {set val [read $fd]} msg]
+    } else {
+        set err [catch {set val [read $fd $len]} msg]
+    }
+    if {!$err} {
+        return $val
+    }
+    if {[string match "*connection abort*" $msg]} {
+        return {}
+    }
+    error $msg
+}
+
+proc ::redis::redis_safe_gets {fd} {
+    if {[catch {set val [gets $fd]} msg]} {
+        if {[string match "*connection abort*" $msg]} {
+            return {}
+        }
+        error $msg
+    }
+    return $val
+}
+
 # This is a wrapper to the actual dispatching procedure that handles
 # reconnection if needed.
 proc ::redis::__dispatch__ {id method args} {
@@ -146,6 +173,10 @@ proc ::redis::__method__read {id fd} {
     ::redis::redis_read_reply $id $fd
 }
 
+proc ::redis::__method__rawread {id fd {len -1}} {
+    return [redis_safe_read $fd $len]
+}
+
 proc ::redis::__method__write {id fd buf} {
     ::redis::redis_write $fd $buf
 }
@@ -192,8 +223,8 @@ proc ::redis::redis_writenl {fd buf} {
 }
 
 proc ::redis::redis_readnl {fd len} {
-    set buf [read $fd $len]
-    read $fd 2 ; # discard CR LF
+    set buf [redis_safe_read $fd $len]
+    redis_safe_read $fd 2 ; # discard CR LF
     return $buf
 }
 
@@ -239,11 +270,11 @@ proc ::redis::redis_read_map {id fd} {
 }
 
 proc ::redis::redis_read_line fd {
-    string trim [gets $fd]
+    string trim [redis_safe_gets $fd]
 }
 
 proc ::redis::redis_read_null fd {
-    gets $fd
+    redis_safe_gets $fd
     return {}
 }
 
@@ -260,7 +291,7 @@ proc ::redis::redis_read_reply {id fd} {
     }
 
     while {1} {
-        set type [read $fd 1]
+        set type [redis_safe_read $fd 1]
         switch -exact -- $type {
             _ {return [redis_read_null $fd]}
             : -

tests/support/test.tcl

@@ -40,6 +40,12 @@ proc assert_failed {expected_err detail} {
      error "assertion:$expected_err $detail"
 }
 
+proc assert_not_equal {value expected {detail ""}} {
+    if {!($expected ne $value)} {
+        assert_failed "Expected '$value' not equal to '$expected'" $detail
+    }
+}
+
 proc assert_equal {value expected {detail ""}} {
     if {$expected ne $value} {
         assert_failed "Expected '$value' to be equal to '$expected'" $detail

tests/unit/introspection.tcl

@@ -162,15 +162,16 @@ 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
+            db-s3-object
         }
 
         if {!$::tls} {

tests/unit/moduleapi/load.tcl

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

tests/unit/obuf-limits.tcl

@@ -111,7 +111,7 @@ start_server {tags {"obuf-limits"} overrides { server-threads 1 }} {
 
         # Read nothing
         set fd [$rd channel]
-        assert_equal {} [read $fd]
+        assert_equal {} [$rd rawread]
     }
 
     # Note: This test assumes that what's written with one write, will be read by redis in one read.
@@ -151,8 +151,7 @@ start_server {tags {"obuf-limits"} overrides { server-threads 1 }} {
         assert_equal "PONG" [r ping]
         set clients [r client list]
         assert_no_match "*name=multicommands*" $clients
-        set fd [$rd2 channel]
-        assert_equal {} [read $fd]
+        assert_equal {} [$rd2 rawread]
     }
 
     test {Execute transactions completely even if client output buffer limit is enforced} {

tests/unit/scripting.tcl

@@ -43,6 +43,10 @@ start_server {tags {"scripting"}} {
         r eval {return redis.call('get',KEYS[1])} 1 mykey
     } {myval}
 
+    test { EVAL - keys command works? } {
+        r eval {return redis.call('keys', 'test')} 0
+    }
+
     test {EVALSHA - Can we call a SHA1 if already defined?} {
         r evalsha fd758d1589d044dd850a6f05d52f2eefd27f033f 1 mykey
     } {myval}

tests/unit/sort.tcl

@@ -315,4 +315,15 @@ start_server {
             }
         }
     }
+
+    test {SETRANGE with huge offset} {
+        r lpush L 2 1 0
+        # expecting a different outcome on 32 and 64 bit systems
+        foreach value {9223372036854775807 2147483647} {
+            catch {[r sort_ro L by a limit 2 $value]} res
+            if {![string match "2" $res] && ![string match "*out of range*" $res]} {
+                assert_not_equal $res "expecting an error or 2"
+            }
+        }
+    }
 }

tests/unit/type/hash.tcl

@@ -68,6 +68,11 @@ start_server {tags {"hash"}} {
         r hrandfield myhash 0
     } {}
 
+    test "HRANDFIELD count overflow" {
+        r hmset myhash a 1
+        assert_error {*value is out of range*} {r hrandfield myhash -9223372036854770000 withvalues}
+    } {}
+
     test "HRANDFIELD with <count> against non existing key" {
         r hrandfield nonexisting_key 100
     } {}

tests/unit/type/string.tcl

@@ -574,4 +574,14 @@ start_server {tags {"string"}} {
     test {LCS indexes with match len and minimum match len} {
         dict get [r STRALGO LCS IDX KEYS virus1 virus2 WITHMATCHLEN MINMATCHLEN 5] matches
     } {{{1 222} {13 234} 222}}
+
+    test {SETRANGE with huge offset} {
+        foreach value {9223372036854775807 2147483647} {
+            catch {[r setrange K $value A]} res
+            # expecting a different error on 32 and 64 bit systems
+            if {![string match "*string exceeds maximum allowed size*" $res] && ![string match "*out of range*" $res]} {
+                assert_equal $res "expecting an error"
+           }
+        }
+    }
 }

tests/unit/type/zset.tcl

@@ -1714,6 +1714,11 @@ start_server {tags {"zset"}} {
         r zrandmember nonexisting_key 100
     } {}
 
+    test "ZRANDMEMBER count overflow" {
+        r zadd myzset 0 a
+        assert_error {*value is out of range*} {r zrandmember myzset -9223372036854770000 withscores}
+    } {}
+
     # Make sure we can distinguish between an empty array and a null response
     r readraw 1