threads: move wait for unpause outside of loop

Threads are only set to paused upon initialization and never again, we
should only have to wait once, so move the wait before any loop that
was waiting before.

Additionally, if the thread was killed while waiting to be unpaused,
don't enter the loop.

diff --git a/src/counters.c b/src/counters.c
index ae0f302b9e84af7f142c03c893748884d8d2fb74..9a845795dded27a61a139bccac46433dda400b0e 100644 (file)
--- a/src/counters.c
+++ b/src/counters.c
@@ -406,9 +406,8 @@ static void *StatsMgmtThread(void *arg)
     SCLogDebug("stats_thread_data %p", &stats_thread_data);
 
     TmThreadsSetFlag(tv_local, THV_INIT_DONE | THV_RUNNING);
-    while (1) {
-        TmThreadsWaitForUnpause(tv_local);
-
+    bool run = TmThreadsWaitForUnpause(tv_local);
+    while (run) {
         struct timeval cur_timev;
         gettimeofday(&cur_timev, NULL);
         struct timespec cond_time = FROM_TIMEVAL(cur_timev);
@@ -483,10 +482,9 @@ static void *StatsWakeupThread(void *arg)
     }
 
     TmThreadsSetFlag(tv_local, THV_INIT_DONE | THV_RUNNING);
+    bool run = TmThreadsWaitForUnpause(tv_local);
 
-    while (1) {
-        TmThreadsWaitForUnpause(tv_local);
-
+    while (run) {
         struct timeval cur_timev;
         gettimeofday(&cur_timev, NULL);
         struct timespec cond_time = FROM_TIMEVAL(cur_timev);

diff --git a/src/detect-engine-loader.c b/src/detect-engine-loader.c
index 6f588deb07cfec62141b311685e0d95d701ecf71..950812a187c925562b76f4e5f8bf7977b3acb989 100644 (file)
--- a/src/detect-engine-loader.c
+++ b/src/detect-engine-loader.c
@@ -595,10 +595,8 @@ static TmEcode DetectLoader(ThreadVars *th_v, void *thread_data)
 
     TmThreadsSetFlag(th_v, THV_INIT_DONE | THV_RUNNING);
     SCLogDebug("loader thread started");
-    while (1)
-    {
-        TmThreadsWaitForUnpause(th_v);
-
+    bool run = TmThreadsWaitForUnpause(th_v);
+    while (run) {
         /* see if we have tasks */
 
         DetectLoaderControl *loader = &loaders[ftd->instance];

diff --git a/src/flow-bypass.c b/src/flow-bypass.c
index 3e0123c8101a56fc886c1f31526040a7272a9723..a57909c6ff295c8b55177650faff7d50c2998125 100644 (file)
--- a/src/flow-bypass.c
+++ b/src/flow-bypass.c
@@ -94,9 +94,9 @@ static TmEcode BypassedFlowManager(ThreadVars *th_v, void *thread_data)
         return TM_ECODE_OK;
 
     TmThreadsSetFlag(th_v, THV_RUNNING);
+    bool run = TmThreadsWaitForUnpause(th_v);
 
-    while (1) {
-        TmThreadsWaitForUnpause(th_v);
+    while (run) {
         SCLogDebug("Dumping the table");
         gettimeofday(&tv, NULL);
         TIMEVAL_TO_TIMESPEC(&tv, &curtime);

diff --git a/src/flow-manager.c b/src/flow-manager.c
index 9448450416ed2ce66bfa3e6c0fb4dcdc5aa4348e..ae60849a84bbfe50f7d4ae51ee5eacd7bae44b08 100644 (file)
--- a/src/flow-manager.c
+++ b/src/flow-manager.c
@@ -817,11 +817,9 @@ static TmEcode FlowManager(ThreadVars *th_v, void *thread_data)
     StatsSetUI64(th_v, ftd->cnt.flow_mgr_rows_sec, rows_sec);
 
     TmThreadsSetFlag(th_v, THV_RUNNING);
+    bool run = TmThreadsWaitForUnpause(th_v);
 
-    while (1)
-    {
-        TmThreadsWaitForUnpause(th_v);
-
+    while (run) {
         bool emerg = ((SC_ATOMIC_GET(flow_flags) & FLOW_EMERGENCY) != 0);
 
         /* Get the time */
@@ -1078,10 +1076,9 @@ static TmEcode FlowRecycler(ThreadVars *th_v, void *thread_data)
     FlowQueuePrivate ret_queue = { NULL, NULL, 0 };
 
     TmThreadsSetFlag(th_v, THV_RUNNING);
+    bool run = TmThreadsWaitForUnpause(th_v);
 
-    while (1)
-    {
-        TmThreadsWaitForUnpause(th_v);
+    while (run) {
         SC_ATOMIC_ADD(flowrec_busy,1);
         FlowQueuePrivate list = FlowQueueExtractPrivate(&flow_recycle_q);
 

diff --git a/src/tm-threads.c b/src/tm-threads.c
index 09bd040d71cc631a9b21a4515d99c3bfc0b4eda3..84ef7d34abee70ef60bdd7520776131aabd82e82 100644 (file)
--- a/src/tm-threads.c
+++ b/src/tm-threads.c
@@ -231,7 +231,6 @@ static void *TmThreadsSlotPktAcqLoop(void *td)
 {
     ThreadVars *tv = (ThreadVars *)td;
     TmSlot *s = tv->tm_slots;
-    char run = 1;
     TmEcode r = TM_ECODE_OK;
     TmSlot *slot = NULL;
 
@@ -303,21 +302,20 @@ static void *TmThreadsSlotPktAcqLoop(void *td)
     StatsSetupPrivate(tv);
 
     TmThreadsSetFlag(tv, THV_INIT_DONE);
+    bool run = TmThreadsWaitForUnpause(tv);
 
-    while(run) {
-        TmThreadsWaitForUnpause(tv);
-
+    while (run) {
         r = s->PktAcqLoop(tv, SC_ATOMIC_GET(s->slot_data), s);
 
         if (r == TM_ECODE_FAILED) {
             TmThreadsSetFlag(tv, THV_FAILED);
-            run = 0;
+            run = false;
         }
         if (TmThreadsCheckFlag(tv, THV_KILL_PKTACQ) || suricata_ctl_flags) {
-            run = 0;
+            run = false;
         }
         if (r == TM_ECODE_DONE) {
-            run = 0;
+            run = false;
         }
     }
     StatsSyncCounters(tv);
@@ -361,7 +359,7 @@ error:
 /**
  * Also returns if the kill flag is set.
  */
-void TmThreadsWaitForUnpause(ThreadVars *tv)
+bool TmThreadsWaitForUnpause(ThreadVars *tv)
 {
     if (TmThreadsCheckFlag(tv, THV_PAUSE)) {
         TmThreadsSetFlag(tv, THV_PAUSED);
@@ -370,11 +368,13 @@ void TmThreadsWaitForUnpause(ThreadVars *tv)
             SleepUsec(100);
 
             if (TmThreadsCheckFlag(tv, THV_KILL))
-                break;
+                return false;
         }
 
         TmThreadsUnsetFlag(tv, THV_PAUSED);
     }
+
+    return true;
 }
 
 static void *TmThreadsSlotVar(void *td)
@@ -382,7 +382,6 @@ static void *TmThreadsSlotVar(void *td)
     ThreadVars *tv = (ThreadVars *)td;
     TmSlot *s = (TmSlot *)tv->tm_slots;
     Packet *p = NULL;
-    char run = 1;
     TmEcode r = TM_ECODE_OK;
 
     CaptureStatsSetup(tv);
@@ -453,12 +452,11 @@ static void *TmThreadsSlotVar(void *td)
     // enter infinite loops. They use this as the core loop. As a result, at this
     // point the worker threads can be considered both initialized and running.
     TmThreadsSetFlag(tv, THV_INIT_DONE | THV_RUNNING);
+    bool run = TmThreadsWaitForUnpause(tv);
 
     s = (TmSlot *)tv->tm_slots;
 
     while (run) {
-        TmThreadsWaitForUnpause(tv);
-
         /* input a packet */
         p = tv->tmqh_in(tv);
 
@@ -490,7 +488,7 @@ static void *TmThreadsSlotVar(void *td)
         }
 
         if (TmThreadsCheckFlag(tv, THV_KILL)) {
-            run = 0;
+            run = false;
         }
     } /* while (run) */
     StatsSyncCounters(tv);

diff --git a/src/tm-threads.h b/src/tm-threads.h
index 2a46b133c4ccf028bfe2b0c4c2349f44009a8143..63fbef85b0a3302f56b02d6b691bda788c7758ac 100644 (file)
--- a/src/tm-threads.h
+++ b/src/tm-threads.h
@@ -291,7 +291,9 @@ bool TmThreadsTimeSubsysIsReady(void);
  *
  * Check if a thread should wait to be unpaused and wait if so, or
  * until the thread kill flag is set.
+ *
+ * \returns true if the thread was unpaused, false if killed.
  */
-void TmThreadsWaitForUnpause(ThreadVars *tv);
+bool TmThreadsWaitForUnpause(ThreadVars *tv);
 
 #endif /* SURICATA_TM_THREADS_H */