Fix TSAN warnings on the repl backlog

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:

pkg/deb/master_changelog

@@ -1,3 +1,17 @@
+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

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

@@ -3263,10 +3263,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))

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/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);
@@ -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);

src/rdb.cpp

@@ -3049,7 +3049,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;

src/replication.cpp

@@ -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
             }
         }
     }
@@ -5599,6 +5600,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);
@@ -5532,6 +5534,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 +5578,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 +7608,7 @@ int main(int argc, char **argv) {
         try {
             loadDataFromDisk();
         } catch (ShutdownException) {
-            exit(EXIT_SUCCESS);
+            _Exit(EXIT_SUCCESS);
         }
 
         if (g_pserver->cluster_enabled) {
@@ -7721,7 +7725,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());