writes shouldn't be under the global lock

Former-commit-id: bcfd9327cb4fcf5e1fca9477862919817ddc5ab8

src/module.c

@@ -3702,7 +3702,10 @@ void moduleHandleBlockedClients(void) {
             {
                 c->flags |= CLIENT_PENDING_WRITE;
                 AssertCorrectThread(c);
+                
+                fastlock_lock(&server.rgthreadvar[c->iel].lockPendingWrite);
                 listAddNodeHead(server.rgthreadvar[c->iel].clients_pending_write,c);
+                fastlock_unlock(&server.rgthreadvar[c->iel].lockPendingWrite);
             }
         }
 

src/networking.cpp

@@ -258,6 +258,7 @@ void clientInstallWriteHandler(client *c) {
          * a system call. We'll only really install the write handler if
          * we'll not be able to write the whole reply at once. */
         c->flags |= CLIENT_PENDING_WRITE;
+        std::unique_lock<fastlock> lockf(server.rgthreadvar[c->iel].lockPendingWrite);
         listAddNodeHead(server.rgthreadvar[c->iel].clients_pending_write,c);
     }
 }
@@ -1203,6 +1204,7 @@ void unlinkClient(client *c) {
 
     /* Remove from the list of pending writes if needed. */
     if (c->flags & CLIENT_PENDING_WRITE) {
+        std::unique_lock<fastlock> lockf(server.rgthreadvar[c->iel].lockPendingWrite);
         ln = listSearchKey(server.rgthreadvar[c->iel].clients_pending_write,c);
         serverAssert(ln != NULL);
         listDelNode(server.rgthreadvar[c->iel].clients_pending_write,ln);
@@ -1462,7 +1464,6 @@ int writeToClient(int fd, client *c, int handler_installed) {
             }
             else
             {
-                lock.unlock();
                 freeClientAsync(c);
             }
             
@@ -1567,6 +1568,7 @@ int handleClientsWithPendingWrites(int iel) {
     listIter li;
     listNode *ln;
 
+    std::unique_lock<fastlock> lockf(server.rgthreadvar[iel].lockPendingWrite);
     list *list = server.rgthreadvar[iel].clients_pending_write;
     int processed = listLength(list);
     serverAssert(iel == (serverTL - server.rgthreadvar));

src/server.c

@@ -2109,7 +2109,9 @@ void beforeSleep(struct aeEventLoop *eventLoop) {
     flushAppendOnlyFile(0);
 
     /* Handle writes with pending output buffers. */
+    aeReleaseLock();
     handleClientsWithPendingWrites(IDX_EVENT_LOOP_MAIN);
+    aeAcquireLock();
 
     /* Before we are going to sleep, let the threads access the dataset by
      * releasing the GIL. Redis main thread will not touch anything at this
@@ -2126,10 +2128,10 @@ void beforeSleepLite(struct aeEventLoop *eventLoop)
     if (listLength(server.rgthreadvar[iel].unblocked_clients)) {
         processUnblockedClients(iel);
     }
+    aeReleaseLock();
 
     /* Handle writes with pending output buffers. */
     handleClientsWithPendingWrites(iel);
-    aeReleaseLock();
 }
 
 /* This function is called immadiately after the event loop multiplexing
@@ -2813,6 +2815,8 @@ static void initServerThread(struct redisServerThreadVars *pvar, int fMain)
         exit(1);
     }
 
+    fastlock_init(&pvar->lockPendingWrite);
+
     if (!fMain)
     {
         if (aeCreateTimeEvent(pvar->el, 1, serverCronLite, NULL, NULL) == AE_ERR) {
@@ -5036,6 +5040,7 @@ int main(int argc, char **argv) {
     int background = server.daemonize && !server.supervised;
     if (background) daemonize();
 
+    server.cthreads = 2;
     initServer();
     initNetworking(server.cthreads > 1 /* fReusePort */);
 

src/server.h

@@ -1051,6 +1051,7 @@ struct redisServerThreadVars {
     list *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;
+    struct fastlock lockPendingWrite;
 };
 
 struct redisServer {