Fix potential race in pubsub

.vscode/settings.json

@@ -1,58 +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",
-        "set": "cpp",
-        "algorithm": "cpp"
-    }
-}

src/ae.cpp

@@ -84,7 +84,7 @@ fastlock g_lock("AE (global)");
 #endif
 thread_local aeEventLoop *g_eventLoopThisThread = NULL;
 
-#define AE_ASSERT(x) if (!(x)) do { fprintf(stderr, "AE_ASSERT FAILURE %s: %d\n", __FILE__, __LINE__); *((volatile int*)0) = 1; } while(0)
+#define AE_ASSERT(x) if (!(x)) do { fprintf(stderr, "AE_ASSER FAILURE\n"); *((volatile int*)0) = 1; } while(0)
 
 /* Include the best multiplexing layer supported by this system.
  * The following should be ordered by performances, descending. */

src/cluster.cpp

@@ -617,17 +617,6 @@ clusterLink *createClusterLink(clusterNode *node) {
  * This function will just make sure that the original node associated
  * with this link will have the 'link' field set to NULL. */
 void freeClusterLink(clusterLink *link) {
-    if (ielFromEventLoop(serverTL->el) != IDX_EVENT_LOOP_MAIN)
-    {
-        // we can't perform this operation on this thread, queue it on the main thread
-        if (link->node)
-            link->node->link = NULL;
-        link->node = nullptr;
-        aePostFunction(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, [link]{
-            freeClusterLink(link);
-        });
-        return;
-    }
     if (link->fd != -1) {
         aeDeleteFileEvent(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, link->fd, AE_READABLE|AE_WRITABLE);
     }
@@ -2150,35 +2139,21 @@ void handleLinkIOError(clusterLink *link) {
  * consumed by write(). We don't try to optimize this for speed too much
  * as this is a very low traffic channel. */
 void clusterWriteHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
-    serverAssert(ielFromEventLoop(el) == IDX_EVENT_LOOP_MAIN);
     clusterLink *link = (clusterLink*) privdata;
     ssize_t nwritten;
     UNUSED(el);
     UNUSED(mask);
 
-    // We're about to release the lock, so the link's sndbuf needs to be owned fully by us
-    //  allocate a new one in case anyone tries to write while we're waiting
-    sds sndbuf = link->sndbuf;
-    link->sndbuf = sdsempty();
-
-    aeReleaseLock();
-    nwritten = write(fd, sndbuf, sdslen(sndbuf));
-    aeAcquireLock();
-
+    nwritten = write(fd, link->sndbuf, sdslen(link->sndbuf));
     if (nwritten <= 0) {
         serverLog(LL_DEBUG,"I/O error writing to node link: %s",
             (nwritten == -1) ? strerror(errno) : "short write");
-        sdsfree(sndbuf);
         handleLinkIOError(link);
         return;
     }
-    sdsrange(sndbuf,nwritten,-1);
-    // Restore our send buffer, ensuring any unsent data is first
-    sndbuf = sdscat(sndbuf, link->sndbuf);
-    sdsfree(link->sndbuf);
-    link->sndbuf = sndbuf;
+    sdsrange(link->sndbuf,nwritten,-1);
     if (sdslen(link->sndbuf) == 0)
-        aeDeleteFileEvent(el, link->fd, AE_WRITABLE);
+        aeDeleteFileEvent(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, link->fd, AE_WRITABLE);
 }
 
 /* Read data. Try to read the first field of the header first to check the
@@ -2253,10 +2228,9 @@ void clusterReadHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
  * the link to be invalidated, so it is safe to call this function
  * from event handlers that will do stuff with the same link later. */
 void clusterSendMessage(clusterLink *link, unsigned char *msg, size_t msglen) {
-    serverAssert(GlobalLocksAcquired());
     if (sdslen(link->sndbuf) == 0 && msglen != 0)
-        aeCreateRemoteFileEvent(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el,link->fd,AE_WRITABLE|AE_BARRIER,
-                    clusterWriteHandler,link,false);
+        aeCreateFileEvent(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el,link->fd,AE_WRITABLE|AE_BARRIER,
+                    clusterWriteHandler,link);
 
     link->sndbuf = sdscatlen(link->sndbuf, msg, msglen);
 
@@ -3310,7 +3284,7 @@ void clusterHandleSlaveMigration(int max_slaves) {
 void resetManualFailover(void) {
     if (g_pserver->cluster->mf_end && clientsArePaused()) {
         g_pserver->clients_pause_end_time = 0;
-        unpauseClientsIfNecessary();
+        clientsArePaused(); /* Just use the side effect of the function. */
     }
     g_pserver->cluster->mf_end = 0; /* No manual failover in progress. */
     g_pserver->cluster->mf_can_start = 0;
@@ -4180,76 +4154,57 @@ void clusterReplyMultiBulkSlots(client *c) {
     dictIterator *di = dictGetSafeIterator(g_pserver->cluster->nodes);
     while((de = dictNext(di)) != NULL) {
         clusterNode *node = (clusterNode*)dictGetVal(de);
-        int start = -1;
+        int j = 0, start = -1;
 
         /* Skip slaves (that are iterated when producing the output of their
          * master) and  masters not serving any slot. */
         if (!nodeIsMaster(node) || node->numslots == 0) continue;
-        
-        static_assert((CLUSTER_SLOTS % (sizeof(uint32_t)*8)) == 0, "code below assumes the bitfield is a multiple of sizeof(unsinged)");
 
-        for (unsigned iw = 0; iw < (CLUSTER_SLOTS/sizeof(uint32_t)/8); ++iw)
-        {
-            uint32_t wordCur = reinterpret_cast<uint32_t*>(node->slots)[iw];
-            if (iw != ((CLUSTER_SLOTS/sizeof(uint32_t)/8)-1))
-            {
-                if (start == -1 && wordCur == 0)
-                    continue;
-                if (start != -1 && (wordCur+1)==0)
-                    continue;
+        for (j = 0; j < CLUSTER_SLOTS; j++) {
+            int bit, i;
+
+            if ((bit = clusterNodeGetSlotBit(node,j)) != 0) {
+                if (start == -1) start = j;
             }
+            if (start != -1 && (!bit || j == CLUSTER_SLOTS-1)) {
+                int nested_elements = 3; /* slots (2) + master addr (1). */
+                void *nested_replylen = addReplyDeferredLen(c);
 
-            unsigned ibitStartLoop = iw*sizeof(uint32_t)*8;
-    
-            for (unsigned j = ibitStartLoop; j < (iw+1)*sizeof(uint32_t)*8; j++) {
-                int i;
-                int bit = (int)(wordCur & 1);
-                wordCur >>= 1;
-                if (bit != 0) {
-                    if (start == -1) start = j;
+                if (bit && j == CLUSTER_SLOTS-1) j++;
+
+                /* If slot exists in output map, add to it's list.
+                 * else, create a new output map for this slot */
+                if (start == j-1) {
+                    addReplyLongLong(c, start); /* only one slot; low==high */
+                    addReplyLongLong(c, start);
+                } else {
+                    addReplyLongLong(c, start); /* low */
+                    addReplyLongLong(c, j-1);   /* high */
                 }
-                if (start != -1 && (!bit || j == CLUSTER_SLOTS-1)) {
-                    int nested_elements = 3; /* slots (2) + master addr (1). */
-                    void *nested_replylen = addReplyDeferredLen(c);
+                start = -1;
 
-                    if (bit && j == CLUSTER_SLOTS-1) j++;
+                /* First node reply position is always the master */
+                addReplyArrayLen(c, 3);
+                addReplyBulkCString(c, node->ip);
+                addReplyLongLong(c, node->port);
+                addReplyBulkCBuffer(c, node->name, CLUSTER_NAMELEN);
 
-                    /* If slot exists in output map, add to it's list.
-                    * else, create a new output map for this slot */
-                    if (start == j-1) {
-                        addReplyLongLong(c, start); /* only one slot; low==high */
-                        addReplyLongLong(c, start);
-                    } else {
-                        addReplyLongLong(c, start); /* low */
-                        addReplyLongLong(c, j-1);   /* high */
-                    }
-                    start = -1;
-
-                    /* First node reply position is always the master */
+                /* Remaining nodes in reply are replicas for slot range */
+                for (i = 0; i < node->numslaves; i++) {
+                    /* This loop is copy/pasted from clusterGenNodeDescription()
+                     * with modifications for per-slot node aggregation */
+                    if (nodeFailed(node->slaves[i])) continue;
                     addReplyArrayLen(c, 3);
-                    addReplyBulkCString(c, node->ip);
-                    addReplyLongLong(c, node->port);
-                    addReplyBulkCBuffer(c, node->name, CLUSTER_NAMELEN);
-
-                    /* Remaining nodes in reply are replicas for slot range */
-                    for (i = 0; i < node->numslaves; i++) {
-                        /* This loop is copy/pasted from clusterGenNodeDescription()
-                        * with modifications for per-slot node aggregation */
-                        if (nodeFailed(node->slaves[i])) continue;
-                        addReplyArrayLen(c, 3);
-                        addReplyBulkCString(c, node->slaves[i]->ip);
-                        addReplyLongLong(c, node->slaves[i]->port);
-                        addReplyBulkCBuffer(c, node->slaves[i]->name, CLUSTER_NAMELEN);
-                        nested_elements++;
-                    }
-                    setDeferredArrayLen(c, nested_replylen, nested_elements);
-                    num_masters++;
+                    addReplyBulkCString(c, node->slaves[i]->ip);
+                    addReplyLongLong(c, node->slaves[i]->port);
+                    addReplyBulkCBuffer(c, node->slaves[i]->name, CLUSTER_NAMELEN);
+                    nested_elements++;
                 }
+                setDeferredArrayLen(c, nested_replylen, nested_elements);
+                num_masters++;
             }
         }
-        serverAssert(start == -1);
     }
-    
     dictReleaseIterator(di);
     setDeferredArrayLen(c, slot_replylen, num_masters);
 }

src/fastlock.cpp

@@ -258,10 +258,7 @@ extern "C" void fastlock_init(struct fastlock *lock, const char *name)
     lock->m_depth = 0;
     lock->m_pidOwner = -1;
     lock->futex = 0;
-    int cch = strlen(name);
-    cch = std::min<int>(cch, sizeof(lock->szName)-1);
-    memcpy(lock->szName, name, cch);
-    lock->szName[cch] = '\0';
+    lock->szName = name;
     ANNOTATE_RWLOCK_CREATE(lock);
 }
 

src/fastlock.h

@@ -1,6 +1,5 @@
 #pragma once
 #include <inttypes.h>
-#include <stddef.h>
 
 #ifdef __cplusplus
 extern "C" {
@@ -41,13 +40,12 @@ struct ticket
 
 struct fastlock
 {
+    volatile struct ticket m_ticket;
+
     volatile int m_pidOwner;
     volatile int m_depth;
-    char szName[56];
-    /* Volatile data on seperate cache line */
-    volatile struct ticket m_ticket;
     unsigned futex;
-    char padding[56];   // ensure ticket and futex are on their own independent cache line
+    const char *szName;
 
 #ifdef __cplusplus
     fastlock(const char *name)
@@ -83,5 +81,3 @@ struct fastlock
     bool fOwnLock();   // true if this thread owns the lock, NOTE: not 100% reliable, use for debugging only
 #endif
 };
-
-static_assert(offsetof(struct fastlock, m_ticket) == 64, "ensure padding is correct");

src/fastlock_x64.asm

@@ -1,98 +1,25 @@
 .intel_syntax noprefix
-.section .note.GNU-stack,"",@progbits
+.text
 
 .extern gettid
 .extern fastlock_sleep
 
-.data
-.align 8
-fastlock_unlock_vec:
-	.quad fastlock_unlock_vanilla
-fastlock_lock_vec:
-	.quad cpuid_check_tsxrtm_polyphill
-fastlock_trylock_vec:
-	.quad fastlock_trylock_vanilla
-
-.text
-.global fastlock_unlock
-fastlock_unlock:
-	mov rax, [rip + fastlock_unlock_vec]
-	jmp rax
-.global fastlock_lock
-.type   fastlock_lock,@function
-fastlock_lock:
-	mov rax, [rip + fastlock_lock_vec]
-	jmp rax
-.global fastlock_trylock
-.type   fastlock_trylock,@function
-fastlock_trylock:
-	mov rax, [rip + fastlock_trylock_vec]
-	jmp rax
-
 #	This is the first use of assembly in this codebase, a valid question is WHY?
 #	The spinlock we implement here is performance critical, and simply put GCC
 #	emits awful code.  The original C code is left in fastlock.cpp for reference
 #	and x-plat.
 
-cpuid_check_tsxrtm_polyphill:
-	mov eax, 7
-	xor ecx, ecx
-	cpuid
-	and ebx, 0x800
-	jnz .LHasTSX
-	# CPU does not have TSX RTM support
-	lea rax, [rip+fastlock_lock_vanilla]
-	mov [rip + fastlock_lock_vec], rax
-	jmp rax
-.LHasTSX:
-	lea rax, [rip+fastlock_lock_tsxrtm]
-	mov [rip + fastlock_lock_vec], rax
-	lea rax, [rip+fastlock_trylock_tsxrtm]
-	mov [rip + fastlock_trylock_vec], rax
-	lea rax, [rip+fastlock_unlock_tsxrtm]
-	mov [rip + fastlock_unlock_vec], rax
-	jmp fastlock_lock_tsxrtm
-
 .ALIGN 16
-fastlock_lock_tsxrtm:
-	xbegin fastlock_lock_vanilla
-fastlock_lock_tsxrtmcore:
-	# make sure the lock is not taken (do this inside the TX)
-	mov eax, [rdi+64]
-	mov ecx, eax            # duplicate in ecx
-	ror ecx, 16             # swap upper and lower 16-bits
-	cmp eax, ecx            # are the upper and lower 16-bits the same?
-	jne .LContested
-	mov eax, 1				# this is not needed for the regular lock, but is for try lock which shares this body
-	ret
-.LContested:
-	xabort 0
-
-.ALIGN 16
-fastlock_trylock_tsxrtm:
-	xbegin fastlock_trylock_vanilla
-	jmp fastlock_lock_tsxrtmcore
-
-.ALIGN 16
-fastlock_unlock_tsxrtm:
-	mov eax, [rdi+64]
-	mov ecx, eax            # duplicate in ecx
-	ror ecx, 16             # swap upper and lower 16-bits
-	cmp eax, ecx            # are the upper and lower 16-bits the same?
-	jne fastlock_unlock_vanilla
-	xend
-	ret
-
-.ALIGN 16
-fastlock_lock_vanilla:
+.global fastlock_lock
+.type   fastlock_lock,@function
+fastlock_lock:
 	.cfi_startproc
 	.cfi_def_cfa rsp, 8
 	# RDI points to the struct:
-	#	int32_t m_pidOwner
-	#	int32_t m_depth
-	# [rdi+64] ...
 	#	uint16_t active
 	#	uint16_t avail
+	#	int32_t m_pidOwner
+	#	int32_t m_depth
 	
 	# First get our TID and put it in ecx
 	push rdi                # we need our struct pointer (also balance the stack for the call)
@@ -102,18 +29,18 @@ fastlock_lock_vanilla:
 	pop rdi                 # get our pointer back
 	.cfi_adjust_cfa_offset -8
 
-	cmp [rdi], esi          # Is the TID we got back the owner of the lock?
+	cmp [rdi+4], esi        # Is the TID we got back the owner of the lock?
 	je .LLocked             # Don't spin in that case
 
 	xor eax, eax            # eliminate partial register dependency
 	inc eax                 # we want to add one
-	lock xadd [rdi+66], ax  # do the xadd, ax contains the value before the addition
+	lock xadd [rdi+2], ax   # do the xadd, ax contains the value before the addition
 	# ax now contains the ticket
 	# OK Start the wait loop
 	xor ecx, ecx
 .ALIGN 16
 .LLoop:
-	mov edx, [rdi+64]
+	mov edx, [rdi]
 	cmp dx, ax              # is our ticket up?
 	je .LLocked             # leave the loop
 	pause
@@ -145,19 +72,20 @@ fastlock_lock_vanilla:
 	jmp .LLoop              # Get back in the game
 .ALIGN 16
 .LLocked:
-	mov [rdi], esi          # lock->m_pidOwner = gettid()
-	inc dword ptr [rdi+4]   # lock->m_depth++
+	mov [rdi+4], esi        # lock->m_pidOwner = gettid()
+	inc dword ptr [rdi+8]   # lock->m_depth++
 	ret
 .cfi_endproc
 
 .ALIGN 16
-fastlock_trylock_vanilla:
+.global fastlock_trylock
+.type   fastlock_trylock,@function
+fastlock_trylock:
 	# RDI points to the struct:
-	#	int32_t m_pidOwner
-	#	int32_t m_depth
-	# [rdi+64] ...
 	#	uint16_t active
 	#	uint16_t avail
+	#	int32_t m_pidOwner
+	#	int32_t m_depth
 	
 	# First get our TID and put it in ecx
 	push rdi                # we need our struct pointer (also balance the stack for the call)
@@ -165,29 +93,29 @@ fastlock_trylock_vanilla:
 	mov esi, eax            # back it up in esi
 	pop rdi                 # get our pointer back
 
-	cmp [rdi], esi          # Is the TID we got back the owner of the lock?
+	cmp [rdi+4], esi        # Is the TID we got back the owner of the lock?
 	je .LRecursive          # Don't spin in that case
 
-	mov eax, [rdi+64]       # get both active and avail counters
+	mov eax, [rdi]          # get both active and avail counters
 	mov ecx, eax            # duplicate in ecx
 	ror ecx, 16             # swap upper and lower 16-bits
 	cmp eax, ecx            # are the upper and lower 16-bits the same?
 	jnz .LAlreadyLocked     #	If not return failure
 
 	# at this point we know eax+ecx have [avail][active] and they are both the same
-	add ecx, 0x10000            # increment avail, ecx is now our wanted value
-	lock cmpxchg [rdi+64], ecx  #	If rdi still contains the value in eax, put in ecx (inc avail)
-	jnz .LAlreadyLocked         # If Z is not set then someone locked it while we were preparing
+	add ecx, 0x10000        # increment avail, ecx is now our wanted value
+	lock cmpxchg [rdi], ecx #	If rdi still contains the value in eax, put in ecx (inc avail)
+	jnz .LAlreadyLocked     # If Z is not set then someone locked it while we were preparing
 	xor eax, eax
-	inc eax                     # return SUCCESS! (eax=1)
-	mov [rdi], esi              # lock->m_pidOwner = gettid()
-	mov dword ptr [rdi+4], eax  # lock->m_depth = 1
+	inc eax                 # return SUCCESS! (eax=1)
+	mov [rdi+4], esi        # lock->m_pidOwner = gettid()
+	mov dword ptr [rdi+8], eax  # lock->m_depth = 1
 	ret
 .ALIGN 16
 .LRecursive:
 	xor eax, eax
 	inc eax                 # return SUCCESS! (eax=1)
-	inc dword ptr [rdi+4]   # lock->m_depth++
+	inc dword ptr [rdi+8]   # lock->m_depth++
 	ret
 .ALIGN 16
 .LAlreadyLocked:
@@ -195,27 +123,26 @@ fastlock_trylock_vanilla:
 	ret
 
 .ALIGN 16
-fastlock_unlock_vanilla:
+.global fastlock_unlock
+fastlock_unlock:
 	# RDI points to the struct:
-	#	int32_t m_pidOwner
-	#	int32_t m_depth
-	# [rdi+64] ...
 	#	uint16_t active
 	#	uint16_t avail
+	#	int32_t m_pidOwner
+	#	int32_t m_depth
 	push r11
-	sub dword ptr [rdi+4], 1     # decrement m_depth, don't use dec because it partially writes the flag register and we don't know its state
+	sub dword ptr [rdi+8], 1     # decrement m_depth, don't use dec because it partially writes the flag register and we don't know its state
 	jnz .LDone                   # if depth is non-zero this is a recursive unlock, and we still hold it
-	mov dword ptr [rdi], -1      # pidOwner = -1 (we don't own it anymore)
-	mov ecx, [rdi+64]            # get current active (this one)
+	mov dword ptr [rdi+4], -1    # pidOwner = -1 (we don't own it anymore)
+	mov ecx, [rdi]               # get current active (this one)
 	inc ecx                      # bump it to the next thread
-	mov [rdi+64], cx             # give up our ticket (note: lock is not required here because the spinlock itself guards this variable)
+	mov [rdi], cx                # give up our ticket (note: lock is not required here because the spinlock itself guards this variable)
 	# At this point the lock is removed, however we must wake up any pending futexs
 	mov r9d, 1                   # eax is the bitmask for 2 threads
 	rol r9d, cl                  # place the mask in the right spot for the next 2 threads
-	add rdi, 64                  # rdi now points to the token
 .ALIGN 16
 .LRetryWake:
-	mov r11d, [rdi+4]            # load the futex mask
+	mov r11d, [rdi+12]           # load the futex mask
 	and r11d, r9d                # are any threads waiting on a futex?
 	jz .LDone                    # if not we're done.
 	# we have to wake the futexs

src/networking.cpp

@@ -2521,17 +2521,9 @@ NULL
                 close_this_client = 1;
             } else {
                 if (FCorrectThread(client))
-                {
                     freeClient(client);
-                }
                 else
-                {
-                    int iel = client->iel;
                     freeClientAsync(client);
-                    aePostFunction(g_pserver->rgthreadvar[client->iel].el, [iel] {
-                        freeClientsInAsyncFreeQueue(iel);
-                    });
-                }
             }
             killed++;
         }
@@ -2960,48 +2952,38 @@ void flushSlavesOutputBuffers(void) {
  * than the time left for the previous pause, no change is made to the
  * left duration. */
 void pauseClients(mstime_t end) {
-    serverAssert(GlobalLocksAcquired());
-    if (!serverTL->clients_paused || end > g_pserver->clients_pause_end_time)
+    if (!g_pserver->clients_paused || end > g_pserver->clients_pause_end_time)
         g_pserver->clients_pause_end_time = end;
-    
-    for (int iel = 0; iel < cserver.cthreads; ++iel)
-    {
-        g_pserver->rgthreadvar[iel].clients_paused = true;
-    }
+    g_pserver->clients_paused = 1;
 }
 
 /* Return non-zero if clients are currently paused. As a side effect the
  * function checks if the pause time was reached and clear it. */
 int clientsArePaused(void) {
-    return serverTL->clients_paused;
-}
-
-void unpauseClientsIfNecessary()
-{
-    serverAssert(GlobalLocksAcquired());
-    if (serverTL->clients_paused &&
+    if (g_pserver->clients_paused &&
         g_pserver->clients_pause_end_time < g_pserver->mstime)
     {
+        aeAcquireLock();
         listNode *ln;
         listIter li;
         client *c;
 
-        serverTL->clients_paused = 0;
+        g_pserver->clients_paused = 0;
 
         /* Put all the clients in the unblocked clients queue in order to
          * force the re-processing of the input buffer if any. */
         listRewind(g_pserver->clients,&li);
         while ((ln = listNext(&li)) != NULL) {
             c = (client*)listNodeValue(ln);
-            if (!FCorrectThread(c))
-                continue;
 
             /* Don't touch slaves and blocked clients.
              * The latter pending requests will be processed when unblocked. */
             if (c->flags & (CLIENT_SLAVE|CLIENT_BLOCKED)) continue;
             queueClientForReprocessing(c);
         }
+        aeReleaseLock();
     }
+    return g_pserver->clients_paused;
 }
 
 /* This function is called by Redis in order to process a few events from

src/server.cpp

@@ -1692,9 +1692,6 @@ void clientsCron(int iel) {
             fastlock_unlock(&c->lock);
         }        
     }
-
-    /* Free any pending clients */
-    freeClientsInAsyncFreeQueue(iel);
 }
 
 /* This function handles 'background' operations we are required to do
@@ -1815,9 +1812,6 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
     /* Update the time cache. */
     updateCachedTime();
 
-    /* Unpause clients if enough time has elapsed */
-    unpauseClientsIfNecessary();
-
     g_pserver->hz = g_pserver->config_hz;
     /* Adapt the g_pserver->hz value to the number of configured clients. If we have
      * many clients, we want to call serverCron() with an higher frequency. */
@@ -1825,7 +1819,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
         while (listLength(g_pserver->clients) / g_pserver->hz >
                MAX_CLIENTS_PER_CLOCK_TICK)
         {
-            g_pserver->hz += g_pserver->hz; // *= 2
+            g_pserver->hz *= 2;
             if (g_pserver->hz > CONFIG_MAX_HZ) {
                 g_pserver->hz = CONFIG_MAX_HZ;
                 break;
@@ -2025,6 +2019,9 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
             flushAppendOnlyFile(0);
     }
 
+    /* Clear the paused clients flag if needed. */
+    clientsArePaused(); /* Don't check return value, just use the side effect.*/
+
     /* Replication cron function -- used to reconnect to master,
      * detect transfer failures, start background RDB transfers and so forth. */
     run_with_period(1000) replicationCron();
@@ -2081,9 +2078,6 @@ int serverCronLite(struct aeEventLoop *eventLoop, long long id, void *clientData
     {
         processUnblockedClients(iel);
     }
-
-    /* Unpause clients if enough time has elapsed */
-    unpauseClientsIfNecessary();
     
     ProcessPendingAsyncWrites();    // A bug but leave for now, events should clean up after themselves
     clientsCron(iel);
@@ -2877,7 +2871,6 @@ static void initServerThread(struct redisServerThreadVars *pvar, int fMain)
     pvar->cclients = 0;
     pvar->el = aeCreateEventLoop(g_pserver->maxclients+CONFIG_FDSET_INCR);
     pvar->current_client = nullptr;
-    pvar->clients_paused = 0;
     if (pvar->el == NULL) {
         serverLog(LL_WARNING,
             "Failed creating the event loop. Error message: '%s'",
@@ -2974,6 +2967,7 @@ void initServer(void) {
     g_pserver->ready_keys = listCreate();
     g_pserver->clients_waiting_acks = listCreate();
     g_pserver->get_ack_from_slaves = 0;
+    g_pserver->clients_paused = 0;
     cserver.system_memory_size = zmalloc_get_memory_size();
 
     createSharedObjects();
@@ -4094,7 +4088,7 @@ sds genRedisInfoString(const char *section) {
             g_pserver->port,
             (intmax_t)uptime,
             (intmax_t)(uptime/(3600*24)),
-            g_pserver->hz.load(),
+            g_pserver->hz,
             g_pserver->config_hz,
             (unsigned long) lruclock,
             cserver.executable ? cserver.executable : "",

src/server.h

@@ -1426,7 +1426,6 @@ struct redisServerThreadVars {
     aeEventLoop *el;
     int ipfd[CONFIG_BINDADDR_MAX]; /* TCP socket file descriptors */
     int ipfd_count;             /* Used slots in ipfd[] */
-    int clients_paused;         /* True if clients are currently paused */
     std::vector<client*> clients_pending_write; /* There is to write or install handler. */
     list *unblocked_clients;     /* list of clients to unblock before next loop NOT THREADSAFE */
     list *clients_pending_asyncwrite;
@@ -1519,7 +1518,7 @@ struct redisServer {
     int config_hz;              /* Configured HZ value. May be different than
                                    the actual 'hz' field value if dynamic-hz
                                    is enabled. */
-    std::atomic<int> hz;                     /* serverCron() calls frequency in hertz */
+    int hz;                     /* serverCron() calls frequency in hertz */
     redisDb *db;
     dict *commands;             /* Command table */
     dict *orig_commands;        /* Command table before command renaming. */
@@ -1554,6 +1553,7 @@ struct redisServer {
     list *clients_to_close;     /* Clients to close asynchronously */
     list *slaves, *monitors;    /* List of slaves and MONITORs */
     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 */
     dict *migrate_cached_sockets;/* MIGRATE cached sockets */
     std::atomic<uint64_t> next_client_id; /* Next client unique ID. Incremental. */
@@ -2042,7 +2042,6 @@ void disconnectSlavesExcept(unsigned char *uuid);
 int listenToPort(int port, int *fds, int *count, int fReusePort, int fFirstListen);
 void pauseClients(mstime_t duration);
 int clientsArePaused(void);
-void unpauseClientsIfNecessary();
 int processEventsWhileBlocked(int iel);
 int handleClientsWithPendingWrites(int iel);
 int clientHasPendingReplies(client *c);

src/version.h

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

tests/cluster/run.tcl

@@ -14,8 +14,6 @@ proc main {} {
     spawn_instance redis $::redis_base_port $::instances_count {
         "cluster-enabled yes"
         "appendonly yes"
-	"testmode yes"
-	"server-threads 3"
     }
     run_tests
     cleanup

tests/unit/bitops.tcl

@@ -238,7 +238,7 @@ start_server {tags {"bitops"}} {
         r set a "abcdefg"
         r bitop lshift x a 8
         r get x
-    } "\000abcdefg"
+    } "\x00abcdefg"
 
     test {BITOP lshift char} {
         r set a "\xAA"