[BUG] Moves key to another DB, the source key was removed if the move failed due to the key exists in the destination db #497 (#498)

Co-authored-by: Paul Chen <mingchen@Mings-MacBook-Pro.local>

.github/workflows/ci.yml

@@ -5,17 +5,14 @@ on: [push, pull_request]
 jobs:
 
   test-ubuntu-latest:
-    runs-on: self-hosted
+    runs-on: ubuntu-latest
     steps:
-    - uses: actions/checkout@v1
-      with:
-        submodules: true
+    - uses: actions/checkout@v2
     - name: make
       run: |
           sudo apt-get update
-          sudo apt-get -y remove libzstd || true
-          sudo apt-get -y install uuid-dev libcurl4-openssl-dev libbz2-dev zlib1g-dev libsnappy-dev liblz4-dev libzstd-dev libgflags-dev
-          make BUILD_TLS=yes -j2 KEYDB_CFLAGS='-Werror' KEYDB_CXXFLAGS='-Werror'
+          sudo apt-get -y install uuid-dev libcurl4-openssl-dev
+          make KEYDB_CFLAGS='-Werror' KEYDB_CXXFLAGS='-Werror' BUILD_TLS=yes -j2
     - name: gen-cert
       run: ./utils/gen-test-certs.sh
     - name: test-tls
@@ -31,9 +28,6 @@ jobs:
     - name: module tests
       run: |
           ./runtest-moduleapi
-    - name: rotation test
-      run: |
-          ./runtest-rotation
         
   build-ubuntu-old:
     runs-on: ubuntu-18.04
@@ -44,7 +38,6 @@ jobs:
         sudo apt-get update
         sudo apt-get -y install uuid-dev libcurl4-openssl-dev
         make -j2
-
   build-macos-latest:
     runs-on: macos-latest
     steps:
@@ -56,4 +49,18 @@ jobs:
     runs-on: ubuntu-latest
     steps:
     - uses: actions/checkout@v2
-    - name: make
+    - name: make
+      run: |
+          sudo apt-get update
+          sudo apt-get -y install uuid-dev libcurl4-openssl-dev
+          make KEYDB_CFLAGS='-Werror' KEYDB_CXXFLAGS='-Werror' MALLOC=libc -j2
+  build-ubuntu-32bit:
+    runs-on: ubuntu-latest
+    steps:
+    - uses: actions/checkout@v2
+    - name: make
+      run: |
+          sudo dpkg --add-architecture i386
+          sudo apt-get update
+          sudo apt-get -y install gcc-multilib g++-multilib libc6-dev-i386 lib32z1 uuid-dev:i386 libcurl4-openssl-dev:i386
+          make KEYDB_CFLAGS='-Werror' KEYDB_CXXFLAGS='-Werror' 32bit -j2

COPYING

@@ -1,12 +1,7 @@
-Code in this repository is non-free.  Portions of this code include or 
-are derivate works of code published under the BSD 3-clause license.  The
-license below applies ONLY TO THOSE PORTIONS.  Code authored by employees 
-or contractors of EQ Alpha Technology are not licensed for use without express
-written permission of EQ Alpha Technology.  All rights are reserved.
-
 Copyright (c) 2006-2020, Salvatore Sanfilippo
 Copyright (C) 2019-2021, John Sully
 Copyright (C) 2020-2021, EQ Alpha Technology Ltd.
+Copyright (C) 2022 Snap Inc.
 All rights reserved.
 
 Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met:

README.md

@@ -99,7 +99,7 @@ Avoid forwarding RREPLAY messages to other masters? WARNING: This setting is dan
 
 
 ```
-    scratch-file-path /path
+    db-s3-object /path/to/bucket
 ```
 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.
 

pkg/deb/deb-buildsource.sh

@@ -23,8 +23,8 @@ pkg_name=keydb-$majorv:$version$distname
 
 # create build tree
 cd ../../../
-tar -czvf keydb_$version.orig.tar.gz --force-local keydb-internal
-cd keydb-internal/pkg/deb/
+tar -czvf keydb_$version.orig.tar.gz --force-local KeyDB
+cd KeyDB/pkg/deb/
 mkdir -p $pkg_name/tmp
 if [[ "$codename" == "xenial" ]] || [[ "$codename" == "stretch" ]]; then
 	cp -r debian_dh9 $pkg_name/tmp/debian

pkg/deb/debian/control

@@ -26,8 +26,8 @@ Build-Depends:
  libzstd-dev <!nocheck>
 Standards-Version: 4.2.1
 Homepage: https://docs.keydb.dev/
-Vcs-Git: https://github.com/EQ-Alpha/KeyDB.git
-Vcs-Browser: https://github.com/EQ-Alpha/KeyDB
+Vcs-Git: https://github.com/Snapchat/KeyDB.git
+Vcs-Browser: https://github.com/Snapchat/KeyDB
 
 Package: keydb
 Architecture: all

pkg/deb/debian_dh9/control

@@ -25,8 +25,8 @@ Build-Depends:
  libzstd-dev <!nocheck>
 Standards-Version: 4.2.1
 Homepage: https://docs.keydb.dev/
-Vcs-Git: https://github.com/EQ-Alpha/KeyDB.git
-Vcs-Browser: https://github.com/EQ-Alpha/KeyDB
+Vcs-Git: https://github.com/Snapchat/KeyDB.git
+Vcs-Browser: https://github.com/Snapchat/KeyDB
 
 Package: keydb
 Architecture: all

pkg/deb/master_changelog

@@ -1,3 +1,25 @@
+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
+  * Issues fixed: #419, #422, #428
+  * PRs: #426, #429, #431, #433
+
+-- Ben Schermel <ben@eqalpha.com>  Mon, 23 May 2022 20:00:37 +0000
+
+keydb (6:6.3.0-1distribution_placeholder) codename_placeholder; urgency=medium
+
+  * This release open sources KeyDB Enterprise features into the open source project along with PSYNC for active replication
+  * Partial synchronization for active replication is introduced
+  * MVCC introduced into codebase from KeyDB Enterprise
+  * Async commands added: GET, MGET. These will see perf improvements
+  * KEYS and SCAN commands will no longer be blocking calls
+  * Async Rehash implemented for additional stability to perf
+  * IStorage interface added
+  * In-process background saving (forkless) to comply with maxmemory setting
+  * See v6.3.0 tagged release notes on github for a detailed explanation of these changes
+
+-- Ben Schermel <ben@eqalpha.com>  Wed, 11 May 2022 20:00:37 +0000
+
 keydb (6:6.2.2-1distribution_placeholder) codename_placeholder; urgency=medium
 
   * Acquire lock in module.cpp to fix module test break

pkg/docker/Dockerfile

@@ -1,4 +1,4 @@
-FROM ubuntu:20.04
+FROM ubuntu:18.04
 SHELL ["/bin/bash","-c"]
 RUN groupadd -r keydb && useradd -r -g keydb keydb
 # use gosu for easy step-down from root: https://github.com/tianon/gosu/releases
@@ -77,7 +77,7 @@ RUN \
         mkdir /data && chown keydb:keydb /data; \
         mkdir /flash && chown keydb:keydb /flash; \
         mkdir -p /etc/keydb; \
-        cp /tmp/keydb-internal/keydb.conf /etc/keydb/; \
+        cp /tmp/KeyDB/keydb.conf /etc/keydb/; \
         sed -i 's/^\(daemonize .*\)$/# \1/' /etc/keydb/keydb.conf; \
         sed -i 's/^\(dir .*\)$/# \1\ndir \/data/' /etc/keydb/keydb.conf; \
         sed -i 's/^\(logfile .*\)$/# \1/' /etc/keydb/keydb.conf; \
@@ -110,8 +110,3 @@ ENV KEYDB_PRO_DIRECTORY=/usr/local/bin/
 ENTRYPOINT ["docker-entrypoint.sh"]
 EXPOSE 6379
 CMD ["keydb-server","/etc/keydb/keydb.conf"]
-
-
-
-
-

pkg/docker/Dockerfile_Alpine

@@ -0,0 +1,77 @@
+FROM alpine:3.12
+# add our user and group first to make sure their IDs get assigned consistently, regardless of whatever dependencies get added
+RUN addgroup -S -g 1000 keydb && adduser -S -G keydb -u 999 keydb
+RUN mkdir -p /etc/keydb
+ARG BRANCH
+RUN set -eux; \
+        \
+        apk add --no-cache su-exec; \
+        apk add --no-cache --virtual .build-deps \
+                coreutils \
+                gcc \
+                linux-headers \
+                make \
+                musl-dev \
+                openssl-dev \
+                git \
+                util-linux-dev \
+                curl-dev \
+                g++ \
+                libunwind-dev \
+                bash \
+                perl \
+                git \
+        ; \
+        cd /tmp && git clone --branch $BRANCH https://github.com/Snapchat/KeyDB.git --recursive; \
+        cd /tmp/KeyDB; \
+        # disable protected mode as it relates to docker
+        grep -E '^ *createBoolConfig[(]"protected-mode",.*, *1 *,.*[)],$' ./src/config.cpp; \
+        sed -ri 's!^( *createBoolConfig[(]"protected-mode",.*, *)1( *,.*[)],)$!\10\2!' ./src/config.cpp; \
+        grep -E '^ *createBoolConfig[(]"protected-mode",.*, *0 *,.*[)],$' ./src/config.cpp; \
+        make -j$(nproc) BUILD_TLS=yes; \
+        cd src; \
+        strip keydb-cli keydb-benchmark keydb-check-rdb keydb-check-aof keydb-diagnostic-tool keydb-sentinel keydb-server; \
+        mv keydb-server keydb-cli keydb-benchmark keydb-check-rdb keydb-check-aof keydb-diagnostic-tool keydb-sentinel /usr/local/bin/; \
+        runDeps="$( \
+                scanelf --needed --nobanner --format '%n#p' --recursive /usr/local \
+                        | tr ',' '\n' \
+                        | sort -u \
+                        | awk 'system("[ -e /usr/local/lib/" $1 " ]") == 0 { next } { print "so:" $1 }' \
+        )"; \
+        apk add --no-network --virtual .keydb-rundeps $runDeps; \
+        apk del --no-network .build-deps; \
+        # create working directories and organize files
+        mkdir /data && chown keydb:keydb /data; \
+        mkdir /flash && chown keydb:keydb /flash; \
+        mkdir -p /etc/keydb; \
+        cp /tmp/KeyDB/keydb.conf /etc/keydb/; \
+        sed -i 's/^\(daemonize .*\)$/# \1/' /etc/keydb/keydb.conf; \
+        sed -i 's/^\(dir .*\)$/# \1\ndir \/data/' /etc/keydb/keydb.conf; \
+        sed -i 's/^\(logfile .*\)$/# \1/' /etc/keydb/keydb.conf; \
+        sed -i 's/protected-mode yes/protected-mode no/g' /etc/keydb/keydb.conf; \
+        sed -i 's/^\(bind .*\)$/# \1/' /etc/keydb/keydb.conf; \
+        ln -s keydb-cli redis-cli; \
+        cd /etc/keydb; \
+        ln -s keydb.conf redis.conf; \
+        rm -rf /tmp/*
+# generate entrypoint script
+RUN set -eux; \
+        echo '#!/bin/sh' > /usr/local/bin/docker-entrypoint.sh; \
+        echo 'set -e' >> /usr/local/bin/docker-entrypoint.sh; \
+        echo "# first arg is '-f' or '--some-option'" >> /usr/local/bin/docker-entrypoint.sh; \
+        echo "# or first arg is `something.conf`" >> /usr/local/bin/docker-entrypoint.sh; \
+        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 "# 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; \
+        echo '        exec su-exec keydb "$0" "$@"' >> /usr/local/bin/docker-entrypoint.sh; \
+        echo 'fi' >> /usr/local/bin/docker-entrypoint.sh; \
+        echo 'exec "$@"' >> /usr/local/bin/docker-entrypoint.sh; \
+        chmod +x /usr/local/bin/docker-entrypoint.sh
+VOLUME /data
+WORKDIR /data
+ENTRYPOINT ["docker-entrypoint.sh"]
+EXPOSE 6379
+CMD ["keydb-server", "/etc/keydb/keydb.conf"]

src/StorageCache.cpp

@@ -203,6 +203,7 @@ void StorageCache::beginWriteBatch() {
 }
 
 void StorageCache::emergencyFreeCache() {
+    std::unique_lock<fastlock> ul(m_lock);
     dict *d = m_pdict;
     m_pdict = nullptr;
     if (d != nullptr) {

src/config.cpp

@@ -2782,7 +2782,7 @@ standardConfig configs[] = {
     createBoolConfig("disable-thp", NULL, MODIFIABLE_CONFIG, g_pserver->disable_thp, 1, NULL, NULL),
     createBoolConfig("cluster-allow-replica-migration", NULL, MODIFIABLE_CONFIG, g_pserver->cluster_allow_replica_migration, 1, NULL, NULL),
     createBoolConfig("replica-announced", NULL, MODIFIABLE_CONFIG, g_pserver->replica_announced, 1, NULL, NULL),
-    createBoolConfig("enable-async-commands", NULL, MODIFIABLE_CONFIG, g_pserver->enable_async_commands, 1, NULL, NULL),
+    createBoolConfig("enable-async-commands", NULL, MODIFIABLE_CONFIG, g_pserver->enable_async_commands, 0, NULL, NULL),
     createBoolConfig("multithread-load-enabled", NULL, MODIFIABLE_CONFIG, g_pserver->multithread_load_enabled, 0, NULL, NULL),
     createBoolConfig("active-client-balancing", NULL, MODIFIABLE_CONFIG, g_pserver->active_client_balancing, 1, NULL, NULL),
 

src/db.cpp

@@ -1568,6 +1568,12 @@ void moveCommand(client *c) {
         return;
     }
 
+    /* Return zero if the key already exists in the target DB */
+    if (lookupKeyWrite(dst,c->argv[1]) != NULL) {
+        addReply(c,shared.czero);
+        return;
+    }
+
     std::unique_ptr<expireEntry> spexpire;
     {   // scope pexpireOld
     std::unique_lock<fastlock> ul(g_expireLock);
@@ -1582,12 +1588,6 @@ void moveCommand(client *c) {
     dbDelete(src,c->argv[1]);
     g_pserver->dirty++;
 
-    /* Return zero if the key already exists in the target DB */
-    if (lookupKeyWrite(dst,c->argv[1]) != NULL) {
-        addReply(c,shared.czero);
-        decrRefCount(o);
-        return;
-    }
     dbAdd(dst,c->argv[1],o);
     if (spexpire != nullptr) setExpire(c,dst,c->argv[1],std::move(*spexpire));
 
@@ -2639,7 +2639,10 @@ bool redisDbPersistentData::insert(char *key, robj *o, bool fAssumeNew, dict_ite
         ensure(key);
     dictEntry *de;
     int res = dictAdd(m_pdict, key, o, &de);
-    serverAssert(FImplies(fAssumeNew, res == DICT_OK));
+    if (!FImplies(fAssumeNew, res == DICT_OK)) {
+        serverLog(LL_WARNING,
+            "Assumed new key %s existed in DB.", key);
+    }
     if (res == DICT_OK)
     {
 #ifdef CHECKED_BUILD
@@ -2669,6 +2672,11 @@ void redisDbPersistentData::prepOverwriteForSnapshot(char *key)
         auto itr = m_pdbSnapshot->find_cached_threadsafe(key);
         if (itr.key() != nullptr)
         {
+            if (itr.val()->FExpires()) {
+                // Note: I'm sure we could handle this, but its too risky at the moment.
+                //  There are known bugs doing this with expires
+                return;
+            }
             sds keyNew = sdsdupshared(itr.key());
             if (dictAdd(m_pdictTombstone, keyNew, (void*)dictHashKey(m_pdict, key)) != DICT_OK)
                 sdsfree(keyNew);
@@ -3263,10 +3271,11 @@ bool redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
                     dictEntry **table;
                     __atomic_load(&c->db->m_pdict->ht[iht].table, &table, __ATOMIC_RELAXED);
                     if (table != nullptr) {
-                        dictEntry *de = table[hT];
+                        dictEntry *de;
+                        __atomic_load(&table[hT], &de, __ATOMIC_ACQUIRE);
                         while (de != nullptr) {
                             _mm_prefetch(dictGetKey(de), _MM_HINT_T2);
-                            de = de->next;
+                            __atomic_load(&de->next, &de, __ATOMIC_ACQUIRE);
                         }
                     }
                     if (!dictIsRehashing(c->db->m_pdict))
@@ -3286,6 +3295,10 @@ bool redisDbPersistentData::prefetchKeysAsync(client *c, parsed_command &command
     auto cmd = lookupCommand(szFromObj(command.argv[0]));
     if (cmd == nullptr)
         return false; // Bad command? It's not for us to judge, just bail
+    
+    if (command.argc < std::abs(cmd->arity))
+        return false; // Invalid number of args
+    
     int numkeys = getKeysFromCommand(cmd, command.argv, command.argc, &result);
     for (int ikey = 0; ikey < numkeys; ++ikey)
     {

src/dict.cpp

@@ -128,6 +128,7 @@ int _dictInit(dict *d, dictType *type,
     d->pauserehash = 0;
     d->asyncdata = nullptr;
     d->refcount = 1;
+    d->noshrink = false;
     return DICT_OK;
 }
 
@@ -204,7 +205,7 @@ int dictMerge(dict *dst, dict *src)
     
     if (dictSize(dst) == 0)
     {
-        std::swap(*dst, *src);
+        dict::swap(*dst, *src);
         std::swap(dst->pauserehash, src->pauserehash);
         return DICT_OK;
     }
@@ -212,7 +213,7 @@ int dictMerge(dict *dst, dict *src)
     size_t expectedSize = dictSize(src) + dictSize(dst);
     if (dictSize(src) > dictSize(dst) && src->asyncdata == nullptr && dst->asyncdata == nullptr)
     {
-        std::swap(*dst, *src);
+        dict::swap(*dst, *src);
         std::swap(dst->pauserehash, src->pauserehash);
     }
 
@@ -402,7 +403,7 @@ int dictRehash(dict *d, int n) {
 
 dictAsyncRehashCtl::dictAsyncRehashCtl(struct dict *d, dictAsyncRehashCtl *next) : dict(d), next(next) {
     queue.reserve(c_targetQueueSize);
-    __atomic_fetch_add(&d->refcount, 1, __ATOMIC_RELEASE);
+    __atomic_fetch_add(&d->refcount, 1, __ATOMIC_ACQ_REL);
     this->rehashIdxBase = d->rehashidx;
 }
 
@@ -446,6 +447,9 @@ LDone:
 }
 
 void dictRehashAsync(dictAsyncRehashCtl *ctl) {
+    if (ctl->abondon.load(std::memory_order_acquire)) {
+        ctl->hashIdx = ctl->queue.size();
+    }
     for (size_t idx = ctl->hashIdx; idx < ctl->queue.size(); ++idx) {
         auto &wi = ctl->queue[idx];
         wi.hash = dictHashKey(ctl->dict, dictGetKey(wi.de));
@@ -455,6 +459,9 @@ void dictRehashAsync(dictAsyncRehashCtl *ctl) {
 }
 
 bool dictRehashSomeAsync(dictAsyncRehashCtl *ctl, size_t hashes) {
+    if (ctl->abondon.load(std::memory_order_acquire)) {
+        ctl->hashIdx = ctl->queue.size();
+    }
     size_t max = std::min(ctl->hashIdx + hashes, ctl->queue.size());
     for (; ctl->hashIdx < max; ++ctl->hashIdx) {
         auto &wi = ctl->queue[ctl->hashIdx];
@@ -465,6 +472,23 @@ bool dictRehashSomeAsync(dictAsyncRehashCtl *ctl, size_t hashes) {
     return ctl->hashIdx < ctl->queue.size();
 }
 
+
+void discontinueAsyncRehash(dict *d) {
+    // We inform our async rehashers and the completion function the results are to be
+    //  abandoned.  We keep the asyncdata linked in so that dictEntry's are still added
+    //  to the GC list.  This is because we can't gurantee when the other threads will
+    //  stop looking at them.
+    if (d->asyncdata != nullptr) {
+        auto adata = d->asyncdata;
+        while (adata != nullptr && !adata->abondon.load(std::memory_order_relaxed)) {
+            adata->abondon = true;
+            adata = adata->next;
+        }
+        if (dictIsRehashing(d))
+            d->rehashidx = 0;
+    }
+}
+
 void dictCompleteRehashAsync(dictAsyncRehashCtl *ctl, bool fFree) {
     dict *d = ctl->dict;
     assert(ctl->done);
@@ -786,6 +810,8 @@ int _dictClear(dict *d, dictht *ht, void(callback)(void *)) {
         if (callback && (i & 65535) == 0) callback(d->privdata);
 
         if ((he = ht->table[i]) == NULL) continue;
+        dictEntry *deNull = nullptr;
+        __atomic_store(&ht->table[i], &deNull, __ATOMIC_RELEASE);
         while(he) {
             nextHe = he->next;
             if (d->asyncdata && (ssize_t)i < d->rehashidx) {

src/dict.h

@@ -110,12 +110,16 @@ struct dictAsyncRehashCtl {
     std::atomic<bool> abondon { false };
 
     dictAsyncRehashCtl(struct dict *d, dictAsyncRehashCtl *next);
+    dictAsyncRehashCtl(const dictAsyncRehashCtl&) = delete;
+    dictAsyncRehashCtl(dictAsyncRehashCtl&&) = delete;
     ~dictAsyncRehashCtl();
 };
 #else
 struct  dictAsyncRehashCtl;
 #endif
 
+void discontinueAsyncRehash(dict *d);
+
 typedef struct dict {
     dictType *type;
     void *privdata;
@@ -125,6 +129,24 @@ typedef struct dict {
     dictAsyncRehashCtl *asyncdata;
     int16_t pauserehash; /* If >0 rehashing is paused (<0 indicates coding error) */
     uint8_t noshrink = false;
+
+#ifdef __cplusplus
+    dict() = default;
+    dict(dict &) = delete;  // No Copy Ctor
+
+    static void swap(dict& a, dict& b) {
+        discontinueAsyncRehash(&a);
+        discontinueAsyncRehash(&b);
+        std::swap(a.type, b.type);
+        std::swap(a.privdata, b.privdata);
+        std::swap(a.ht[0], b.ht[0]);
+        std::swap(a.ht[1], b.ht[1]);
+        std::swap(a.rehashidx, b.rehashidx);
+        // Never swap refcount - they are attached to the specific dict obj
+        std::swap(a.pauserehash, b.pauserehash);
+        std::swap(a.noshrink, b.noshrink);
+    }
+#endif
 } dict;
 
 /* If safe is set to 1 this is a safe iterator, that means, you can call

src/gc.h

@@ -2,6 +2,7 @@
 #include <vector>
 #include <assert.h>
 #include <unordered_set>
+#include <list>
 
 struct ICollectable
 {
@@ -15,7 +16,17 @@ class GarbageCollector
     struct EpochHolder
     {
         uint64_t tstamp;
-        std::vector<std::unique_ptr<T>> m_vecObjs;
+        std::unique_ptr<std::vector<std::unique_ptr<T>>> m_spvecObjs;
+
+        EpochHolder() {
+            m_spvecObjs = std::make_unique<std::vector<std::unique_ptr<T>>>();
+        }
+
+        // Support move operators
+        EpochHolder(EpochHolder &&other) = default;
+        EpochHolder &operator=(EpochHolder &&) = default;
+
+
 
         bool operator<(uint64_t tstamp) const
         {
@@ -45,14 +56,14 @@ public:
     void shutdown()
     {
         std::unique_lock<fastlock> lock(m_lock);
-        m_vecepochs.clear();
+        m_listepochs.clear();
         m_setepochOutstanding.clear();
     }
 
     bool empty() const
     {
         std::unique_lock<fastlock> lock(m_lock);
-        return m_vecepochs.empty();
+        return m_listepochs.empty();
     }
 
     void endEpoch(uint64_t epoch, bool fNoFree = false)
@@ -63,12 +74,12 @@ public:
         m_setepochOutstanding.erase(epoch);
         if (fNoFree)
             return;
-        std::vector<EpochHolder> vecclean;
+        std::list<EpochHolder> listclean;
         
         // No outstanding epochs?
         if (m_setepochOutstanding.empty())
         {
-            vecclean = std::move(m_vecepochs);  // Everything goes!
+            listclean = std::move(m_listepochs);  // Everything goes!
         }
         else
         {
@@ -77,18 +88,20 @@ public:
                 return; // No available epochs to free
 
             // Clean any epochs available (after the lock)
-            for (size_t iepoch = 0; iepoch < m_vecepochs.size(); ++iepoch)
+            for (auto itr = m_listepochs.begin(); itr != m_listepochs.end(); /* itr incremented in loop*/)
             {
-                auto &e = m_vecepochs[iepoch];
+                auto &e = *itr;
+                auto itrNext = itr;
+                ++itrNext;
                 if (e < minepoch)
                 {
-                    vecclean.emplace_back(std::move(e));
-                    m_vecepochs.erase(m_vecepochs.begin() + iepoch);
-                    --iepoch;
+                    listclean.emplace_back(std::move(e));
+                    m_listepochs.erase(itr);
                 }
+                itr = itrNext;
             }
 
-            assert(vecclean.empty() || fMinElement);
+            assert(listclean.empty() || fMinElement);
         }
 
         lock.unlock();  // don't hold it for the potentially long delete of vecclean
@@ -100,24 +113,24 @@ public:
         serverAssert(m_setepochOutstanding.find(epoch) != m_setepochOutstanding.end());
         serverAssert(sp->FWillFreeChildDebug() == false);
 
-        auto itr = std::find(m_vecepochs.begin(), m_vecepochs.end(), m_epochNext+1);
-        if (itr == m_vecepochs.end())
+        auto itr = std::find(m_listepochs.begin(), m_listepochs.end(), m_epochNext+1);
+        if (itr == m_listepochs.end())
         {
             EpochHolder e;
             e.tstamp = m_epochNext+1;
-            e.m_vecObjs.push_back(std::move(sp));
-            m_vecepochs.emplace_back(std::move(e));
+            e.m_spvecObjs->push_back(std::move(sp));
+            m_listepochs.emplace_back(std::move(e));
         }
         else
         {
-            itr->m_vecObjs.push_back(std::move(sp));
+            itr->m_spvecObjs->push_back(std::move(sp));
         }
     }
 
 private:
     mutable fastlock m_lock { "Garbage Collector"};
 
-    std::vector<EpochHolder> m_vecepochs;
+    std::list<EpochHolder> m_listepochs;
     std::unordered_set<uint64_t> m_setepochOutstanding;
     uint64_t m_epochNext = 0;
-};
+};

src/networking.cpp

@@ -1810,6 +1810,8 @@ int writeToClient(client *c, int handler_installed) {
        is a replica, so only attempt to do so if that's the case. */
     if (c->flags & CLIENT_SLAVE && !(c->flags & CLIENT_MONITOR) && c->replstate == SLAVE_STATE_ONLINE) {
         std::unique_lock<fastlock> repl_backlog_lock (g_pserver->repl_backlog_lock);
+        // Ensure all writes to the repl backlog are visible
+        std::atomic_thread_fence(std::memory_order_acquire);
 
         while (clientHasPendingReplies(c)) {
             long long repl_end_idx = getReplIndexFromOffset(c->repl_end_off);
@@ -1983,9 +1985,9 @@ void ProcessPendingAsyncWrites()
          * writes may have been signalled without having been copied to the replyAsync buffer,
          * thus causing the buffer to be NULL */ 
         if (c->replyAsync != nullptr){
-            int size = c->replyAsync->used;
+            size_t size = c->replyAsync->used;
 
-            if (listLength(c->reply) == 0 && size <= (PROTO_REPLY_CHUNK_BYTES - c->bufpos)) {
+            if (listLength(c->reply) == 0 && size <= static_cast<size_t>(PROTO_REPLY_CHUNK_BYTES - c->bufpos)) {
                 memcpy(c->buf + c->bufpos, c->replyAsync->buf(), size);
                 c->bufpos += size;
             } else {
@@ -2077,8 +2079,6 @@ int handleClientsWithPendingWrites(int iel, int aof_state) {
         * that may trigger write error or recreate handler. */
         if ((flags & CLIENT_PROTECTED) && !(flags & CLIENT_SLAVE)) continue;
 
-        //std::unique_lock<decltype(c->lock)> lock(c->lock);
-
         /* Don't write to clients that are going to be closed anyway. */
         if (c->flags & CLIENT_CLOSE_ASAP) continue;
 
@@ -2096,6 +2096,7 @@ int handleClientsWithPendingWrites(int iel, int aof_state) {
 
         /* If after the synchronous writes above we still have data to
         * output to the client, we need to install the writable handler. */
+        std::unique_lock<decltype(c->lock)> lock(c->lock);
         if (clientHasPendingReplies(c)) {
             if (connSetWriteHandlerWithBarrier(c->conn, sendReplyToClient, ae_flags, true) == C_ERR) {
                 freeClientAsync(c);
@@ -2742,9 +2743,10 @@ void readQueryFromClient(connection *conn) {
         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)) {
             // 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
+            //  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
             bool fSnapshotExists = c->db->mvccLastSnapshot >= c->mvccCheckpoint;
-            bool fWriteTooRecent = (((getMvccTstamp() - c->mvccCheckpoint) >> MVCC_MS_SHIFT) < static_cast<uint64_t>(g_pserver->snapshot_slip)/2);
+            bool fWriteTooRecent = !g_fTestMode && (((getMvccTstamp() - c->mvccCheckpoint) >> MVCC_MS_SHIFT) < static_cast<uint64_t>(g_pserver->snapshot_slip)/2);
 
             // The check below avoids running async commands if this is a frequent writer unless a snapshot is already there to service it
             if (!fWriteTooRecent || fSnapshotExists) {

src/object.cpp

@@ -1091,7 +1091,7 @@ struct redisMemOverhead *getMemoryOverheadData(void) {
     mem_total += g_pserver->initial_memory_usage;
 
     mem = 0;
-    if (g_pserver->repl_backlog)
+    if (g_pserver->repl_backlog && g_pserver->repl_backlog != g_pserver->repl_backlog_disk)
         mem += zmalloc_size(g_pserver->repl_backlog);
     mh->repl_backlog = mem;
     mem_total += mem;

src/rdb.cpp

@@ -1657,13 +1657,18 @@ int launchRdbSaveThread(pthread_t &child, rdbSaveInfo *rsi)
 
         g_pserver->rdbThreadVars.tmpfileNum++;
         g_pserver->rdbThreadVars.fRdbThreadCancel = false;
-        if (pthread_create(&child, NULL, rdbSaveThread, args)) {
+        pthread_attr_t tattr;
+        pthread_attr_init(&tattr);
+        pthread_attr_setstacksize(&tattr, 1 << 23); // 8 MB
+        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);
             return C_ERR;
         }
+        pthread_attr_destroy(&tattr);
         g_pserver->child_type = CHILD_TYPE_RDB;
     }
     return C_OK;
@@ -3049,7 +3054,9 @@ void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) {
         (r->keys_since_last_callback >= g_pserver->loading_process_events_interval_keys)))
     {
         rdbAsyncWorkThread *pwthread = reinterpret_cast<rdbAsyncWorkThread*>(r->chksum_arg);
-        bool fUpdateReplication = (g_pserver->mstime - r->last_update) > 1000;
+        mstime_t mstime;
+        __atomic_load(&g_pserver->mstime, &mstime, __ATOMIC_RELAXED);
+        bool fUpdateReplication = (mstime - r->last_update) > 1000;
 
         if (fUpdateReplication) {
             listIter li;
@@ -3832,7 +3839,11 @@ int rdbSaveToSlavesSockets(rdbSaveInfo *rsi) {
 
     g_pserver->rdbThreadVars.tmpfileNum++;
     g_pserver->rdbThreadVars.fRdbThreadCancel = false;
-    if (pthread_create(&child, nullptr, rdbSaveToSlavesSocketsThread, args)) {
+    pthread_attr_t tattr;
+    pthread_attr_init(&tattr);
+    pthread_attr_setstacksize(&tattr, 1 << 23); // 8 MB
+    if (pthread_create(&child, &tattr, rdbSaveToSlavesSocketsThread, args)) {
+        pthread_attr_destroy(&tattr);
         serverLog(LL_WARNING,"Can't save in background: fork: %s",
             strerror(errno));
 
@@ -3858,6 +3869,7 @@ int rdbSaveToSlavesSockets(rdbSaveInfo *rsi) {
         closeChildInfoPipe();
         return C_ERR;
     }
+    pthread_attr_destroy(&tattr);
     g_pserver->child_type = CHILD_TYPE_RDB;
 
     serverLog(LL_NOTICE,"Background RDB transfer started");

src/redis-cli.c

@@ -5485,8 +5485,8 @@ static int clusterManagerCommandRebalance(int argc, char **argv) {
             listAddNodeTail(lsrc, src);
             table = clusterManagerComputeReshardTable(lsrc, numslots);
             listRelease(lsrc);
-            int table_len = (int) listLength(table);
-            if (!table || table_len != numslots) {
+            int table_len = 0;
+            if (!table || (table_len = (int) listLength(table)) != numslots) {
                 clusterManagerLogErr("*** Assertion failed: Reshard table "
                                      "!= number of slots");
                 result = 0;

src/replication.cpp

@@ -190,7 +190,7 @@ int bg_unlink(const char *filename) {
 bool createDiskBacklog() {
     // Lets create some disk backed pages and add them here
     std::string path = "./repl-backlog-temp" + std::to_string(gettid());
-#ifdef __APPLE__
+#if (defined __APPLE__ || defined __FreeBSD__)
     int fd = open(path.c_str(), O_CREAT | O_RDWR, S_IRUSR | S_IWUSR);
 #else
     int fd = open(path.c_str(), O_CREAT | O_RDWR | O_LARGEFILE, S_IRUSR | S_IWUSR);
@@ -264,9 +264,8 @@ void resizeReplicationBacklog(long long newsize) {
         newsize = CONFIG_REPL_BACKLOG_MIN_SIZE;
     if (g_pserver->repl_backlog_size == newsize) return;
 
-    std::unique_lock<fastlock> repl_backlog_lock (g_pserver->repl_backlog_lock);
-
     if (g_pserver->repl_backlog != NULL) {
+        std::unique_lock<fastlock> repl_backlog_lock(g_pserver->repl_backlog_lock);
         /* What we actually do is to flush the old buffer and realloc a new
          * empty one. It will refill with new data incrementally.
          * The reason is that copying a few gigabytes adds latency and even
@@ -357,7 +356,7 @@ void freeReplicationBacklog(void) {
 void feedReplicationBacklog(const void *ptr, size_t len) {
     serverAssert(GlobalLocksAcquired());
     const unsigned char *p = (const unsigned char*)ptr;
-
+    std::unique_lock<fastlock> repl_backlog_lock(g_pserver->repl_backlog_lock, std::defer_lock);
 
     if (g_pserver->repl_batch_idxStart >= 0) {
         /* We are lower bounded by the lowest replica offset, or the batch offset start if not applicable */
@@ -417,6 +416,8 @@ void feedReplicationBacklog(const void *ptr, size_t len) {
                 // We need to update a few variables or later asserts will notice we dropped data
                 g_pserver->repl_batch_offStart = g_pserver->master_repl_offset + len;
                 g_pserver->repl_lowest_off = -1;
+                if (!repl_backlog_lock.owns_lock())
+                    repl_backlog_lock.lock();   // we need to acquire the lock if we'll be overwriting data that writeToClient may be reading
             }
         }
     }
@@ -1131,6 +1132,14 @@ public:
             replica->repl_put_online_on_ack = 1;
         }
     }
+
+    void abort() {
+        for (auto replica : replicas) {
+            // Close the connection to force a resync
+            freeClientAsync(replica);
+        }
+        replicas.clear();
+    }
 };
 
 int rdbSaveSnapshotForReplication(struct rdbSaveInfo *rsi) {
@@ -1226,7 +1235,11 @@ int rdbSaveSnapshotForReplication(struct rdbSaveInfo *rsi) {
                 retval = C_ERR;
                 break;
             }
-            serverAssert(count == snapshotDeclaredCount);
+            if (count != snapshotDeclaredCount) {
+                serverLog(LL_WARNING, "Replication BUG: Count of keys sent does not match actual count.  Aborting full sync.");
+                replBuf.abort();
+                break;
+            }
         }
         
         replBuf.end();
@@ -5599,6 +5612,9 @@ void flushReplBacklogToClients()
         serverAssert(g_pserver->master_repl_offset - g_pserver->repl_batch_offStart <= g_pserver->repl_backlog_size);
         serverAssert(g_pserver->repl_batch_idxStart != g_pserver->repl_backlog_idx);
 
+        // Repl backlog writes must become visible to all threads at this point
+        std::atomic_thread_fence(std::memory_order_release);
+
         listIter li;
         listNode *ln;
         listRewind(g_pserver->slaves, &li);

src/server.cpp

@@ -2109,8 +2109,10 @@ void databasesCron(bool fMainThread) {
                         aeAcquireLock();
                     }
 
-                    dictCompleteRehashAsync(serverTL->rehashCtl, true /*fFree*/);
-                    serverTL->rehashCtl = nullptr;
+                    if (serverTL->rehashCtl->done.load(std::memory_order_relaxed)) {
+                        dictCompleteRehashAsync(serverTL->rehashCtl, true /*fFree*/);
+                        serverTL->rehashCtl = nullptr;
+                    }
                 }
 
                 serverAssert(serverTL->rehashCtl == nullptr);
@@ -2886,7 +2888,7 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
     static thread_local bool fFirstRun = true;
     // note: we also copy the DB pointer in case a DB swap is done while the lock is released
     std::vector<redisDb*> vecdb;    // note we cache the database pointer in case a dbswap is done while the lock is released
-    if (cserver.storage_memory_model == STORAGE_WRITETHROUGH && g_pserver->m_pstorageFactory != nullptr && !g_pserver->loading)
+    if (cserver.storage_memory_model == STORAGE_WRITETHROUGH && !g_pserver->loading)
     {
         if (!fFirstRun) {
             mstime_t storage_process_latency;
@@ -5051,6 +5053,8 @@ bool client::asyncCommand(std::function<void(const redisDbPersistentDataSnapshot
                             std::function<void(const redisDbPersistentDataSnapshot *)> &&postFn) 
 {
     serverAssert(FCorrectThread(this));
+    if (serverTL->in_eval)
+        return false;   // we cannot block clients in EVAL
     const redisDbPersistentDataSnapshot *snapshot = nullptr;
     if (!(this->flags & (CLIENT_MULTI | CLIENT_BLOCKED)))
         snapshot = this->db->createSnapshot(this->mvccCheckpoint, false /* fOptional */);
@@ -5532,6 +5536,8 @@ sds genRedisInfoString(const char *section) {
         }
 
         unsigned int lruclock = g_pserver->lruclock.load();
+        ustime_t ustime;
+        __atomic_load(&g_pserver->ustime, &ustime, __ATOMIC_RELAXED);
         info = sdscatfmt(info,
             "# Server\r\n"
             "redis_version:%s\r\n"
@@ -5574,7 +5580,7 @@ sds genRedisInfoString(const char *section) {
             supervised,
             g_pserver->runid,
             g_pserver->port ? g_pserver->port : g_pserver->tls_port,
-            (int64_t)g_pserver->ustime,
+            (int64_t)ustime,
             (int64_t)uptime,
             (int64_t)(uptime/(3600*24)),
             g_pserver->hz.load(),
@@ -7604,7 +7610,7 @@ int main(int argc, char **argv) {
         try {
             loadDataFromDisk();
         } catch (ShutdownException) {
-            exit(EXIT_SUCCESS);
+            _Exit(EXIT_SUCCESS);
         }
 
         if (g_pserver->cluster_enabled) {
@@ -7721,7 +7727,9 @@ int main(int argc, char **argv) {
     g_pserver->garbageCollector.shutdown();
     delete g_pserver->m_pstorageFactory;
 
-    return 0;
+    // Don't return because we don't want to run any global dtors
+    _Exit(EXIT_SUCCESS);
+    return 0;   // Ensure we're well formed even though this won't get hit
 }
 
 /* The End */

src/snapshot.cpp

@@ -26,17 +26,6 @@ public:
     std::vector<dictEntry*> vecde;
 };
 
-void discontinueAsyncRehash(dict *d) {
-    if (d->asyncdata != nullptr) {
-        auto adata = d->asyncdata;
-        while (adata != nullptr) {
-            adata->abondon = true;
-            adata = adata->next;
-        }
-        d->rehashidx = 0;
-    }
-}
-
 const redisDbPersistentDataSnapshot *redisDbPersistentData::createSnapshot(uint64_t mvccCheckpoint, bool fOptional)
 {
     serverAssert(GlobalLocksAcquired());

src/storage/teststorageprovider.cpp

@@ -26,12 +26,10 @@ TestStorageProvider::~TestStorageProvider()
 {
 }
 
-void TestStorageProvider::insert(const char *key, size_t cchKey, void *data, size_t cb, bool fOverwrite)
+void TestStorageProvider::insert(const char *key, size_t cchKey, void *data, size_t cb, bool /*fOverwrite*/)
 {
     auto strkey = std::string(key, cchKey);
-    bool fActuallyExists = m_map.find(strkey) != m_map.end();
-    serverAssert(fActuallyExists == fOverwrite);
-    m_map.insert(std::make_pair(strkey, std::string((char*)data, cb)));
+    m_map[strkey] = std::string((char*)data, cb);
 }
 
 

src/t_stream.cpp

@@ -789,13 +789,13 @@ int64_t streamTrim(stream *s, streamAddTrimArgs *args) {
              * update it after (and if) we actually remove the entry */
             unsigned char *pcopy = p;
 
-            int flags = lpGetInteger(p);
+            int64_t flags = lpGetInteger(p);
             p = lpNext(lp, p); /* Skip flags. */
-            int to_skip;
+            int64_t to_skip;
 
-            int ms_delta = lpGetInteger(p);
+            int64_t ms_delta = lpGetInteger(p);
             p = lpNext(lp, p); /* Skip ID ms delta */
-            int seq_delta = lpGetInteger(p);
+            int64_t seq_delta = lpGetInteger(p);
             p = lpNext(lp, p); /* Skip ID seq delta */
 
             streamID currid = {0}; /* For MINID */
@@ -1125,7 +1125,7 @@ int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
              * the first entry emitted for this listpack, then we already
              * emitted the current entry, and have to go back to the previous
              * one. */
-            int lp_count = lpGetInteger(si->lp_ele);
+            int64_t lp_count = lpGetInteger(si->lp_ele);
             while(lp_count--) si->lp_ele = lpPrev(si->lp,si->lp_ele);
             /* Seek lp-count of prev entry. */
             si->lp_ele = lpPrev(si->lp,si->lp_ele);
@@ -1155,7 +1155,7 @@ int streamIteratorGetID(streamIterator *si, streamID *id, int64_t *numfields) {
 
             /* Get the flags entry. */
             si->lp_flags = si->lp_ele;
-            int flags = lpGetInteger(si->lp_ele);
+            int64_t flags = lpGetInteger(si->lp_ele);
             si->lp_ele = lpNext(si->lp,si->lp_ele); /* Seek ID. */
 
             /* Get the ID: it is encoded as difference between the master
@@ -1264,7 +1264,7 @@ void streamIteratorRemoveEntry(streamIterator *si, streamID *current) {
      * deleted entries in the listpack header.
      *
      * We start flagging: */
-    int flags = lpGetInteger(si->lp_flags);
+    int64_t flags = lpGetInteger(si->lp_flags);
     flags |= STREAM_ITEM_FLAG_DELETED;
     lp = lpReplaceInteger(lp,&si->lp_flags,flags);
 

tests/unit/scripting.tcl

@@ -126,6 +126,10 @@ start_server {tags {"scripting"}} {
         r select 9
         set res
     } {original value}
+    
+    test {EVAL background command} {
+	r eval {redis.call("SCAN", "0", "MATCH", "key*", "COUNT", 1000)} 0
+    } {}
 
     if 0 {
         test {EVAL - Script can't run more than configured time limit} {

tests/unit/type/stream.tcl

@@ -172,6 +172,15 @@ start_server {
         assert_equal [r XRANGE mystream - +] {{3-0 {f v}} {4-0 {f v}} {5-0 {f v}}}
     }
 
+    test {XTRIM with MINID option, big delta from master record} {
+        r DEL mystream
+        r XADD mystream 1-0 f v
+        r XADD mystream 1641544570597-0 f v
+        r XADD mystream 1641544570597-1 f v
+        r XTRIM mystream MINID 1641544570597-0
+        assert_equal [r XRANGE mystream - +] {{1641544570597-0 {f v}} {1641544570597-1 {f v}}}
+    }
+
     test {XADD mass insertion and XLEN} {
         r DEL mystream
         r multi

utils/redis-copy.rb

@@ -1,27 +1,27 @@
-# keydb-copy.rb - Copyright (C) 2009-2010 Salvatore Sanfilippo
+# redis-copy.rb - Copyright (C) 2009-2010 Salvatore Sanfilippo
 # BSD license, See the COPYING file for more information.
 #
-# Copy the whole dataset from one Redis instance to another one
+# Copy the whole dataset from one KeyDB/Redis instance to another one
 #
 # WARNING: this utility is deprecated and serves as a legacy adapter
-#          for the more-robust keydb-copy gem.
+#          for the more-robust redis-copy gem.
 
 require 'shellwords'
 
 def redisCopy(opts={})
   src = "#{opts[:srchost]}:#{opts[:srcport]}"
   dst = "#{opts[:dsthost]}:#{opts[:dstport]}"
-  `keydb-copy #{src.shellescape} #{dst.shellescape}`
+  `redis-copy #{src.shellescape} #{dst.shellescape}`
 rescue Errno::ENOENT
-  $stderr.puts 'This utility requires the keydb-copy executable',
-               'from the keydb-copy gem on https://rubygems.org',
-               'To install it, run `gem install keydb-copy`.'
+  $stderr.puts 'This utility requires the redis-copy executable',
+               'from the redis-copy gem on https://rubygems.org',
+               'To install it, run `gem install redis-copy`.'
   exit 1
 end
 
-$stderr.puts "This utility is deprecated. Use the keydb-copy gem instead."
+$stderr.puts "This utility is deprecated. Use the redis-copy gem instead."
 if ARGV.length != 4
-    puts "Usage: keydb-copy.rb <srchost> <srcport> <dsthost> <dstport>"
+    puts "Usage: redis-copy.rb <srchost> <srcport> <dsthost> <dstport>"
     exit 1
 end
 puts "WARNING: it's up to you to FLUSHDB the destination host before to continue, press any key when ready."