Graceful shutdown of server threads when quit is requested

Former-commit-id: b9db899f6ccea62222170c6eec264d403a7a911d

src/db.cpp

@@ -977,7 +977,7 @@ void shutdownCommand(client *c) {
      * Also when in Sentinel mode clear the SAVE flag and force NOSAVE. */
     if (g_pserver->loading || g_pserver->sentinel_mode)
         flags = (flags & ~SHUTDOWN_SAVE) | SHUTDOWN_NOSAVE;
-    if (prepareForShutdown(flags) == C_OK) exit(0);
+    if (prepareForShutdown(flags) == C_OK) throw ShutdownException();
     addReplyError(c,"Errors trying to SHUTDOWN. Check logs.");
 }
 

src/server.cpp

@@ -1894,7 +1894,7 @@ int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
     /* We received a SIGTERM, shutting down here in a safe way, as it is
      * not ok doing so inside the signal handler. */
     if (g_pserver->shutdown_asap) {
-        if (prepareForShutdown(SHUTDOWN_NOFLAGS) == C_OK) exit(0);
+        if (prepareForShutdown(SHUTDOWN_NOFLAGS) == C_OK) throw ShutdownException();
         serverLog(LL_WARNING,"SIGTERM received but errors trying to shut down the server, check the logs for more information");
         g_pserver->shutdown_asap = 0;
     }
@@ -3812,8 +3812,17 @@ int prepareForShutdown(int flags) {
 
     /* Close the listening sockets. Apparently this allows faster restarts. */
     closeListeningSockets(1);
+
+    for (int iel = 0; iel < cserver.cthreads; ++iel)
+    {
+        aePostFunction(g_pserver->rgthreadvar[iel].el, [iel]{
+            g_pserver->rgthreadvar[iel].el->stop = 1;
+        });
+    }
+
     serverLog(LL_WARNING,"%s is now ready to exit, bye bye...",
         g_pserver->sentinel_mode ? "Sentinel" : "KeyDB");
+
     return C_OK;
 }
 
@@ -5033,8 +5042,16 @@ void *workerThreadMain(void *parg)
     aeEventLoop *el = g_pserver->rgthreadvar[iel].el;
     aeSetBeforeSleepProc(el, isMainThread ? beforeSleep : beforeSleepLite, isMainThread ? 0 : AE_SLEEP_THREADSAFE);
     aeSetAfterSleepProc(el, afterSleep, AE_SLEEP_THREADSAFE);
-    aeMain(el);
+    try
+    {
+        aeMain(el);
+    }
+    catch (ShutdownException)
+    {
+    }
+    serverAssert(!GlobalLocksAcquired());
     aeDeleteEventLoop(el);
+
     return NULL;
 }
 
@@ -5334,7 +5351,9 @@ int main(int argc, char **argv) {
     /* The main thread sleeps until all the workers are done.
         this is so that all worker threads are orthogonal in their startup/shutdown */
     void *pvRet;
-    pthread_join(rgthread[IDX_EVENT_LOOP_MAIN], &pvRet);
+    for (int iel = 0; iel < cserver.cthreads; ++iel)
+        pthread_join(rgthread[iel], &pvRet);
+
     return 0;
 }
 

src/server.h

@@ -2922,6 +2922,9 @@ inline int FCorrectThread(client *c)
 }
 #define AssertCorrectThread(c) serverAssert(FCorrectThread(c))
 
+class ShutdownException
+{};
+
 #define redisDebug(fmt, ...) \
     printf("DEBUG %s:%d > " fmt "\n", __FILE__, __LINE__, __VA_ARGS__)
 #define redisDebugMark() \