bump version

.vscode/settings.json

@@ -1,56 +0,0 @@
-{
-    "files.associations": {
-        "zmalloc.h": "c",
-        "stat.h": "c",
-        "array": "cpp",
-        "atomic": "cpp",
-        "*.tcc": "cpp",
-        "cctype": "cpp",
-        "chrono": "cpp",
-        "clocale": "cpp",
-        "cmath": "cpp",
-        "condition_variable": "cpp",
-        "cstdarg": "cpp",
-        "cstddef": "cpp",
-        "cstdint": "cpp",
-        "cstdio": "cpp",
-        "cstdlib": "cpp",
-        "cstring": "cpp",
-        "ctime": "cpp",
-        "cwchar": "cpp",
-        "cwctype": "cpp",
-        "deque": "cpp",
-        "list": "cpp",
-        "unordered_map": "cpp",
-        "vector": "cpp",
-        "exception": "cpp",
-        "fstream": "cpp",
-        "functional": "cpp",
-        "future": "cpp",
-        "initializer_list": "cpp",
-        "iomanip": "cpp",
-        "iosfwd": "cpp",
-        "iostream": "cpp",
-        "istream": "cpp",
-        "limits": "cpp",
-        "memory": "cpp",
-        "mutex": "cpp",
-        "new": "cpp",
-        "numeric": "cpp",
-        "optional": "cpp",
-        "ostream": "cpp",
-        "ratio": "cpp",
-        "scoped_allocator": "cpp",
-        "sstream": "cpp",
-        "stdexcept": "cpp",
-        "streambuf": "cpp",
-        "string_view": "cpp",
-        "system_error": "cpp",
-        "thread": "cpp",
-        "cinttypes": "cpp",
-        "tuple": "cpp",
-        "type_traits": "cpp",
-        "typeinfo": "cpp",
-        "utility": "cpp"
-    }
-}

src/Makefile

@@ -45,6 +45,17 @@ endif
 endif
 
 USEASM?=true
+
+ifneq ($(SANITIZE),)
+	CC=clang
+	CXX=clang++
+	CFLAGS+= -fsanitize=$(SANITIZE) -DSANITIZE
+	CXXFLAGS+= -fsanitize=$(SANITIZE) -DSANITIZE
+	LDFLAGS+= -fsanitize=$(SANITIZE)
+	MALLOC=libc
+	USEASM=false
+endif
+
 # Do we use our assembly spinlock? X64 only
 ifeq ($(uname_S),Linux)
 ifeq ($(uname_M),x86_64)

src/ae.cpp

@@ -191,6 +191,36 @@ void aeProcessCmd(aeEventLoop *eventLoop, int fd, void *, int )
     }
 }
 
+// Unlike write() this is an all or nothing thing.  We will block if a partial write is hit
+ssize_t safe_write(int fd, const void *pv, size_t cb)
+{
+    const char *pcb = (const char*)pv;
+    ssize_t written = 0;
+    do
+    {
+        ssize_t rval = write(fd, pcb, cb);
+        if (rval > 0)
+        {
+            pcb += rval;
+            cb -= rval;
+            written += rval;
+        }
+        else if (errno == EAGAIN)
+        {
+            if (written == 0)
+                break;
+            // if we've already written something then we're committed so keep trying
+        }
+        else
+        {
+            if (rval == 0)
+                return written;
+            return rval;
+        }
+    } while (cb);
+    return written;
+}
+
 int aeCreateRemoteFileEvent(aeEventLoop *eventLoop, int fd, int mask,
         aeFileProc *proc, void *clientData, int fSynchronous)
 {
@@ -199,7 +229,7 @@ int aeCreateRemoteFileEvent(aeEventLoop *eventLoop, int fd, int mask,
 
     int ret = AE_OK;
     
-    aeCommand cmd;
+    aeCommand cmd = {};
     cmd.op = AE_ASYNC_OP::CreateFileEvent;
     cmd.fd = fd;
     cmd.mask = mask;
@@ -212,9 +242,10 @@ int aeCreateRemoteFileEvent(aeEventLoop *eventLoop, int fd, int mask,
     std::unique_lock<std::mutex> ulock(cmd.pctl->mutexcv, std::defer_lock);
     if (fSynchronous)
         cmd.pctl->mutexcv.lock();
-    auto size = write(eventLoop->fdCmdWrite, &cmd, sizeof(cmd));
+    auto size = safe_write(eventLoop->fdCmdWrite, &cmd, sizeof(cmd));
     if (size != sizeof(cmd))
     {
+        AE_ASSERT(size == sizeof(cmd) || size <= 0);
         AE_ASSERT(errno == EAGAIN);
         ret = AE_ERR;
     }
@@ -236,7 +267,7 @@ int aePostFunction(aeEventLoop *eventLoop, aePostFunctionProc *proc, void *arg)
         proc(arg);
         return AE_OK;
     }
-    aeCommand cmd;
+    aeCommand cmd = {};
     cmd.op = AE_ASYNC_OP::PostFunction;
     cmd.proc = proc;
     cmd.clientData = arg;
@@ -253,7 +284,7 @@ int aePostFunction(aeEventLoop *eventLoop, std::function<void()> fn, bool fSynch
         return AE_OK;
     }
 
-    aeCommand cmd;
+    aeCommand cmd = {};
     cmd.op = AE_ASYNC_OP::PostCppFunction;
     cmd.pfn = new (MALLOC_LOCAL) std::function<void()>(fn);
     cmd.pctl = nullptr;

src/cluster.cpp

@@ -643,11 +643,11 @@ void clusterAcceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
     if (listLength(g_pserver->masters) == 0 && g_pserver->loading) return;
 
     while(max--) {
-        cfd = anetTcpAccept(g_pserver->neterr, fd, cip, sizeof(cip), &cport);
+        cfd = anetTcpAccept(serverTL->neterr, fd, cip, sizeof(cip), &cport);
         if (cfd == ANET_ERR) {
             if (errno != EWOULDBLOCK)
                 serverLog(LL_VERBOSE,
-                    "Error accepting cluster node: %s", g_pserver->neterr);
+                    "Error accepting cluster node: %s", serverTL->neterr);
             return;
         }
         anetNonBlock(NULL,cfd);
@@ -3407,7 +3407,7 @@ void clusterCron(void) {
             mstime_t old_ping_sent;
             clusterLink *link;
 
-            fd = anetTcpNonBlockBindConnect(g_pserver->neterr, node->ip,
+            fd = anetTcpNonBlockBindConnect(serverTL->neterr, node->ip,
                 node->cport, NET_FIRST_BIND_ADDR);
             if (fd == -1) {
                 /* We got a synchronous error from connect before
@@ -3418,7 +3418,7 @@ void clusterCron(void) {
                 if (node->ping_sent == 0) node->ping_sent = mstime();
                 serverLog(LL_DEBUG, "Unable to connect to "
                     "Cluster Node [%s]:%d -> %s", node->ip,
-                    node->cport, g_pserver->neterr);
+                    node->cport, serverTL->neterr);
                 continue;
             }
             link = createClusterLink(node);
@@ -5010,15 +5010,15 @@ migrateCachedSocket* migrateGetSocket(client *c, robj *host, robj *port, long ti
     }
 
     /* Create the socket */
-    fd = anetTcpNonBlockConnect(g_pserver->neterr,szFromObj(c->argv[1]),
+    fd = anetTcpNonBlockConnect(serverTL->neterr,szFromObj(c->argv[1]),
                                 atoi(szFromObj(c->argv[2])));
     if (fd == -1) {
         sdsfree(name);
         addReplyErrorFormat(c,"Can't connect to target node: %s",
-            g_pserver->neterr);
+            serverTL->neterr);
         return NULL;
     }
-    anetEnableTcpNoDelay(g_pserver->neterr,fd);
+    anetEnableTcpNoDelay(serverTL->neterr,fd);
 
     /* Check if it connects within the specified timeout. */
     if ((aeWait(fd,AE_WRITABLE,timeout) & AE_WRITABLE) == 0) {

src/debug.cpp

@@ -55,6 +55,8 @@ typedef ucontext_t sigcontext_t;
 #endif
 #endif
 
+bool g_fInCrash = false;
+
 /* ================================= Debugging ============================== */
 
 /* Compute the sha1 of string at 's' with 'len' bytes long.
@@ -436,7 +438,7 @@ NULL
             "Value at:%p refcount:%d "
             "encoding:%s serializedlength:%zu "
             "lru:%d lru_seconds_idle:%llu%s",
-            (void*)val, val->refcount,
+            (void*)val, static_cast<int>(val->refcount),
             strenc, rdbSavedObjectLen(val),
             val->lru, estimateObjectIdleTime(val)/1000, extra);
     } else if (!strcasecmp(szFromObj(c->argv[1]),"sdslen") && c->argc == 3) {
@@ -721,14 +723,14 @@ void _serverAssertPrintClientInfo(const client *c) {
             arg = buf;
         }
         serverLog(LL_WARNING,"client->argv[%d] = \"%s\" (refcount: %d)",
-            j, arg, c->argv[j]->refcount);
+            j, arg, static_cast<int>(c->argv[j]->refcount));
     }
 }
 
 void serverLogObjectDebugInfo(robj_roptr o) {
     serverLog(LL_WARNING,"Object type: %d", o->type);
     serverLog(LL_WARNING,"Object encoding: %d", o->encoding);
-    serverLog(LL_WARNING,"Object refcount: %d", o->refcount);
+    serverLog(LL_WARNING,"Object refcount: %d", static_cast<int>(o->refcount));
     if (o->type == OBJ_STRING && sdsEncodedObject(o)) {
         serverLog(LL_WARNING,"Object raw string len: %zu", sdslen(szFromObj(o)));
         if (sdslen(szFromObj(o)) < 4096) {
@@ -1356,6 +1358,7 @@ void dumpX86Calls(void *addr, size_t len) {
 
 void sigsegvHandler(int sig, siginfo_t *info, void *secret) {
     ucontext_t *uc = (ucontext_t*) secret;
+    g_fInCrash = true;
     void *eip = getMcontextEip(uc);
     sds infostring, clients;
     struct sigaction act;

src/fastlock.cpp

@@ -57,6 +57,7 @@
 #define UNUSED(x) ((void)x)
 #endif
 
+
 /****************************************************
  *
  *      Implementation of a fair spinlock.  To promote fairness we
@@ -64,12 +65,74 @@
  * 
  ****************************************************/
 
+
+#if !defined(__has_feature)
+    #define __has_feature(x) 0
+#endif
+
+#if __has_feature(thread_sanitizer)
+
+    /* Report that a lock has been created at address "lock". */
+    #define ANNOTATE_RWLOCK_CREATE(lock) \
+        AnnotateRWLockCreate(__FILE__, __LINE__, lock)
+
+    /* Report that the lock at address "lock" is about to be destroyed. */
+    #define ANNOTATE_RWLOCK_DESTROY(lock) \
+        AnnotateRWLockDestroy(__FILE__, __LINE__, lock)
+
+    /* Report that the lock at address "lock" has been acquired.
+       is_w=1 for writer lock, is_w=0 for reader lock. */
+    #define ANNOTATE_RWLOCK_ACQUIRED(lock, is_w) \
+        AnnotateRWLockAcquired(__FILE__, __LINE__, lock, is_w)
+
+    /* Report that the lock at address "lock" is about to be released. */
+    #define ANNOTATE_RWLOCK_RELEASED(lock, is_w) \
+      AnnotateRWLockReleased(__FILE__, __LINE__, lock, is_w)
+
+    #if defined(DYNAMIC_ANNOTATIONS_WANT_ATTRIBUTE_WEAK)
+        #if defined(__GNUC__)
+            #define DYNAMIC_ANNOTATIONS_ATTRIBUTE_WEAK __attribute__((weak))
+        #else
+            /* TODO(glider): for Windows support we may want to change this macro in order
+               to prepend __declspec(selectany) to the annotations' declarations. */
+            #error weak annotations are not supported for your compiler
+        #endif
+    #else
+        #define DYNAMIC_ANNOTATIONS_ATTRIBUTE_WEAK
+    #endif
+
+    extern "C" {
+    void AnnotateRWLockCreate(
+        const char *file, int line,
+        const volatile void *lock) DYNAMIC_ANNOTATIONS_ATTRIBUTE_WEAK;
+    void AnnotateRWLockDestroy(
+        const char *file, int line,
+        const volatile void *lock) DYNAMIC_ANNOTATIONS_ATTRIBUTE_WEAK;
+    void AnnotateRWLockAcquired(
+        const char *file, int line,
+        const volatile void *lock, long is_w) DYNAMIC_ANNOTATIONS_ATTRIBUTE_WEAK;
+    void AnnotateRWLockReleased(
+        const char *file, int line,
+        const volatile void *lock, long is_w) DYNAMIC_ANNOTATIONS_ATTRIBUTE_WEAK;
+    }
+
+#else
+
+    #define ANNOTATE_RWLOCK_CREATE(lock)
+    #define ANNOTATE_RWLOCK_DESTROY(lock)
+    #define ANNOTATE_RWLOCK_ACQUIRED(lock, is_w)
+    #define ANNOTATE_RWLOCK_RELEASED(lock, is_w)
+
+#endif
+
 static_assert(sizeof(pid_t) <= sizeof(fastlock::m_pidOwner), "fastlock::m_pidOwner not large enough");
 uint64_t g_longwaits = 0;
 
 uint64_t fastlock_getlongwaitcount()
 {
-    return g_longwaits;
+    uint64_t rval;
+    __atomic_load(&g_longwaits, &rval, __ATOMIC_RELAXED);
+    return rval;
 }
 
 #ifndef ASM_SPINLOCK
@@ -107,12 +170,15 @@ extern "C" void fastlock_init(struct fastlock *lock)
     lock->m_depth = 0;
     lock->m_pidOwner = -1;
     lock->futex = 0;
+    ANNOTATE_RWLOCK_CREATE(lock);
 }
 
 #ifndef ASM_SPINLOCK
 extern "C" void fastlock_lock(struct fastlock *lock)
 {
-    if ((int)__atomic_load_4(&lock->m_pidOwner, __ATOMIC_ACQUIRE) == gettid())
+    int pidOwner;
+    __atomic_load(&lock->m_pidOwner, &pidOwner, __ATOMIC_ACQUIRE);
+    if (pidOwner == gettid())
     {
         ++lock->m_depth;
         return;
@@ -124,8 +190,12 @@ extern "C" void fastlock_lock(struct fastlock *lock)
 #endif
     int cloops = 0;
     ticket ticketT;
-    while (((ticketT.u = __atomic_load_4(&lock->m_ticket.m_active, __ATOMIC_ACQUIRE)) & 0xffff) != myticket)
+    for (;;)
     {
+        __atomic_load(&lock->m_ticket.u, &ticketT.u, __ATOMIC_ACQUIRE);
+        if ((ticketT.u & 0xffff) == myticket)
+            break;
+
 #if defined(__i386__) || defined(__amd64__)
         __asm__ ("pause");
 #endif
@@ -136,28 +206,34 @@ extern "C" void fastlock_lock(struct fastlock *lock)
             futex(&lock->m_ticket.u, FUTEX_WAIT_BITSET_PRIVATE, ticketT.u, nullptr, mask);
             __atomic_fetch_and(&lock->futex, ~mask, __ATOMIC_RELEASE);
 #endif
-            ++g_longwaits;
+            __atomic_fetch_add(&g_longwaits, 1, __ATOMIC_RELAXED);
         }
     }
 
     lock->m_depth = 1;
-    __atomic_store_4(&lock->m_pidOwner, gettid(), __ATOMIC_RELEASE);
+    int tid = gettid();
+    __atomic_store(&lock->m_pidOwner, &tid, __ATOMIC_RELEASE);
+    ANNOTATE_RWLOCK_ACQUIRED(lock, true);
     std::atomic_thread_fence(std::memory_order_acquire);
 }
 
 extern "C" int fastlock_trylock(struct fastlock *lock, int fWeak)
 {
-    if ((int)__atomic_load_4(&lock->m_pidOwner, __ATOMIC_ACQUIRE) == gettid())
+    int tid;
+    __atomic_load(&lock->m_pidOwner, &tid, __ATOMIC_ACQUIRE);
+    if (tid == gettid())
     {
         ++lock->m_depth;
         return true;
     }
 
     // cheap test
-    if (lock->m_ticket.m_active != lock->m_ticket.m_avail)
+    struct ticket ticketT;
+    __atomic_load(&lock->m_ticket.u, &ticketT.u, __ATOMIC_ACQUIRE);
+    if (ticketT.m_active != ticketT.m_avail)
         return false;
 
-    uint16_t active = __atomic_load_2(&lock->m_ticket.m_active, __ATOMIC_RELAXED);
+    uint16_t active = ticketT.m_active;
     uint16_t next = active + 1;
 
     struct ticket ticket_expect { { { active, active } } };
@@ -165,7 +241,9 @@ extern "C" int fastlock_trylock(struct fastlock *lock, int fWeak)
     if (__atomic_compare_exchange(&lock->m_ticket, &ticket_expect, &ticket_setiflocked, fWeak /*weak*/, __ATOMIC_ACQUIRE, __ATOMIC_RELAXED))
     {
         lock->m_depth = 1;
-        __atomic_store_4(&lock->m_pidOwner, gettid(), __ATOMIC_RELEASE);
+        tid = gettid();
+        __atomic_store(&lock->m_pidOwner, &tid,  __ATOMIC_RELEASE);
+        ANNOTATE_RWLOCK_ACQUIRED(lock, true);
         return true;
     }
     return false;
@@ -176,13 +254,20 @@ extern "C" int fastlock_trylock(struct fastlock *lock, int fWeak)
 void unlock_futex(struct fastlock *lock, uint16_t ifutex)
 {
     unsigned mask = (1U << (ifutex % 32));
-    unsigned futexT = __atomic_load_4(&lock->futex, __ATOMIC_RELAXED) & mask;
+    unsigned futexT;
+    __atomic_load(&lock->futex, &futexT, __ATOMIC_RELAXED);
+    futexT &= mask;
     
     if (futexT == 0)
         return;
     
-    while (__atomic_load_4(&lock->futex, __ATOMIC_ACQUIRE) & mask)
+    for (;;)
     {
+        __atomic_load(&lock->futex, &futexT, __ATOMIC_ACQUIRE);
+        futexT &= mask;
+        if (!futexT)
+            break;
+
         if (futex(&lock->m_ticket.u, FUTEX_WAKE_BITSET_PRIVATE, INT_MAX, nullptr, mask) == 1)
             break;
     }
@@ -194,9 +279,13 @@ extern "C" void fastlock_unlock(struct fastlock *lock)
     --lock->m_depth;
     if (lock->m_depth == 0)
     {
-        assert((int)__atomic_load_4(&lock->m_pidOwner, __ATOMIC_RELAXED) >= 0);  // unlock after free
-        lock->m_pidOwner = -1;
+        int pidT;
+        __atomic_load(&lock->m_pidOwner, &pidT, __ATOMIC_RELAXED);
+        assert(pidT >= 0);  // unlock after free
+        int t = -1;
+        __atomic_store(&lock->m_pidOwner, &t, __ATOMIC_RELEASE);
         std::atomic_thread_fence(std::memory_order_release);
+        ANNOTATE_RWLOCK_RELEASED(lock, true);
         uint16_t activeNew = __atomic_add_fetch(&lock->m_ticket.m_active, 1, __ATOMIC_RELEASE);  // on x86 the atomic is not required here, but ASM handles that case
 #ifdef __linux__
         unlock_futex(lock, activeNew);
@@ -213,12 +302,15 @@ extern "C" void fastlock_free(struct fastlock *lock)
     assert((lock->m_ticket.m_active == lock->m_ticket.m_avail)                                        // Asser the lock is unlocked
         || (lock->m_pidOwner == gettid() && (lock->m_ticket.m_active == lock->m_ticket.m_avail-1)));  // OR we own the lock and nobody else is waiting
     lock->m_pidOwner = -2;  // sentinal value indicating free
+    ANNOTATE_RWLOCK_DESTROY(lock);
 }
 
 
 bool fastlock::fOwnLock()
 {
-    return gettid() == m_pidOwner;
+    int tid;
+    __atomic_load(&m_pidOwner, &tid, __ATOMIC_RELAXED);
+    return gettid() == tid;
 }
 
 int fastlock_unlock_recursive(struct fastlock *lock)

src/module.cpp

@@ -3920,7 +3920,9 @@ RedisModuleCtx *RM_GetThreadSafeContext(RedisModuleBlockedClient *bc) {
 
 /* Release a thread safe context. */
 void RM_FreeThreadSafeContext(RedisModuleCtx *ctx) {
+    moduleAcquireGIL(false /*fServerThread*/);
     moduleFreeContext(ctx);
+    moduleReleaseGIL(false /*fServerThread*/);
     zfree(ctx);
 }
 
@@ -3930,6 +3932,8 @@ void RM_FreeThreadSafeContext(RedisModuleCtx *ctx) {
 void RM_ThreadSafeContextLock(RedisModuleCtx *ctx) {
     UNUSED(ctx);
     moduleAcquireGIL(FALSE /*fServerThread*/);
+    if (serverTL == nullptr)
+        serverTL = &g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN];    // arbitrary module threads get the main thread context
 }
 
 /* Release the server lock after a thread safe API call was executed. */

src/networking.cpp

@@ -174,6 +174,7 @@ client *createClient(int fd, int iel) {
     c->bufAsync = NULL;
     c->buflenAsync = 0;
     c->bufposAsync = 0;
+    c->casyncOpsPending = 0;
     memset(c->uuid, 0, UUID_BINARY_LEN);
 
     listSetFreeMethod(c->pubsub_patterns,decrRefCountVoid);
@@ -1003,7 +1004,6 @@ static void acceptCommonHandler(int fd, int flags, char *ip, int iel) {
         serverLog(LL_WARNING,
             "Error registering fd event for the new client: %s (fd=%d)",
             strerror(errno),fd);
-        close(fd); /* May be already closed, just ignore errors */
         return;
     }
 
@@ -1087,11 +1087,11 @@ void acceptTcpHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
     UNUSED(privdata);
 
     while(max--) {
-        cfd = anetTcpAccept(g_pserver->neterr, fd, cip, sizeof(cip), &cport);
+        cfd = anetTcpAccept(serverTL->neterr, fd, cip, sizeof(cip), &cport);
         if (cfd == ANET_ERR) {
             if (errno != EWOULDBLOCK)
                 serverLog(LL_WARNING,
-                    "Accepting client connection: %s", g_pserver->neterr);
+                    "Accepting client connection: %s", serverTL->neterr);
             return;
         }
         serverLog(LL_VERBOSE,"Accepted %s:%d", cip, cport);
@@ -1131,11 +1131,11 @@ void acceptUnixHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
     UNUSED(privdata);
 
     while(max--) {
-        cfd = anetUnixAccept(g_pserver->neterr, fd);
+        cfd = anetUnixAccept(serverTL->neterr, fd);
         if (cfd == ANET_ERR) {
             if (errno != EWOULDBLOCK)
                 serverLog(LL_WARNING,
-                    "Accepting client connection: %s", g_pserver->neterr);
+                    "Accepting client connection: %s", serverTL->neterr);
             return;
         }
         int ielCur = ielFromEventLoop(el);
@@ -1266,7 +1266,7 @@ void unlinkClient(client *c) {
     }
 }
 
-void freeClient(client *c) {
+bool freeClient(client *c) {
     listNode *ln;
     serverAssert(c->fd == -1 || GlobalLocksAcquired());
     AssertCorrectThread(c);
@@ -1274,9 +1274,9 @@ void freeClient(client *c) {
 
     /* If a client is protected, yet we need to free it right now, make sure
      * to at least use asynchronous freeing. */
-    if (c->flags & CLIENT_PROTECTED) {
+    if (c->flags & CLIENT_PROTECTED || c->casyncOpsPending) {
         freeClientAsync(c);
-        return;
+        return false;
     }
 
     /* If it is our master that's beging disconnected we should make sure
@@ -1291,7 +1291,7 @@ void freeClient(client *c) {
                         CLIENT_BLOCKED)))
         {
             replicationCacheMaster(MasterInfoFromClient(c), c);
-            return;
+            return false;
         }
     }
 
@@ -1370,6 +1370,7 @@ void freeClient(client *c) {
     ulock.unlock();
     fastlock_free(&c->lock);
     zfree(c);
+    return true;
 }
 
 /* Schedule a client to free it at a safe time in the serverCron() function.
@@ -1382,28 +1383,37 @@ void freeClientAsync(client *c) {
      * may access the list while Redis uses I/O threads. All the other accesses
      * are in the context of the main thread while the other threads are
      * idle. */
-    if (c->flags & CLIENT_CLOSE_ASAP || c->flags & CLIENT_LUA) return;
+    if (c->flags & CLIENT_CLOSE_ASAP || c->flags & CLIENT_LUA) return;  // check without the lock first
     std::lock_guard<decltype(c->lock)> clientlock(c->lock);
     AeLocker lock;
     lock.arm(c);
+    if (c->flags & CLIENT_CLOSE_ASAP || c->flags & CLIENT_LUA) return;  // race condition after we acquire the lock
     c->flags |= CLIENT_CLOSE_ASAP;    
     listAddNodeTail(g_pserver->clients_to_close,c);
 }
 
 void freeClientsInAsyncFreeQueue(int iel) {
+    serverAssert(GlobalLocksAcquired());
     listIter li;
     listNode *ln;
     listRewind(g_pserver->clients_to_close,&li);
 
-    while((ln = listNext(&li))) {
+    // Store the clients in a temp vector since freeClient will modify this list
+    std::vector<client*> vecclientsFree;
+    while((ln = listNext(&li))) 
+    {
         client *c = (client*)listNodeValue(ln);
-        if (c->iel != iel)
-            continue;   // wrong thread
+        if (c->iel == iel)
+        {
+            vecclientsFree.push_back(c);
+            listDelNode(g_pserver->clients_to_close, ln);
+        }
+    }
 
+    for (client *c : vecclientsFree)
+    {
         c->flags &= ~CLIENT_CLOSE_ASAP;
         freeClient(c);
-        listDelNode(g_pserver->clients_to_close,ln);
-        listRewind(g_pserver->clients_to_close,&li);
     }
 }
 
@@ -1551,6 +1561,15 @@ void ProcessPendingAsyncWrites()
         std::lock_guard<decltype(c->lock)> lock(c->lock);
 
         serverAssert(c->fPendingAsyncWrite);
+        if (c->flags & (CLIENT_CLOSE_ASAP | CLIENT_CLOSE_AFTER_REPLY))
+        {
+            c->bufposAsync = 0;
+            c->buflenAsync = 0;
+            zfree(c->bufAsync);
+            c->bufAsync = nullptr;
+            c->fPendingAsyncWrite = FALSE;
+            continue;
+        }
 
         // TODO: Append to end of reply block?
 
@@ -1587,8 +1606,36 @@ void ProcessPendingAsyncWrites()
             continue;
 
         asyncCloseClientOnOutputBufferLimitReached(c);
-        if (aeCreateRemoteFileEvent(g_pserver->rgthreadvar[c->iel].el, c->fd, ae_flags, sendReplyToClient, c, FALSE) == AE_ERR)
-            continue;   // We can retry later in the cron
+        if (c->flags & CLIENT_CLOSE_ASAP)
+            continue;   // we will never write this so don't post an op
+        
+        std::atomic_thread_fence(std::memory_order_seq_cst);
+        
+        if (c->casyncOpsPending == 0)
+        {
+            if (FCorrectThread(c))
+            {
+                prepareClientToWrite(c, false); // queue an event
+            }
+            else
+            {
+                // We need to start the write on the client's thread
+                if (aePostFunction(g_pserver->rgthreadvar[c->iel].el, [c]{
+                        // Install a write handler.  Don't do the actual write here since we don't want
+                        //  to duplicate the throttling and safety mechanisms of the normal write code
+                        std::lock_guard<decltype(c->lock)> lock(c->lock);
+                        serverAssert(c->casyncOpsPending > 0);
+                        c->casyncOpsPending--;
+                        aeCreateFileEvent(g_pserver->rgthreadvar[c->iel].el, c->fd, AE_WRITABLE|AE_WRITE_THREADSAFE, sendReplyToClient, c);
+                    }, false) == AE_ERR
+                )
+                {
+                    // Posting the function failed
+                    continue;   // We can retry later in the cron
+                }
+                ++c->casyncOpsPending; // race is handled by the client lock in the lambda
+            }
+        }
     }
 }
 
@@ -1628,13 +1675,15 @@ int handleClientsWithPendingWrites(int iel) {
         std::unique_lock<decltype(c->lock)> lock(c->lock);
 
         /* Try to write buffers to the client socket. */
-        if (writeToClient(c->fd,c,0) == C_ERR) {
+        if (writeToClient(c->fd,c,0) == C_ERR) 
+        {
             if (c->flags & CLIENT_CLOSE_ASAP)
             {
-                c->lock.lock();
+                lock.release(); // still locked
                 AeLocker ae;
                 ae.arm(c);
-                freeClient(c);  // writeToClient will only async close, but there's no need to wait
+                if (!freeClient(c))  // writeToClient will only async close, but there's no need to wait
+                    c->lock.unlock();   // if we just got put on the async close list, then we need to remove the lock
             }
             continue;
         }
@@ -2290,6 +2339,7 @@ sds getAllClientsInfoString(int type) {
     listRewind(g_pserver->clients,&li);
     while ((ln = listNext(&li)) != NULL) {
         client = reinterpret_cast<struct client*>(listNodeValue(ln));
+        std::unique_lock<decltype(client->lock)> lock(client->lock);
         if (type != -1 && getClientType(client) != type) continue;
         o = catClientInfoString(o,client);
         o = sdscatlen(o,"\n",1);

src/new.cpp

@@ -2,6 +2,14 @@
 #include "server.h"
 #include "new.h"
 
+#ifdef SANITIZE
+void *operator new(size_t size, enum MALLOC_CLASS mclass)
+{
+    (void)mclass;
+    return ::operator new(size);
+}
+
+#else
 [[deprecated]]
 void *operator new(size_t size)
 {
@@ -21,4 +29,6 @@ void operator delete(void * p) noexcept
 void operator delete(void *p, std::size_t) noexcept
 {
     zfree(p);
-}
+}
+
+#endif

src/new.h

@@ -1,10 +1,12 @@
 #pragma once
 #include <cstddef>  // std::size_t
 
+void *operator new(size_t size, enum MALLOC_CLASS mclass);
+
+#ifndef SANITIZE
 [[deprecated]]
 void *operator new(size_t size);
 
-void *operator new(size_t size, enum MALLOC_CLASS mclass);
-
 void operator delete(void * p) noexcept;
 void operator delete(void *p, std::size_t) noexcept;
+#endif

src/object.cpp

@@ -43,7 +43,7 @@ robj *createObject(int type, void *ptr) {
     o->type = type;
     o->encoding = OBJ_ENCODING_RAW;
     o->m_ptr = ptr;
-    o->refcount = 1;
+    o->refcount.store(1, std::memory_order_relaxed);
     o->mvcc_tstamp = OBJ_MVCC_INVALID;
 
     /* Set the LRU to the current lruclock (minutes resolution), or
@@ -69,7 +69,7 @@ robj *createObject(int type, void *ptr) {
  */
 robj *makeObjectShared(robj *o) {
     serverAssert(o->refcount == 1);
-    o->refcount = OBJ_SHARED_REFCOUNT;
+    o->refcount.store(OBJ_SHARED_REFCOUNT, std::memory_order_relaxed);
     return o;
 }
 
@@ -91,7 +91,7 @@ robj *createEmbeddedStringObject(const char *ptr, size_t len) {
 
     o->type = OBJ_STRING;
     o->encoding = OBJ_ENCODING_EMBSTR;
-    o->refcount = 1;
+    o->refcount.store(1, std::memory_order_relaxed);
     o->mvcc_tstamp = OBJ_MVCC_INVALID;
 
     if (g_pserver->maxmemory_policy & MAXMEMORY_FLAG_LFU) {
@@ -352,11 +352,11 @@ void freeStreamObject(robj_roptr o) {
 }
 
 void incrRefCount(robj_roptr o) {
-    if (o->refcount != OBJ_SHARED_REFCOUNT) o->refcount++;
+    if (o->refcount != OBJ_SHARED_REFCOUNT) o->refcount.fetch_add(1, std::memory_order_acquire);
 }
 
 void decrRefCount(robj_roptr o) {
-    if (o->refcount == 1) {
+    if (o->refcount.load(std::memory_order_acquire) == 1) {
         switch(o->type) {
         case OBJ_STRING: freeStringObject(o); break;
         case OBJ_LIST: freeListObject(o); break;
@@ -370,7 +370,7 @@ void decrRefCount(robj_roptr o) {
         zfree(o.unsafe_robjcast());
     } else {
         if (o->refcount <= 0) serverPanic("decrRefCount against refcount <= 0");
-        if (o->refcount != OBJ_SHARED_REFCOUNT) o->refcount--;
+        if (o->refcount != OBJ_SHARED_REFCOUNT) o->refcount.fetch_sub(1, std::memory_order_acquire);
     }
 }
 

src/pubsub.cpp

@@ -143,6 +143,8 @@ int clientSubscriptionsCount(client *c) {
 /* Subscribe a client to a channel. Returns 1 if the operation succeeded, or
  * 0 if the client was already subscribed to that channel. */
 int pubsubSubscribeChannel(client *c, robj *channel) {
+    serverAssert(GlobalLocksAcquired());
+    serverAssert(c->lock.fOwnLock());
     dictEntry *de;
     list *clients = NULL;
     int retval = 0;
@@ -202,6 +204,7 @@ int pubsubUnsubscribeChannel(client *c, robj *channel, int notify) {
 
 /* Subscribe a client to a pattern. Returns 1 if the operation succeeded, or 0 if the client was already subscribed to that pattern. */
 int pubsubSubscribePattern(client *c, robj *pattern) {
+    serverAssert(GlobalLocksAcquired());
     int retval = 0;
 
     if (listSearchKey(c->pubsub_patterns,pattern) == NULL) {
@@ -244,6 +247,7 @@ int pubsubUnsubscribePattern(client *c, robj *pattern, int notify) {
 /* Unsubscribe from all the channels. Return the number of channels the
  * client was subscribed to. */
 int pubsubUnsubscribeAllChannels(client *c, int notify) {
+    serverAssert(GlobalLocksAcquired());
     dictIterator *di = dictGetSafeIterator(c->pubsub_channels);
     dictEntry *de;
     int count = 0;
@@ -262,6 +266,7 @@ int pubsubUnsubscribeAllChannels(client *c, int notify) {
 /* Unsubscribe from all the patterns. Return the number of patterns the
  * client was subscribed from. */
 int pubsubUnsubscribeAllPatterns(client *c, int notify) {
+    serverAssert(GlobalLocksAcquired());
     listNode *ln;
     listIter li;
     int count = 0;
@@ -278,6 +283,7 @@ int pubsubUnsubscribeAllPatterns(client *c, int notify) {
 
 /* Publish a message */
 int pubsubPublishMessage(robj *channel, robj *message) {
+    serverAssert(GlobalLocksAcquired());
     int receivers = 0;
     dictEntry *de;
     listNode *ln;
@@ -293,6 +299,8 @@ int pubsubPublishMessage(robj *channel, robj *message) {
         listRewind(list,&li);
         while ((ln = listNext(&li)) != NULL) {
             client *c = reinterpret_cast<client*>(ln->value);
+            if (c->flags & CLIENT_CLOSE_ASAP)   // avoid blocking if the write will be ignored
+                continue;
             fastlock_lock(&c->lock);
             addReplyPubsubMessage(c,channel,message);
             fastlock_unlock(&c->lock);
@@ -311,6 +319,8 @@ int pubsubPublishMessage(robj *channel, robj *message) {
                                 (char*)ptrFromObj(channel),
                                 sdslen(szFromObj(channel)),0))
             {
+                if (pat->pclient->flags & CLIENT_CLOSE_ASAP)
+                    continue;
                 fastlock_lock(&pat->pclient->lock);
                 addReplyPubsubPatMessage(pat->pclient,
                     pat->pattern,channel,message);

src/rdb.cpp

@@ -2547,7 +2547,7 @@ rdbSaveInfo *rdbPopulateSaveInfo(rdbSaveInfo *rsi) {
      * connects to us, the NULL repl_backlog will trigger a full
      * synchronization, at the same time we will use a new replid and clear
      * replid2. */
-    if (!listLength(g_pserver->masters) && g_pserver->repl_backlog) {
+    if (g_pserver->fActiveReplica || (!listLength(g_pserver->masters) && g_pserver->repl_backlog)) {
         /* Note that when g_pserver->slaveseldb is -1, it means that this master
          * didn't apply any write commands after a full synchronization.
          * So we can let repl_stream_db be 0, this allows a restarted slave

src/replication.cpp

@@ -3265,11 +3265,21 @@ void replicaReplayCommand(client *c)
     cFake->lock.lock();
     cFake->authenticated = c->authenticated;
     cFake->puser = c->puser;
-    cFake->querybuf = sdscat(cFake->querybuf,(sds)ptrFromObj(c->argv[2]));
+    cFake->querybuf = sdscatsds(cFake->querybuf,(sds)ptrFromObj(c->argv[2]));
     selectDb(cFake, c->db->id);
+    auto ccmdPrev = serverTL->commandsExecuted;
     processInputBuffer(cFake, (CMD_CALL_FULL & (~CMD_CALL_PROPAGATE)));
+    bool fExec = ccmdPrev != serverTL->commandsExecuted;
     cFake->lock.unlock();
-    addReply(c, shared.ok);
+    if (fExec)
+    {
+        addReply(c, shared.ok);
+        selectDb(c, cFake->db->id);
+    }
+    else
+    {
+        addReplyError(c, "command did not execute");
+    }
     freeClient(cFake);
     serverTL->current_client = current_clientSave;
 

src/server.cpp

@@ -1023,7 +1023,7 @@ extern "C" void nolocks_localtime(struct tm *tmp, time_t t, time_t tz, int dst);
  * serverLog() is to prefer. */
 void serverLogRaw(int level, const char *msg) {
     const int syslogLevelMap[] = { LOG_DEBUG, LOG_INFO, LOG_NOTICE, LOG_WARNING };
-    const char *c = ".-*#";
+    const char *c = ".-*#                                                             ";
     FILE *fp;
     char buf[64];
     int rawmode = (level & LL_RAW);
@@ -1749,7 +1749,8 @@ void databasesCron(void) {
  * a lot faster than calling time(NULL) */
 void updateCachedTime(void) {
     g_pserver->unixtime = time(NULL);
-    g_pserver->mstime = mstime();
+    long long ms = mstime();
+    __atomic_store(&g_pserver->mstime, &ms, __ATOMIC_RELAXED);
 
     /* To get information about daylight saving time, we need to call
      * localtime_r and cache the result. However calling localtime_r in this
@@ -2712,7 +2713,7 @@ int listenToPort(int port, int *fds, int *count, int fReusePort) {
             int unsupported = 0;
             /* Bind * for both IPv6 and IPv4, we enter here only if
              * g_pserver->bindaddr_count == 0. */
-            fds[*count] = anetTcp6Server(g_pserver->neterr,port,NULL,
+            fds[*count] = anetTcp6Server(serverTL->neterr,port,NULL,
                 g_pserver->tcp_backlog, fReusePort);
             if (fds[*count] != ANET_ERR) {
                 anetNonBlock(NULL,fds[*count]);
@@ -2724,7 +2725,7 @@ int listenToPort(int port, int *fds, int *count, int fReusePort) {
 
             if (*count == 1 || unsupported) {
                 /* Bind the IPv4 address as well. */
-                fds[*count] = anetTcpServer(g_pserver->neterr,port,NULL,
+                fds[*count] = anetTcpServer(serverTL->neterr,port,NULL,
                     g_pserver->tcp_backlog, fReusePort);
                 if (fds[*count] != ANET_ERR) {
                     anetNonBlock(NULL,fds[*count]);
@@ -2740,18 +2741,18 @@ int listenToPort(int port, int *fds, int *count, int fReusePort) {
             if (*count + unsupported == 2) break;
         } else if (strchr(g_pserver->bindaddr[j],':')) {
             /* Bind IPv6 address. */
-            fds[*count] = anetTcp6Server(g_pserver->neterr,port,g_pserver->bindaddr[j],
+            fds[*count] = anetTcp6Server(serverTL->neterr,port,g_pserver->bindaddr[j],
                 g_pserver->tcp_backlog, fReusePort);
         } else {
             /* Bind IPv4 address. */
-            fds[*count] = anetTcpServer(g_pserver->neterr,port,g_pserver->bindaddr[j],
+            fds[*count] = anetTcpServer(serverTL->neterr,port,g_pserver->bindaddr[j],
                 g_pserver->tcp_backlog, fReusePort);
         }
         if (fds[*count] == ANET_ERR) {
             serverLog(LL_WARNING,
                 "Could not create server TCP listening socket %s:%d: %s",
                 g_pserver->bindaddr[j] ? g_pserver->bindaddr[j] : "*",
-                port, g_pserver->neterr);
+                port, serverTL->neterr);
                 if (errno == ENOPROTOOPT     || errno == EPROTONOSUPPORT ||
                     errno == ESOCKTNOSUPPORT || errno == EPFNOSUPPORT ||
                     errno == EAFNOSUPPORT    || errno == EADDRNOTAVAIL)
@@ -2837,10 +2838,10 @@ static void initNetworking(int fReusePort)
     /* Open the listening Unix domain socket. */
     if (g_pserver->unixsocket != NULL) {
         unlink(g_pserver->unixsocket); /* don't care if this fails */
-        g_pserver->sofd = anetUnixServer(g_pserver->neterr,g_pserver->unixsocket,
+        g_pserver->sofd = anetUnixServer(serverTL->neterr,g_pserver->unixsocket,
             g_pserver->unixsocketperm, g_pserver->tcp_backlog);
         if (g_pserver->sofd == ANET_ERR) {
-            serverLog(LL_WARNING, "Opening Unix socket: %s", g_pserver->neterr);
+            serverLog(LL_WARNING, "Opening Unix socket: %s", serverTL->neterr);
             exit(1);
         }
         anetNonBlock(NULL,g_pserver->sofd);
@@ -3347,6 +3348,7 @@ void call(client *c, int flags) {
     dirty = g_pserver->dirty;
     start = ustime();
     c->cmd->proc(c);
+    serverTL->commandsExecuted++;
     duration = ustime()-start;
     dirty = g_pserver->dirty-dirty;
     if (dirty < 0) dirty = 0;
@@ -4941,13 +4943,20 @@ int redisIsSupervised(int mode) {
 
 uint64_t getMvccTstamp()
 {
-    return g_pserver->mvcc_tstamp;
+    uint64_t rval;
+    __atomic_load(&g_pserver->mvcc_tstamp, &rval, __ATOMIC_ACQUIRE);
+    return rval;
 }
 
 void incrementMvccTstamp()
 {
-    uint64_t msPrev = g_pserver->mvcc_tstamp >> 20;
-    if (msPrev >= (uint64_t)g_pserver->mstime)  // we can be greater if the count overflows
+    uint64_t msPrev;
+    __atomic_load(&g_pserver->mvcc_tstamp, &msPrev, __ATOMIC_ACQUIRE);
+    msPrev >>= 20;  // convert to milliseconds
+
+    long long mst;
+    __atomic_load(&g_pserver->mstime, &mst, __ATOMIC_RELAXED);
+    if (msPrev >= (uint64_t)mst)  // we can be greater if the count overflows
     {
         atomicIncr(g_pserver->mvcc_tstamp, 1);
     }

src/server.h

@@ -709,13 +709,14 @@ typedef struct RedisModuleDigest {
 
 #define OBJ_SHARED_REFCOUNT INT_MAX
 #define OBJ_MVCC_INVALID (0xFFFFFFFFFFFFFFFFULL)
+
 typedef struct redisObject {
     unsigned type:4;
     unsigned encoding:4;
     unsigned lru:LRU_BITS; /* LRU time (relative to global lru_clock) or
                             * LFU data (least significant 8 bits frequency
                             * and most significant 16 bits access time). */
-    mutable int refcount;
+    mutable std::atomic<int> refcount;
     uint64_t mvcc_tstamp;
     void *m_ptr;
 } robj;
@@ -924,6 +925,7 @@ typedef struct client {
     time_t lastinteraction; /* Time of the last interaction, used for timeout */
     time_t obuf_soft_limit_reached_time;
     std::atomic<int> flags;              /* Client flags: CLIENT_* macros. */
+    int casyncOpsPending;
     int fPendingAsyncWrite; /* NOTE: Not a flag because it is written to outside of the client lock (locked by the global lock instead) */
     int authenticated;      /* Needed when the default user requires auth. */
     int replstate;          /* Replication state if this is a slave. */
@@ -1157,6 +1159,8 @@ struct redisServerThreadVars {
                                 to be processed. */
     client *lua_client = nullptr;   /* The "fake client" to query Redis from Lua */
     struct fastlock lockPendingWrite;
+    char neterr[ANET_ERR_LEN];   /* Error buffer for anet.c */
+    long unsigned commandsExecuted = 0;
 };
 
 struct redisMaster {
@@ -1274,7 +1278,6 @@ struct redisServer {
     rax *clients_index;         /* Active clients dictionary by client ID. */
     int clients_paused;         /* True if clients are currently paused */
     mstime_t clients_pause_end_time; /* Time when we undo clients_paused */
-    char neterr[ANET_ERR_LEN];   /* Error buffer for anet.c */
     dict *migrate_cached_sockets;/* MIGRATE cached sockets */
     std::atomic<uint64_t> next_client_id; /* Next client unique ID. Incremental. */
     int protected_mode;         /* Don't accept external connections. */
@@ -1692,7 +1695,7 @@ void redisSetProcTitle(const char *title);
 /* networking.c -- Networking and Client related operations */
 client *createClient(int fd, int iel);
 void closeTimedoutClients(void);
-void freeClient(client *c);
+bool freeClient(client *c);
 void freeClientAsync(client *c);
 void resetClient(client *c);
 void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask);
@@ -2501,9 +2504,10 @@ void xorDigest(unsigned char *digest, const void *ptr, size_t len);
 int populateCommandTableParseFlags(struct redisCommand *c, const char *strflags);
 
 int moduleGILAcquiredByModule(void);
+extern bool g_fInCrash;
 static inline int GlobalLocksAcquired(void)  // Used in asserts to verify all global locks are correctly acquired for a server-thread to operate
 {
-    return aeThreadOwnsLock() || moduleGILAcquiredByModule();
+    return aeThreadOwnsLock() || moduleGILAcquiredByModule() || g_fInCrash;
 }
 
 inline int ielFromEventLoop(const aeEventLoop *eventLoop)

src/version.h

@@ -1,3 +1,3 @@
-#define KEYDB_REAL_VERSION "0.0.0"
+#define KEYDB_REAL_VERSION "5.0.2"
 extern const char *KEYDB_SET_VERSION;   // Unlike real version, this can be overriden by the config
 

tests/integration/replication-active.tcl

@@ -4,7 +4,8 @@ start_server {tags {"active-repl"} overrides {active-replica yes}} {
     set slave_port [srv 0 port]
     set slave_log [srv 0 stdout]
     set slave_pid [s process_id]
-    start_server {overrides {active-replica yes}} {
+
+    start_server [list overrides [list active-replica yes replicaof [list $slave_host $slave_port]]] {
         set master [srv 0 client]
         set master_host [srv 0 host]
         set master_port [srv 0 port]
@@ -13,10 +14,11 @@ start_server {tags {"active-repl"} overrides {active-replica yes}} {
         # are no bugs the timeout is triggered in a reasonable amount
         # of time.
         $slave config set repl-timeout 5
+        $master config set repl-timeout 5
 
         # Start the replication process...
         $slave slaveof $master_host $master_port
-        $master slaveof $slave_host $slave_port
+        #note the master is a replica via the config (see start_server above)
 
         test {Active replicas report the correct role} {
             wait_for_condition 50 100 {
@@ -47,6 +49,15 @@ start_server {tags {"active-repl"} overrides {active-replica yes}} {
             }
         }
 
+        test {Active replicas propogate binary} {
+            $master set binkey "\u0000foo"
+            wait_for_condition 50 500 {
+                [string match *foo* [$slave get binkey]]
+            } else {
+                fail "replication failed to propogate binary data"
+            }
+        }
+
         test {Active replicas WAIT} {
             # Test that wait succeeds since replicas should be syncronized
             $master set testkey foo