Initial implementation of TSXRTM

This reverts commit 41b678814b.

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_ASSER FAILURE\n"); *((volatile int*)0) = 1; } while(0)
+#define AE_ASSERT(x) if (!(x)) do { fprintf(stderr, "AE_ASSERT FAILURE %s: %d\n", __FILE__, __LINE__); *((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,6 +617,17 @@ 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);
     }
@@ -2139,21 +2150,35 @@ 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);
 
-    nwritten = write(fd, link->sndbuf, sdslen(link->sndbuf));
+    // 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();
+
     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(link->sndbuf,nwritten,-1);
+    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;
     if (sdslen(link->sndbuf) == 0)
-        aeDeleteFileEvent(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el, link->fd, AE_WRITABLE);
+        aeDeleteFileEvent(el, link->fd, AE_WRITABLE);
 }
 
 /* Read data. Try to read the first field of the header first to check the
@@ -2228,9 +2253,10 @@ 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)
-        aeCreateFileEvent(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el,link->fd,AE_WRITABLE|AE_BARRIER,
-                    clusterWriteHandler,link);
+        aeCreateRemoteFileEvent(g_pserver->rgthreadvar[IDX_EVENT_LOOP_MAIN].el,link->fd,AE_WRITABLE|AE_BARRIER,
+                    clusterWriteHandler,link,false);
 
     link->sndbuf = sdscatlen(link->sndbuf, msg, msglen);
 
@@ -3284,7 +3310,7 @@ void clusterHandleSlaveMigration(int max_slaves) {
 void resetManualFailover(void) {
     if (g_pserver->cluster->mf_end && clientsArePaused()) {
         g_pserver->clients_pause_end_time = 0;
-        clientsArePaused(); /* Just use the side effect of the function. */
+        unpauseClientsIfNecessary();
     }
     g_pserver->cluster->mf_end = 0; /* No manual failover in progress. */
     g_pserver->cluster->mf_can_start = 0;
@@ -4154,57 +4180,76 @@ void clusterReplyMultiBulkSlots(client *c) {
     dictIterator *di = dictGetSafeIterator(g_pserver->cluster->nodes);
     while((de = dictNext(di)) != NULL) {
         clusterNode *node = (clusterNode*)dictGetVal(de);
-        int j = 0, start = -1;
+        int 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 (j = 0; j < CLUSTER_SLOTS; j++) {
-            int bit, i;
-
-            if ((bit = clusterNodeGetSlotBit(node,j)) != 0) {
-                if (start == -1) start = j;
+        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;
             }
-            if (start != -1 && (!bit || j == CLUSTER_SLOTS-1)) {
-                int nested_elements = 3; /* slots (2) + master addr (1). */
-                void *nested_replylen = addReplyDeferredLen(c);
 
-                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 */
+            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;
                 }
-                start = -1;
+                if (start != -1 && (!bit || j == CLUSTER_SLOTS-1)) {
+                    int nested_elements = 3; /* slots (2) + master addr (1). */
+                    void *nested_replylen = addReplyDeferredLen(c);
 
-                /* 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 (bit && j == CLUSTER_SLOTS-1) j++;
 
-                /* 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;
+                    /* 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 */
                     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++;
+                    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++;
                 }
-                setDeferredArrayLen(c, nested_replylen, nested_elements);
-                num_masters++;
             }
         }
+        serverAssert(start == -1);
     }
+    
     dictReleaseIterator(di);
     setDeferredArrayLen(c, slot_replylen, num_masters);
 }

src/fastlock.cpp

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

src/fastlock.h

@@ -1,5 +1,6 @@
 #pragma once
 #include <inttypes.h>
+#include <stddef.h>
 
 #ifdef __cplusplus
 extern "C" {
@@ -40,12 +41,13 @@ 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;
-    const char *szName;
+    char padding[56];   // ensure ticket and futex are on their own independent cache line
 
 #ifdef __cplusplus
     fastlock(const char *name)
@@ -81,3 +83,5 @@ 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,25 +1,98 @@
 .intel_syntax noprefix
-.text
+.section .note.GNU-stack,"",@progbits
 
 .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
-.global fastlock_lock
-.type   fastlock_lock,@function
-fastlock_lock:
+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:
 	.cfi_startproc
 	.cfi_def_cfa rsp, 8
 	# RDI points to the struct:
-	#	uint16_t active
-	#	uint16_t avail
 	#	int32_t m_pidOwner
 	#	int32_t m_depth
+	# [rdi+64] ...
+	#	uint16_t active
+	#	uint16_t avail
 	
 	# First get our TID and put it in ecx
 	push rdi                # we need our struct pointer (also balance the stack for the call)
@@ -29,18 +102,18 @@ fastlock_lock:
 	pop rdi                 # get our pointer back
 	.cfi_adjust_cfa_offset -8
 
-	cmp [rdi+4], esi        # Is the TID we got back the owner of the lock?
+	cmp [rdi], 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+2], ax   # do the xadd, ax contains the value before the addition
+	lock xadd [rdi+66], 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]
+	mov edx, [rdi+64]
 	cmp dx, ax              # is our ticket up?
 	je .LLocked             # leave the loop
 	pause
@@ -72,20 +145,19 @@ fastlock_lock:
 	jmp .LLoop              # Get back in the game
 .ALIGN 16
 .LLocked:
-	mov [rdi+4], esi        # lock->m_pidOwner = gettid()
-	inc dword ptr [rdi+8]   # lock->m_depth++
+	mov [rdi], esi          # lock->m_pidOwner = gettid()
+	inc dword ptr [rdi+4]   # lock->m_depth++
 	ret
 .cfi_endproc
 
 .ALIGN 16
-.global fastlock_trylock
-.type   fastlock_trylock,@function
-fastlock_trylock:
+fastlock_trylock_vanilla:
 	# RDI points to the struct:
-	#	uint16_t active
-	#	uint16_t avail
 	#	int32_t m_pidOwner
 	#	int32_t m_depth
+	# [rdi+64] ...
+	#	uint16_t active
+	#	uint16_t avail
 	
 	# First get our TID and put it in ecx
 	push rdi                # we need our struct pointer (also balance the stack for the call)
@@ -93,29 +165,29 @@ fastlock_trylock:
 	mov esi, eax            # back it up in esi
 	pop rdi                 # get our pointer back
 
-	cmp [rdi+4], esi        # Is the TID we got back the owner of the lock?
+	cmp [rdi], esi          # Is the TID we got back the owner of the lock?
 	je .LRecursive          # Don't spin in that case
 
-	mov eax, [rdi]          # get both active and avail counters
+	mov eax, [rdi+64]       # 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], 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+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
 	xor eax, eax
-	inc eax                 # return SUCCESS! (eax=1)
-	mov [rdi+4], esi        # lock->m_pidOwner = gettid()
-	mov dword ptr [rdi+8], eax  # lock->m_depth = 1
+	inc eax                     # return SUCCESS! (eax=1)
+	mov [rdi], esi              # lock->m_pidOwner = gettid()
+	mov dword ptr [rdi+4], eax  # lock->m_depth = 1
 	ret
 .ALIGN 16
 .LRecursive:
 	xor eax, eax
 	inc eax                 # return SUCCESS! (eax=1)
-	inc dword ptr [rdi+8]   # lock->m_depth++
+	inc dword ptr [rdi+4]   # lock->m_depth++
 	ret
 .ALIGN 16
 .LAlreadyLocked:
@@ -123,26 +195,27 @@ fastlock_trylock:
 	ret
 
 .ALIGN 16
-.global fastlock_unlock
-fastlock_unlock:
+fastlock_unlock_vanilla:
 	# RDI points to the struct:
-	#	uint16_t active
-	#	uint16_t avail
 	#	int32_t m_pidOwner
 	#	int32_t m_depth
+	# [rdi+64] ...
+	#	uint16_t active
+	#	uint16_t avail
 	push r11
-	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
+	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
 	jnz .LDone                   # if depth is non-zero this is a recursive unlock, and we still hold it
-	mov dword ptr [rdi+4], -1    # pidOwner = -1 (we don't own it anymore)
-	mov ecx, [rdi]               # get current active (this one)
+	mov dword ptr [rdi], -1      # pidOwner = -1 (we don't own it anymore)
+	mov ecx, [rdi+64]            # get current active (this one)
 	inc ecx                      # bump it to the next thread
-	mov [rdi], cx                # give up our ticket (note: lock is not required here because the spinlock itself guards this variable)
+	mov [rdi+64], 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+12]           # load the futex mask
+	mov r11d, [rdi+4]            # 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,9 +2521,17 @@ 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++;
         }
@@ -2952,38 +2960,48 @@ void flushSlavesOutputBuffers(void) {
  * than the time left for the previous pause, no change is made to the
  * left duration. */
 void pauseClients(mstime_t end) {
-    if (!g_pserver->clients_paused || end > g_pserver->clients_pause_end_time)
+    serverAssert(GlobalLocksAcquired());
+    if (!serverTL->clients_paused || end > g_pserver->clients_pause_end_time)
         g_pserver->clients_pause_end_time = end;
-    g_pserver->clients_paused = 1;
+    
+    for (int iel = 0; iel < cserver.cthreads; ++iel)
+    {
+        g_pserver->rgthreadvar[iel].clients_paused = true;
+    }
 }
 
 /* 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) {
-    if (g_pserver->clients_paused &&
+    return serverTL->clients_paused;
+}
+
+void unpauseClientsIfNecessary()
+{
+    serverAssert(GlobalLocksAcquired());
+    if (serverTL->clients_paused &&
         g_pserver->clients_pause_end_time < g_pserver->mstime)
     {
-        aeAcquireLock();
         listNode *ln;
         listIter li;
         client *c;
 
-        g_pserver->clients_paused = 0;
+        serverTL->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/pubsub.cpp

@@ -301,9 +301,11 @@ int pubsubPublishMessage(robj *channel, robj *message) {
             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);
+            if (FCorrectThread(c))
+                fastlock_lock(&c->lock);
             addReplyPubsubMessage(c,channel,message);
-            fastlock_unlock(&c->lock);
+            if (FCorrectThread(c))
+                fastlock_unlock(&c->lock);
             receivers++;
         }
     }
@@ -321,10 +323,12 @@ int pubsubPublishMessage(robj *channel, robj *message) {
             {
                 if (pat->pclient->flags & CLIENT_CLOSE_ASAP)
                     continue;
-                fastlock_lock(&pat->pclient->lock);
+                if (FCorrectThread(pat->pclient))
+                    fastlock_lock(&pat->pclient->lock);
                 addReplyPubsubPatMessage(pat->pclient,
                     pat->pattern,channel,message);
-                fastlock_unlock(&pat->pclient->lock);
+                if (FCorrectThread(pat->pclient))
+                    fastlock_unlock(&pat->pclient->lock);
                 receivers++;
             }
         }

src/server.cpp

@@ -1692,6 +1692,9 @@ void clientsCron(int iel) {
             fastlock_unlock(&c->lock);
         }        
     }
+
+    /* Free any pending clients */
+    freeClientsInAsyncFreeQueue(iel);
 }
 
 /* This function handles 'background' operations we are required to do
@@ -1812,6 +1815,9 @@ 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. */
@@ -1819,7 +1825,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 *= 2;
+            g_pserver->hz += g_pserver->hz; // *= 2
             if (g_pserver->hz > CONFIG_MAX_HZ) {
                 g_pserver->hz = CONFIG_MAX_HZ;
                 break;
@@ -2019,9 +2025,6 @@ 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();
@@ -2078,6 +2081,9 @@ 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);
@@ -2871,6 +2877,7 @@ 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'",
@@ -2967,7 +2974,6 @@ 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();
@@ -4088,7 +4094,7 @@ sds genRedisInfoString(const char *section) {
             g_pserver->port,
             (intmax_t)uptime,
             (intmax_t)(uptime/(3600*24)),
-            g_pserver->hz,
+            g_pserver->hz.load(),
             g_pserver->config_hz,
             (unsigned long) lruclock,
             cserver.executable ? cserver.executable : "",

src/server.h

@@ -1426,6 +1426,7 @@ 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;
@@ -1518,7 +1519,7 @@ struct redisServer {
     int config_hz;              /* Configured HZ value. May be different than
                                    the actual 'hz' field value if dynamic-hz
                                    is enabled. */
-    int hz;                     /* serverCron() calls frequency in hertz */
+    std::atomic<int> hz;                     /* serverCron() calls frequency in hertz */
     redisDb *db;
     dict *commands;             /* Command table */
     dict *orig_commands;        /* Command table before command renaming. */
@@ -1553,7 +1554,6 @@ 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,6 +2042,7 @@ 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);

tests/cluster/run.tcl

@@ -14,6 +14,8 @@ 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
-    } "\x00abcdefg"
+    } "\000abcdefg"
 
     test {BITOP lshift char} {
         r set a "\xAA"