update clean up of old detection engine contexts for live rule swap

src/detect-engine.c

@@ -166,10 +166,10 @@ static void *DetectEngineLiveRuleSwap(void *arg)
                 TmThreadsSetFlag(tv_local, THV_CLOSED);
 
                 SCLogInfo("===== Live rule swap premature exit, since "
-                          "suricta_ctl_flags != 0 =====");
-
-                UtilSignalHandlerSetup(SIGUSR2, SignalHandlerSigusr2EngineShutdown);
+                          "engine is in shutdown phase =====");
 
+                UtilSignalHandlerSetup(SIGUSR2, SignalHandlerSigusr2);
+                SCMutexUnlock(&tv_root_lock);
                 pthread_exit(NULL);
             }
 
@@ -214,19 +214,6 @@ static void *DetectEngineLiveRuleSwap(void *arg)
             new_det_ctx[i] = det_ctx;
             i++;
 
-            if (suricata_ctl_flags != 0) {
-                TmThreadsSetFlag(tv_local, THV_CLOSED);
-
-                SCLogInfo("===== Live rule swap premature exit between "
-                          "swapping det_ctxs, since "
-                          "suricta_ctl_flags != 0 =====");
-
-                UtilSignalHandlerSetup(SIGUSR2, SignalHandlerSigusr2EngineShutdown);
-
-                pthread_exit(NULL);
-            }
-
-
             SCLogDebug("swapping new det_ctx - %p with older one - %p", det_ctx,
                        SC_ATOMIC_GET(slots->slot_data));
             SC_ATOMIC_SET(slots->slot_data, det_ctx);
@@ -243,43 +230,54 @@ static void *DetectEngineLiveRuleSwap(void *arg)
               "along with the new de_ctx", no_of_detect_tvs);
 
     for (i = 0; i < no_of_detect_tvs; i++) {
+        int break_out = 0;
         while (new_det_ctx[i]->so_far_used_by_detect != 1) {
-            SCLogDebug("new_det_ctx - %p used by detect", new_det_ctx[i]);
             if (suricata_ctl_flags != 0) {
-                TmThreadsSetFlag(tv_local, THV_CLOSED);
+                break_out = 1;
+                break;
+            }
 
-                SCLogInfo("===== Live rule swap done, but premature exit at "
-                          "de-init phase, since suricta_ctl_flags != 0 =====");
+            usleep(1000);
+        }
+        if (break_out)
+            break;
+        SCLogDebug("new_det_ctx - %p used by detect engine", new_det_ctx[i]);
+    }
 
-                UtilSignalHandlerSetup(SIGUSR2, SignalHandlerSigusr2EngineShutdown);
+    if (i != no_of_detect_tvs) {
+        ThreadVars *tv = tv_root[TVT_PPT];
+        while (tv) {
+            /* obtain the slots for this TV */
+            TmSlot *slots = tv->tm_slots;
+            while (slots != NULL) {
+                TmModule *tm = TmModuleGetById(slots->tm_id);
+
+                if (!(tm->flags & TM_FLAG_DETECT_TM)) {
+                    slots = slots->slot_next;
+                    continue;
+                }
 
-                pthread_exit(NULL);
+                while (!TmThreadsCheckFlag(tv, THV_RUNNING_DONE)) {
+                    usleep(100);
+                }
+
+                slots = slots->slot_next;
             }
 
-            usleep(1000);
+            tv = tv->next;
         }
     }
 
+    /* free all the ctxs */
     DetectEngineCtx *old_de_ctx = old_det_ctx[0]->de_ctx;
     for (i = 0; i < no_of_detect_tvs; i++) {
         SCLogDebug("Freeing old_det_ctx - %p used by detect",
                    old_det_ctx[i]);
-        if (suricata_ctl_flags != 0) {
-            TmThreadsSetFlag(tv_local, THV_CLOSED);
-
-            SCLogInfo("===== Live rule swap done, but premature exit at "
-                      "de-init phase, since suricta_ctl_flags != 0 =====");
-
-            UtilSignalHandlerSetup(SIGUSR2, SignalHandlerSigusr2EngineShutdown);
-
-            pthread_exit(NULL);
-        }
-
-
         DetectEngineThreadCtxDeinit(NULL, old_det_ctx[i]);
     }
     DetectEngineCtxFree(old_de_ctx);
 
+    /* reset the handler */
     UtilSignalHandlerSetup(SIGUSR2, SignalHandlerSigusr2);
 
     TmThreadsSetFlag(tv_local, THV_CLOSED);
@@ -310,6 +308,36 @@ void DetectEngineSpawnLiveRuleSwapMgmtThread(void)
     SCReturn;
 }
 
+DetectEngineCtx *DetectEngineGetGlobalDeCtx(void)
+{
+    DetectEngineCtx *de_ctx = NULL;
+
+    SCMutexLock(&tv_root_lock);
+
+    ThreadVars *tv = tv_root[TVT_PPT];
+    while (tv) {
+        /* obtain the slots for this TV */
+        TmSlot *slots = tv->tm_slots;
+        while (slots != NULL) {
+            TmModule *tm = TmModuleGetById(slots->tm_id);
+
+            if (tm->flags & TM_FLAG_DETECT_TM) {
+                DetectEngineThreadCtx *det_ctx = SC_ATOMIC_GET(slots->slot_data);
+                de_ctx = det_ctx->de_ctx;
+                SCMutexUnlock(&tv_root_lock);
+                return de_ctx;
+            }
+
+            slots = slots->slot_next;
+        }
+
+        tv = tv->next;
+    }
+
+    SCMutexUnlock(&tv_root_lock);
+    return NULL;
+}
+
 DetectEngineCtx *DetectEngineCtxInit(void) {
     DetectEngineCtx *de_ctx;
 

src/detect-engine.h

@@ -30,6 +30,7 @@
 /* prototypes */
 void DetectEngineSpawnLiveRuleSwapMgmtThread(void);
 DetectEngineCtx *DetectEngineCtxInit(void);
+DetectEngineCtx *DetectEngineGetGlobalDeCtx(void);
 void DetectEngineCtxFree(DetectEngineCtx *);
 
 TmEcode DetectEngineThreadCtxInit(ThreadVars *, void *, void **);

src/suricata.c

@@ -241,13 +241,6 @@ static void SignalHandlerSigterm(/*@unused@*/ int sig) {
     suricata_ctl_flags |= SURICATA_KILL;
 }
 
-void SignalHandlerSigusr2EngineShutdown(int sig)
-{
-    SCLogInfo("Live rule swap no longer possible.  Engine in shutdown mode.");
-
-    return;
-}
-
 void SignalHandlerSigusr2SigFileStartup(int sig)
 {
     SCLogInfo("Live rule not possible if -s or -S option used at runtime.");
@@ -275,6 +268,11 @@ void SignalHandlerSigusr2(int sig)
         return;
     }
 
+    if (suricata_ctl_flags != 0) {
+        SCLogInfo("Live rule swap no longer possible.  Engine in shutdown mode.");
+        return;
+    }
+
     UtilSignalHandlerSetup(SIGUSR2, SignalHandlerSigusr2Idle);
 
     DetectEngineSpawnLiveRuleSwapMgmtThread();
@@ -1876,8 +1874,6 @@ int main(int argc, char **argv)
         usleep(10* 1000);
     }
 
-    UtilSignalHandlerSetup(SIGUSR2, SignalHandlerSigusr2EngineShutdown);
-
     /* Update the engine stage/status flag */
     SC_ATOMIC_CAS(&engine_stage, SURICATA_RUNTIME, SURICATA_DEINIT);
 
@@ -1898,7 +1894,26 @@ int main(int argc, char **argv)
         (((1000000 + end_time.tv_usec - start_time.tv_usec) / 1000) - 1000);
     SCLogInfo("time elapsed %.3fs", (float)milliseconds/(float)1000);
 
+    /* Disable detect threads first.  This is required by live rule swap */
+    TmThreadDisableDetectThreads();
+
+    /* wait if live rule swap is in progress */
+    if (UtilSignalIsHandler(SIGUSR2, SignalHandlerSigusr2Idle)) {
+        SCLogInfo("Live rule swap in progress.  Waiting for it to end "
+                  "before we shut the engine/threads down");
+        while (UtilSignalIsHandler(SIGUSR2, SignalHandlerSigusr2Idle)) {
+            /* sleep for 0.5 seconds */
+            usleep(500000);
+        }
+        SCLogInfo("Received notification that live rule swap is done.  "
+                  "Continuing with engine/threads shutdown");
+    }
+
+    DetectEngineCtx *global_de_ctx = DetectEngineGetGlobalDeCtx();
+    BUG_ON(global_de_ctx == NULL);
+
     TmThreadKillThreads();
+
     SCPerfReleaseResources();
     FlowShutdown();
     HostShutdown();
@@ -1941,10 +1956,7 @@ int main(int argc, char **argv)
 
     AppLayerHtpPrintStats();
 
-    /* updated by AS.  Don't clean up de_ctx.  Necessiated by live rule swap */
-#if 0
-    DetectEngineCtxFree(de_ctx);
-#endif
+    DetectEngineCtxFree(global_de_ctx);
     AlpProtoDestroy();
 
     TagDestroyCtx();

src/tm-threads.c

@@ -1536,6 +1536,77 @@ void TmThreadDisableReceiveThreads(void)
     return;
 }
 
+/**
+ * \brief Disable all detect threads.
+ */
+void TmThreadDisableDetectThreads(void)
+{
+    /* value in seconds */
+#define THREAD_KILL_MAX_WAIT_TIME 60
+    /* value in microseconds */
+#define WAIT_TIME 100
+
+    double total_wait_time = 0;
+
+    ThreadVars *tv = NULL;
+
+    SCMutexLock(&tv_root_lock);
+
+    /* all receive threads are part of packet processing threads */
+    tv = tv_root[TVT_PPT];
+
+    /* we do have to keep in mind that TVs are arranged in the order
+     * right from receive to log.  The moment we fail to find a
+     * receive TM amongst the slots in a tv, it indicates we are done
+     * with all receive threads */
+    while (tv) {
+        /* obtain the slots for this TV */
+        TmSlot *slots = tv->tm_slots;
+        while (slots != NULL) {
+            TmModule *tm = TmModuleGetById(slots->tm_id);
+
+            if (!(tm->flags & TM_FLAG_DETECT_TM)) {
+                slots = slots->slot_next;
+                continue;
+            }
+
+            /* we found our receive TV.  Send it a KILL signal.  This is all
+             * we need to do to kill receive threads */
+            TmThreadsSetFlag(tv, THV_KILL);
+
+            if (tv->inq != NULL) {
+                int i;
+                for (i = 0; i < (tv->inq->reader_cnt + tv->inq->writer_cnt); i++) {
+                    if (tv->inq->q_type == 0)
+                        SCCondSignal(&trans_q[tv->inq->id].cond_q);
+                    else
+                        SCCondSignal(&data_queues[tv->inq->id].cond_q);
+                }
+                SCLogDebug("signalled tv->inq->id %" PRIu32 "", tv->inq->id);
+            }
+
+            while (!TmThreadsCheckFlag(tv, THV_RUNNING_DONE)) {
+                usleep(WAIT_TIME);
+                total_wait_time += WAIT_TIME / 1000000.0;
+                if (total_wait_time > THREAD_KILL_MAX_WAIT_TIME) {
+                    SCLogError(SC_ERR_FATAL, "Engine unable to "
+                               "disable detect thread - \"%s\".  "
+                               "Killing engine", tv->name);
+                    exit(EXIT_FAILURE);
+                }
+            }
+
+            break;
+        }
+
+        tv = tv->next;
+    }
+
+    SCMutexUnlock(&tv_root_lock);
+
+    return;
+}
+
 TmSlot *TmThreadGetFirstTmSlotForPartialPattern(const char *tm_name)
 {
     ThreadVars *tv = NULL;

src/tm-threads.h

@@ -113,6 +113,7 @@ TmEcode TmThreadsSlotVarRun (ThreadVars *tv, Packet *p, TmSlot *slot);
 
 ThreadVars *TmThreadsGetTVContainingSlot(TmSlot *);
 void TmThreadDisableReceiveThreads(void);
+void TmThreadDisableDetectThreads(void);
 TmSlot *TmThreadGetFirstTmSlotForPartialPattern(const char *);
 
 /**

src/util-signal.c

@@ -55,3 +55,13 @@ void UtilSignalHandlerSetup(int sig, void (*handler)())
 
     return;
 }
+
+int UtilSignalIsHandler(int sig, void (*handler)())
+{
+    struct sigaction action;
+    memset(&action, 0x00, sizeof(struct sigaction));
+
+    sigaction(sig, NULL, &action);
+
+    return (action.sa_handler == handler);
+}

src/util-signal.h

@@ -26,5 +26,6 @@
 
 int UtilSignalBlock(int);
 void UtilSignalHandlerSetup(int, void (*handler)());;
+int UtilSignalIsHandler(int sig, void (*handler)());
 
 #endif /* __UTIL_STRING_H__ */