Merge branch 'unstable' into RELEASE_0_9

Merge readme changes

.vscode/settings.json

@@ -0,0 +1,56 @@
+{
+    "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,17 +45,6 @@ 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,36 +191,6 @@ 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)
 {
@@ -229,7 +199,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;
@@ -242,10 +212,9 @@ 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 = safe_write(eventLoop->fdCmdWrite, &cmd, sizeof(cmd));
+    auto size = write(eventLoop->fdCmdWrite, &cmd, sizeof(cmd));
     if (size != sizeof(cmd))
     {
-        AE_ASSERT(size == sizeof(cmd) || size <= 0);
         AE_ASSERT(errno == EAGAIN);
         ret = AE_ERR;
     }
@@ -267,7 +236,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;
@@ -284,7 +253,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(serverTL->neterr, fd, cip, sizeof(cip), &cport);
+        cfd = anetTcpAccept(g_pserver->neterr, fd, cip, sizeof(cip), &cport);
         if (cfd == ANET_ERR) {
             if (errno != EWOULDBLOCK)
                 serverLog(LL_VERBOSE,
-                    "Error accepting cluster node: %s", serverTL->neterr);
+                    "Error accepting cluster node: %s", g_pserver->neterr);
             return;
         }
         anetNonBlock(NULL,cfd);
@@ -3407,7 +3407,7 @@ void clusterCron(void) {
             mstime_t old_ping_sent;
             clusterLink *link;
 
-            fd = anetTcpNonBlockBindConnect(serverTL->neterr, node->ip,
+            fd = anetTcpNonBlockBindConnect(g_pserver->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, serverTL->neterr);
+                    node->cport, g_pserver->neterr);
                 continue;
             }
             link = createClusterLink(node);
@@ -5010,15 +5010,15 @@ migrateCachedSocket* migrateGetSocket(client *c, robj *host, robj *port, long ti
     }
 
     /* Create the socket */
-    fd = anetTcpNonBlockConnect(serverTL->neterr,szFromObj(c->argv[1]),
+    fd = anetTcpNonBlockConnect(g_pserver->neterr,szFromObj(c->argv[1]),
                                 atoi(szFromObj(c->argv[2])));
     if (fd == -1) {
         sdsfree(name);
         addReplyErrorFormat(c,"Can't connect to target node: %s",
-            serverTL->neterr);
+            g_pserver->neterr);
         return NULL;
     }
-    anetEnableTcpNoDelay(serverTL->neterr,fd);
+    anetEnableTcpNoDelay(g_pserver->neterr,fd);
 
     /* Check if it connects within the specified timeout. */
     if ((aeWait(fd,AE_WRITABLE,timeout) & AE_WRITABLE) == 0) {

src/debug.cpp

@@ -55,8 +55,6 @@ typedef ucontext_t sigcontext_t;
 #endif
 #endif
 
-bool g_fInCrash = false;
-
 /* ================================= Debugging ============================== */
 
 /* Compute the sha1 of string at 's' with 'len' bytes long.
@@ -438,7 +436,7 @@ NULL
             "Value at:%p refcount:%d "
             "encoding:%s serializedlength:%zu "
             "lru:%d lru_seconds_idle:%llu%s",
-            (void*)val, static_cast<int>(val->refcount),
+            (void*)val, val->refcount,
             strenc, rdbSavedObjectLen(val),
             val->lru, estimateObjectIdleTime(val)/1000, extra);
     } else if (!strcasecmp(szFromObj(c->argv[1]),"sdslen") && c->argc == 3) {
@@ -723,14 +721,14 @@ void _serverAssertPrintClientInfo(const client *c) {
             arg = buf;
         }
         serverLog(LL_WARNING,"client->argv[%d] = \"%s\" (refcount: %d)",
-            j, arg, static_cast<int>(c->argv[j]->refcount));
+            j, arg, 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", static_cast<int>(o->refcount));
+    serverLog(LL_WARNING,"Object refcount: %d", 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) {
@@ -1358,7 +1356,6 @@ 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,7 +57,6 @@
 #define UNUSED(x) ((void)x)
 #endif
 
-
 /****************************************************
  *
  *      Implementation of a fair spinlock.  To promote fairness we
@@ -65,74 +64,12 @@
  * 
  ****************************************************/
 
-
-#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()
 {
-    uint64_t rval;
-    __atomic_load(&g_longwaits, &rval, __ATOMIC_RELAXED);
-    return rval;
+    return g_longwaits;
 }
 
 #ifndef ASM_SPINLOCK
@@ -170,15 +107,12 @@ 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)
 {
-    int pidOwner;
-    __atomic_load(&lock->m_pidOwner, &pidOwner, __ATOMIC_ACQUIRE);
-    if (pidOwner == gettid())
+    if ((int)__atomic_load_4(&lock->m_pidOwner, __ATOMIC_ACQUIRE) == gettid())
     {
         ++lock->m_depth;
         return;
@@ -190,12 +124,8 @@ extern "C" void fastlock_lock(struct fastlock *lock)
 #endif
     int cloops = 0;
     ticket ticketT;
-    for (;;)
+    while (((ticketT.u = __atomic_load_4(&lock->m_ticket.m_active, __ATOMIC_ACQUIRE)) & 0xffff) != myticket)
     {
-        __atomic_load(&lock->m_ticket.u, &ticketT.u, __ATOMIC_ACQUIRE);
-        if ((ticketT.u & 0xffff) == myticket)
-            break;
-
 #if defined(__i386__) || defined(__amd64__)
         __asm__ ("pause");
 #endif
@@ -206,34 +136,28 @@ 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
-            __atomic_fetch_add(&g_longwaits, 1, __ATOMIC_RELAXED);
+            ++g_longwaits;
         }
     }
 
     lock->m_depth = 1;
-    int tid = gettid();
-    __atomic_store(&lock->m_pidOwner, &tid, __ATOMIC_RELEASE);
-    ANNOTATE_RWLOCK_ACQUIRED(lock, true);
+    __atomic_store_4(&lock->m_pidOwner, gettid(), __ATOMIC_RELEASE);
     std::atomic_thread_fence(std::memory_order_acquire);
 }
 
 extern "C" int fastlock_trylock(struct fastlock *lock, int fWeak)
 {
-    int tid;
-    __atomic_load(&lock->m_pidOwner, &tid, __ATOMIC_ACQUIRE);
-    if (tid == gettid())
+    if ((int)__atomic_load_4(&lock->m_pidOwner, __ATOMIC_ACQUIRE) == gettid())
     {
         ++lock->m_depth;
         return true;
     }
 
     // cheap test
-    struct ticket ticketT;
-    __atomic_load(&lock->m_ticket.u, &ticketT.u, __ATOMIC_ACQUIRE);
-    if (ticketT.m_active != ticketT.m_avail)
+    if (lock->m_ticket.m_active != lock->m_ticket.m_avail)
         return false;
 
-    uint16_t active = ticketT.m_active;
+    uint16_t active = __atomic_load_2(&lock->m_ticket.m_active, __ATOMIC_RELAXED);
     uint16_t next = active + 1;
 
     struct ticket ticket_expect { { { active, active } } };
@@ -241,9 +165,7 @@ 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;
-        tid = gettid();
-        __atomic_store(&lock->m_pidOwner, &tid,  __ATOMIC_RELEASE);
-        ANNOTATE_RWLOCK_ACQUIRED(lock, true);
+        __atomic_store_4(&lock->m_pidOwner, gettid(), __ATOMIC_RELEASE);
         return true;
     }
     return false;
@@ -254,20 +176,13 @@ 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(&lock->futex, &futexT, __ATOMIC_RELAXED);
-    futexT &= mask;
+    unsigned futexT = __atomic_load_4(&lock->futex, __ATOMIC_RELAXED) & mask;
     
     if (futexT == 0)
         return;
     
-    for (;;)
+    while (__atomic_load_4(&lock->futex, __ATOMIC_ACQUIRE) & mask)
     {
-        __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;
     }
@@ -279,13 +194,9 @@ extern "C" void fastlock_unlock(struct fastlock *lock)
     --lock->m_depth;
     if (lock->m_depth == 0)
     {
-        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);
+        assert((int)__atomic_load_4(&lock->m_pidOwner, __ATOMIC_RELAXED) >= 0);  // unlock after free
+        lock->m_pidOwner = -1;
         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);
@@ -302,15 +213,12 @@ 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()
 {
-    int tid;
-    __atomic_load(&m_pidOwner, &tid, __ATOMIC_RELAXED);
-    return gettid() == tid;
+    return gettid() == m_pidOwner;
 }
 
 int fastlock_unlock_recursive(struct fastlock *lock)

src/module.cpp

@@ -3920,9 +3920,7 @@ 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);
 }
 
@@ -3932,8 +3930,6 @@ 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,7 +174,6 @@ 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);
@@ -1004,6 +1003,7 @@ 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(serverTL->neterr, fd, cip, sizeof(cip), &cport);
+        cfd = anetTcpAccept(g_pserver->neterr, fd, cip, sizeof(cip), &cport);
         if (cfd == ANET_ERR) {
             if (errno != EWOULDBLOCK)
                 serverLog(LL_WARNING,
-                    "Accepting client connection: %s", serverTL->neterr);
+                    "Accepting client connection: %s", g_pserver->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(serverTL->neterr, fd);
+        cfd = anetUnixAccept(g_pserver->neterr, fd);
         if (cfd == ANET_ERR) {
             if (errno != EWOULDBLOCK)
                 serverLog(LL_WARNING,
-                    "Accepting client connection: %s", serverTL->neterr);
+                    "Accepting client connection: %s", g_pserver->neterr);
             return;
         }
         int ielCur = ielFromEventLoop(el);
@@ -1266,7 +1266,7 @@ void unlinkClient(client *c) {
     }
 }
 
-bool freeClient(client *c) {
+void freeClient(client *c) {
     listNode *ln;
     serverAssert(c->fd == -1 || GlobalLocksAcquired());
     AssertCorrectThread(c);
@@ -1274,9 +1274,9 @@ bool freeClient(client *c) {
 
     /* If a client is protected, yet we need to free it right now, make sure
      * to at least use asynchronous freeing. */
-    if (c->flags & CLIENT_PROTECTED || c->casyncOpsPending) {
+    if (c->flags & CLIENT_PROTECTED) {
         freeClientAsync(c);
-        return false;
+        return;
     }
 
     /* If it is our master that's beging disconnected we should make sure
@@ -1291,7 +1291,7 @@ bool freeClient(client *c) {
                         CLIENT_BLOCKED)))
         {
             replicationCacheMaster(MasterInfoFromClient(c), c);
-            return false;
+            return;
         }
     }
 
@@ -1370,7 +1370,6 @@ bool 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.
@@ -1383,37 +1382,28 @@ 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;  // check without the lock first
+    if (c->flags & CLIENT_CLOSE_ASAP || c->flags & CLIENT_LUA) return;
     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);
 
-    // Store the clients in a temp vector since freeClient will modify this list
-    std::vector<client*> vecclientsFree;
-    while((ln = listNext(&li))) 
-    {
+    while((ln = listNext(&li))) {
         client *c = (client*)listNodeValue(ln);
-        if (c->iel == iel)
-        {
-            vecclientsFree.push_back(c);
-            listDelNode(g_pserver->clients_to_close, ln);
-        }
-    }
+        if (c->iel != iel)
+            continue;   // wrong thread
 
-    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);
     }
 }
 
@@ -1561,15 +1551,6 @@ 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?
 
@@ -1606,36 +1587,8 @@ void ProcessPendingAsyncWrites()
             continue;
 
         asyncCloseClientOnOutputBufferLimitReached(c);
-        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
-            }
-        }
+        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
     }
 }
 
@@ -1675,15 +1628,13 @@ 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)
             {
-                lock.release(); // still locked
+                c->lock.lock();
                 AeLocker ae;
                 ae.arm(c);
-                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
+                freeClient(c);  // writeToClient will only async close, but there's no need to wait
             }
             continue;
         }
@@ -2339,7 +2290,6 @@ 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,14 +2,6 @@
 #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)
 {
@@ -29,6 +21,4 @@ void operator delete(void * p) noexcept
 void operator delete(void *p, std::size_t) noexcept
 {
     zfree(p);
-}
-
-#endif
+}

src/new.h

@@ -1,12 +1,10 @@
 #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.store(1, std::memory_order_relaxed);
+    o->refcount = 1;
     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.store(OBJ_SHARED_REFCOUNT, std::memory_order_relaxed);
+    o->refcount = OBJ_SHARED_REFCOUNT;
     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.store(1, std::memory_order_relaxed);
+    o->refcount = 1;
     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.fetch_add(1, std::memory_order_acquire);
+    if (o->refcount != OBJ_SHARED_REFCOUNT) o->refcount++;
 }
 
 void decrRefCount(robj_roptr o) {
-    if (o->refcount.load(std::memory_order_acquire) == 1) {
+    if (o->refcount == 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.fetch_sub(1, std::memory_order_acquire);
+        if (o->refcount != OBJ_SHARED_REFCOUNT) o->refcount--;
     }
 }
 

src/pubsub.cpp

@@ -143,8 +143,6 @@ 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;
@@ -204,7 +202,6 @@ 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) {
@@ -247,7 +244,6 @@ 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;
@@ -266,7 +262,6 @@ 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;
@@ -283,7 +278,6 @@ int pubsubUnsubscribeAllPatterns(client *c, int notify) {
 
 /* Publish a message */
 int pubsubPublishMessage(robj *channel, robj *message) {
-    serverAssert(GlobalLocksAcquired());
     int receivers = 0;
     dictEntry *de;
     listNode *ln;
@@ -299,8 +293,6 @@ 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);
@@ -319,8 +311,6 @@ 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 (g_pserver->fActiveReplica || (!listLength(g_pserver->masters) && g_pserver->repl_backlog)) {
+    if (!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,21 +3265,11 @@ void replicaReplayCommand(client *c)
     cFake->lock.lock();
     cFake->authenticated = c->authenticated;
     cFake->puser = c->puser;
-    cFake->querybuf = sdscatsds(cFake->querybuf,(sds)ptrFromObj(c->argv[2]));
+    cFake->querybuf = sdscat(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();
-    if (fExec)
-    {
-        addReply(c, shared.ok);
-        selectDb(c, cFake->db->id);
-    }
-    else
-    {
-        addReplyError(c, "command did not execute");
-    }
+    addReply(c, shared.ok);
     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,8 +1749,7 @@ void databasesCron(void) {
  * a lot faster than calling time(NULL) */
 void updateCachedTime(void) {
     g_pserver->unixtime = time(NULL);
-    long long ms = mstime();
-    __atomic_store(&g_pserver->mstime, &ms, __ATOMIC_RELAXED);
+    g_pserver->mstime = mstime();
 
     /* To get information about daylight saving time, we need to call
      * localtime_r and cache the result. However calling localtime_r in this
@@ -2713,7 +2712,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(serverTL->neterr,port,NULL,
+            fds[*count] = anetTcp6Server(g_pserver->neterr,port,NULL,
                 g_pserver->tcp_backlog, fReusePort);
             if (fds[*count] != ANET_ERR) {
                 anetNonBlock(NULL,fds[*count]);
@@ -2725,7 +2724,7 @@ int listenToPort(int port, int *fds, int *count, int fReusePort) {
 
             if (*count == 1 || unsupported) {
                 /* Bind the IPv4 address as well. */
-                fds[*count] = anetTcpServer(serverTL->neterr,port,NULL,
+                fds[*count] = anetTcpServer(g_pserver->neterr,port,NULL,
                     g_pserver->tcp_backlog, fReusePort);
                 if (fds[*count] != ANET_ERR) {
                     anetNonBlock(NULL,fds[*count]);
@@ -2741,18 +2740,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(serverTL->neterr,port,g_pserver->bindaddr[j],
+            fds[*count] = anetTcp6Server(g_pserver->neterr,port,g_pserver->bindaddr[j],
                 g_pserver->tcp_backlog, fReusePort);
         } else {
             /* Bind IPv4 address. */
-            fds[*count] = anetTcpServer(serverTL->neterr,port,g_pserver->bindaddr[j],
+            fds[*count] = anetTcpServer(g_pserver->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, serverTL->neterr);
+                port, g_pserver->neterr);
                 if (errno == ENOPROTOOPT     || errno == EPROTONOSUPPORT ||
                     errno == ESOCKTNOSUPPORT || errno == EPFNOSUPPORT ||
                     errno == EAFNOSUPPORT    || errno == EADDRNOTAVAIL)
@@ -2838,10 +2837,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(serverTL->neterr,g_pserver->unixsocket,
+        g_pserver->sofd = anetUnixServer(g_pserver->neterr,g_pserver->unixsocket,
             g_pserver->unixsocketperm, g_pserver->tcp_backlog);
         if (g_pserver->sofd == ANET_ERR) {
-            serverLog(LL_WARNING, "Opening Unix socket: %s", serverTL->neterr);
+            serverLog(LL_WARNING, "Opening Unix socket: %s", g_pserver->neterr);
             exit(1);
         }
         anetNonBlock(NULL,g_pserver->sofd);
@@ -3348,7 +3347,6 @@ 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;
@@ -4943,20 +4941,13 @@ int redisIsSupervised(int mode) {
 
 uint64_t getMvccTstamp()
 {
-    uint64_t rval;
-    __atomic_load(&g_pserver->mvcc_tstamp, &rval, __ATOMIC_ACQUIRE);
-    return rval;
+    return g_pserver->mvcc_tstamp;
 }
 
 void incrementMvccTstamp()
 {
-    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
+    uint64_t msPrev = g_pserver->mvcc_tstamp >> 20;
+    if (msPrev >= (uint64_t)g_pserver->mstime)  // we can be greater if the count overflows
     {
         atomicIncr(g_pserver->mvcc_tstamp, 1);
     }

src/server.h

@@ -709,14 +709,13 @@ 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 std::atomic<int> refcount;
+    mutable int refcount;
     uint64_t mvcc_tstamp;
     void *m_ptr;
 } robj;
@@ -925,7 +924,6 @@ 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. */
@@ -1159,8 +1157,6 @@ 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 {
@@ -1278,6 +1274,7 @@ 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. */
@@ -1695,7 +1692,7 @@ void redisSetProcTitle(const char *title);
 /* networking.c -- Networking and Client related operations */
 client *createClient(int fd, int iel);
 void closeTimedoutClients(void);
-bool freeClient(client *c);
+void freeClient(client *c);
 void freeClientAsync(client *c);
 void resetClient(client *c);
 void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask);
@@ -2504,10 +2501,9 @@ 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() || g_fInCrash;
+    return aeThreadOwnsLock() || moduleGILAcquiredByModule();
 }
 
 inline int ielFromEventLoop(const aeEventLoop *eventLoop)

src/version.h

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

tests/integration/replication-active.tcl

@@ -4,8 +4,7 @@ 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 [list overrides [list active-replica yes replicaof [list $slave_host $slave_port]]] {
+    start_server {overrides {active-replica yes}} {
         set master [srv 0 client]
         set master_host [srv 0 host]
         set master_port [srv 0 port]
@@ -14,11 +13,10 @@ 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
-        #note the master is a replica via the config (see start_server above)
+        $master slaveof $slave_host $slave_port
 
         test {Active replicas report the correct role} {
             wait_for_condition 50 100 {
@@ -49,15 +47,6 @@ 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