updated to fix unix shutdown sequence

Should fix crashes occuring from unix mode shutdown/cleanup phase.

diff --git a/src/app-layer-detect-proto.c b/src/app-layer-detect-proto.c
index 3baad6f3d9aefa2048238593cf39b03b9bc50152..bd5ebd41ac090fa8f79448bf668567e2ea9f09d7 100644 (file)
--- a/src/app-layer-detect-proto.c
+++ b/src/app-layer-detect-proto.c
@@ -698,7 +698,11 @@ void *AppLayerDetectProtoThread(void *td)
 
     /* main loop */
     while(run) {
-        TmThreadTestThreadUnPaused(tv);
+        if (TmThreadsCheckFlag(tv, THV_PAUSE)) {
+            TmThreadsSetFlag(tv, THV_PAUSED);
+            TmThreadTestThreadUnPaused(tv);
+            TmThreadsUnsetFlag(tv, THV_PAUSED);
+        }
 
         /* grab a msg, can return NULL on signals */
         StreamMsg *smsg = StreamMsgGetFromQueue(stream_q);

diff --git a/src/counters.c b/src/counters.c
index 2c260b09eb373296ecb8931d93d060c0062bc953..eb0287646d02a70aa5908ace9bcbff7ddce30abd 100644 (file)
--- a/src/counters.c
+++ b/src/counters.c
@@ -465,7 +465,11 @@ static void *SCPerfMgmtThread(void *arg)
 
     TmThreadsSetFlag(tv_local, THV_INIT_DONE);
     while (run) {
-        TmThreadTestThreadUnPaused(tv_local);
+        if (TmThreadsCheckFlag(tv_local, THV_PAUSE)) {
+            TmThreadsSetFlag(tv_local, THV_PAUSED);
+            TmThreadTestThreadUnPaused(tv_local);
+            TmThreadsUnsetFlag(tv_local, THV_PAUSED);
+        }
 
         cond_time.tv_sec = time(NULL) + sc_counter_tts;
         cond_time.tv_nsec = 0;
@@ -529,7 +533,11 @@ static void *SCPerfWakeupThread(void *arg)
 
     TmThreadsSetFlag(tv_local, THV_INIT_DONE);
     while (run) {
-        TmThreadTestThreadUnPaused(tv_local);
+        if (TmThreadsCheckFlag(tv_local, THV_PAUSE)) {
+            TmThreadsSetFlag(tv_local, THV_PAUSED);
+            TmThreadTestThreadUnPaused(tv_local);
+            TmThreadsUnsetFlag(tv_local, THV_PAUSED);
+        }
 
         cond_time.tv_sec = time(NULL) + SC_PERF_WUT_TTS;
         cond_time.tv_nsec = 0;

diff --git a/src/flow-manager.c b/src/flow-manager.c
index ac4f22b1f3055e5b9a3312b04693c3f7ccf008d1..0f57430dd71b90db5571205e8541940984146810 100644 (file)
--- a/src/flow-manager.c
+++ b/src/flow-manager.c
@@ -445,7 +445,11 @@ void *FlowManagerThread(void *td)
     TmThreadsSetFlag(th_v, THV_INIT_DONE);
     while (1)
     {
-        TmThreadTestThreadUnPaused(th_v);
+        if (TmThreadsCheckFlag(th_v, THV_PAUSE)) {
+            TmThreadsSetFlag(th_v, THV_PAUSED);
+            TmThreadTestThreadUnPaused(th_v);
+            TmThreadsUnsetFlag(th_v, THV_PAUSED);
+        }
 
         if (SC_ATOMIC_GET(flow_flags) & FLOW_EMERGENCY) {
             emerg = TRUE;

diff --git a/src/flow-timeout.c b/src/flow-timeout.c
index 6ef218e4f5fdd42f2e0ae657045a6802b7703ae5..af298f2df1c23111d349ddd9ec145e246196a6a6 100644 (file)
--- a/src/flow-timeout.c
+++ b/src/flow-timeout.c
@@ -681,6 +681,19 @@ void FlowForceReassembly(void)
                 while (q->len != 0) {
                     usleep(100);
                 }
+                TmThreadsSetFlag(tv, THV_PAUSE);
+                if (tv->inq->q_type == 0)
+                    SCCondSignal(&trans_q[tv->inq->id].cond_q);
+                else
+                    SCCondSignal(&data_queues[tv->inq->id].cond_q);
+                while (!TmThreadsCheckFlag(tv, THV_PAUSED)) {
+                    if (tv->inq->q_type == 0)
+                        SCCondSignal(&trans_q[tv->inq->id].cond_q);
+                    else
+                        SCCondSignal(&data_queues[tv->inq->id].cond_q);
+                    usleep(100);
+                }
+                TmThreadsUnsetFlag(tv, THV_PAUSE);
             }
         }
         tv = tv->next;

diff --git a/src/runmode-unix-socket.c b/src/runmode-unix-socket.c
index 83259c908a1e87026608abe134ff907a02884054..5b2c80b4346e532504172e01e5d0fa1e416f68e3 100644 (file)
--- a/src/runmode-unix-socket.c
+++ b/src/runmode-unix-socket.c
@@ -270,6 +270,7 @@ TmEcode UnixSocketPcapFilesCheck(void *data)
         this->running = 0;
         TmThreadKillThreadsFamily(TVT_MGMT);
         TmThreadClearThreadsFamily(TVT_MGMT);
+        TmThreadDisableThreadsWithTMS(TM_FLAG_RECEIVE_TM | TM_FLAG_DECODE_TM);
         FlowForceReassembly();
         TmThreadKillThreadsFamily(TVT_PPT);
         TmThreadClearThreadsFamily(TVT_PPT);

diff --git a/src/threadvars.h b/src/threadvars.h
index b119204fc7ae84b70dfe668113852e01e7707a21..293322d9bc042de65c688193a5e90f6632a42c22 100644 (file)
--- a/src/threadvars.h
+++ b/src/threadvars.h
@@ -33,17 +33,18 @@
 struct TmSlot_;
 
 /** Thread flags set and read by threads to control the threads */
-#define THV_USE       0x01 /** thread is in use */
-#define THV_INIT_DONE 0x02 /** thread initialization done */
-#define THV_PAUSE     0x04 /** thread has been paused */
-#define THV_KILL      0x08 /** thread has been asked to cleanup and exit */
-#define THV_FAILED    0x10 /** thread has encountered an error and failed */
-#define THV_CLOSED    0x20 /** thread done, should be joinable */
+#define THV_USE       1 /** thread is in use */
+#define THV_INIT_DONE (1 << 1) /** thread initialization done */
+#define THV_PAUSE     (1 << 2) /** signal thread to pause itself */
+#define THV_PAUSED    (1 << 3) /** the thread is paused atm */
+#define THV_KILL      (1 << 4) /** thread has been asked to cleanup and exit */
+#define THV_FAILED    (1 << 5) /** thread has encountered an error and failed */
+#define THV_CLOSED    (1 << 6) /** thread done, should be joinable */
 /* used to indicate the thread is going through de-init.  Introduced as more
  * of a hack for solving stream-timeout-shutdown.  Is set by the main thread. */
-#define THV_DEINIT    0x40
-#define THV_RUNNING_DONE 0x80 /** thread has completed running and is entering
-                              * the de-init phase */
+#define THV_DEINIT    (1 << 7)
+#define THV_RUNNING_DONE (1 << 8) /** thread has completed running and is entering
+                                   * the de-init phase */
 
 /** Thread flags set and read by threads, to control the threads, when they
  *  encounter certain conditions like failure */
@@ -59,7 +60,7 @@ typedef struct ThreadVars_ {
     char *name;
     char *thread_group_name;
 
-    SC_ATOMIC_DECLARE(unsigned char, flags);
+    SC_ATOMIC_DECLARE(unsigned short, flags);
 
     /** aof(action on failure) determines what should be done with the thread
         when it encounters certain conditions like failures */

diff --git a/src/tm-threads.c b/src/tm-threads.c
index c926e0fa2c22a61d38cca16b646b093f5668fc89..22bb04542d8b2ffe3df1d4a90f9c9fba63792776 100644 (file)
--- a/src/tm-threads.c
+++ b/src/tm-threads.c
@@ -89,7 +89,7 @@ uint8_t tv_aof = THV_RESTART_THREAD;
  * \retval 1 flag is set.
  * \retval 0 flag is not set.
  */
-int TmThreadsCheckFlag(ThreadVars *tv, uint8_t flag)
+int TmThreadsCheckFlag(ThreadVars *tv, uint16_t flag)
 {
     return (SC_ATOMIC_GET(tv->flags) & flag) ? 1 : 0;
 }
@@ -97,7 +97,7 @@ int TmThreadsCheckFlag(ThreadVars *tv, uint8_t flag)
 /**
  * \brief Set a thread flag.
  */
-void TmThreadsSetFlag(ThreadVars *tv, uint8_t flag)
+void TmThreadsSetFlag(ThreadVars *tv, uint16_t flag)
 {
     SC_ATOMIC_OR(tv->flags, flag);
 }
@@ -105,7 +105,7 @@ void TmThreadsSetFlag(ThreadVars *tv, uint8_t flag)
 /**
  * \brief Unset a thread flag.
  */
-void TmThreadsUnsetFlag(ThreadVars *tv, uint8_t flag)
+void TmThreadsUnsetFlag(ThreadVars *tv, uint16_t flag)
 {
     SC_ATOMIC_AND(tv->flags, ~flag);
 }
@@ -160,7 +160,11 @@ void *TmThreadsSlot1NoIn(void *td)
     TmThreadsSetFlag(tv, THV_INIT_DONE);
 
     while (run) {
-        TmThreadTestThreadUnPaused(tv);
+        if (TmThreadsCheckFlag(tv, THV_PAUSE)) {
+            TmThreadsSetFlag(tv, THV_PAUSED);
+            TmThreadTestThreadUnPaused(tv);
+            TmThreadsUnsetFlag(tv, THV_PAUSED);
+        }
         TmSlotFunc SlotFunc = SC_ATOMIC_GET(s->SlotFunc);
 
         r = SlotFunc(tv, NULL, SC_ATOMIC_GET(s->slot_data), &s->slot_pre_pq, &s->slot_post_pq);
@@ -262,7 +266,11 @@ void *TmThreadsSlot1NoOut(void *td)
     TmThreadsSetFlag(tv, THV_INIT_DONE);
 
     while (run) {
-        TmThreadTestThreadUnPaused(tv);
+        if (TmThreadsCheckFlag(tv, THV_PAUSE)) {
+            TmThreadsSetFlag(tv, THV_PAUSED);
+            TmThreadTestThreadUnPaused(tv);
+            TmThreadsUnsetFlag(tv, THV_PAUSED);
+        }
         TmSlotFunc SlotFunc = SC_ATOMIC_GET(s->SlotFunc);
 
         p = tv->tmqh_in(tv);
@@ -348,7 +356,11 @@ void *TmThreadsSlot1NoInOut(void *td)
 
     while (run) {
         TmSlotFunc SlotFunc = SC_ATOMIC_GET(s->SlotFunc);
-        TmThreadTestThreadUnPaused(tv);
+        if (TmThreadsCheckFlag(tv, THV_PAUSE)) {
+            TmThreadsSetFlag(tv, THV_PAUSED);
+            TmThreadTestThreadUnPaused(tv);
+            TmThreadsUnsetFlag(tv, THV_PAUSED);
+        }
 
         r = SlotFunc(tv, NULL, SC_ATOMIC_GET(s->slot_data), /* no outqh, no pq */NULL, NULL);
 
@@ -428,7 +440,11 @@ void *TmThreadsSlot1(void *td)
 
     TmThreadsSetFlag(tv, THV_INIT_DONE);
     while (run) {
-        TmThreadTestThreadUnPaused(tv);
+        if (TmThreadsCheckFlag(tv, THV_PAUSE)) {
+            TmThreadsSetFlag(tv, THV_PAUSED);
+            TmThreadTestThreadUnPaused(tv);
+            TmThreadsUnsetFlag(tv, THV_PAUSED);
+        }
 
         /* input a packet */
         p = tv->tmqh_in(tv);
@@ -657,7 +673,11 @@ void *TmThreadsSlotPktAcqLoop(void *td) {
     TmThreadsSetFlag(tv, THV_INIT_DONE);
 
     while(run) {
-        TmThreadTestThreadUnPaused(tv);
+        if (TmThreadsCheckFlag(tv, THV_PAUSE)) {
+            TmThreadsSetFlag(tv, THV_PAUSED);
+            TmThreadTestThreadUnPaused(tv);
+            TmThreadsUnsetFlag(tv, THV_PAUSED);
+        }
 
         r = s->PktAcqLoop(tv, SC_ATOMIC_GET(s->slot_data), s);
 
@@ -755,7 +775,11 @@ void *TmThreadsSlotVar(void *td)
     s = (TmSlot *)tv->tm_slots;
 
     while (run) {
-        TmThreadTestThreadUnPaused(tv);
+        if (TmThreadsCheckFlag(tv, THV_PAUSE)) {
+            TmThreadsSetFlag(tv, THV_PAUSED);
+            TmThreadTestThreadUnPaused(tv);
+            TmThreadsUnsetFlag(tv, THV_PAUSED);
+        }
 
         /* input a packet */
         p = tv->tmqh_in(tv);
@@ -1954,7 +1978,7 @@ void TmThreadTestThreadUnPaused(ThreadVars *tv)
  *
  * \param tv Pointer to the TV instance.
  */
-void TmThreadWaitForFlag(ThreadVars *tv, uint8_t flags)
+void TmThreadWaitForFlag(ThreadVars *tv, uint16_t flags)
 {
     while (!TmThreadsCheckFlag(tv, flags)) {
         usleep(100);

diff --git a/src/tm-threads.h b/src/tm-threads.h
index 804a9b2e2013699220534ddaa33bf3cfb8568781..2653dbb84cd2c2763526e5488efe749f8de4ba5d 100644 (file)
--- a/src/tm-threads.h
+++ b/src/tm-threads.h
@@ -113,10 +113,10 @@ ThreadVars *TmThreadsGetCallingThread(void);
 void TmThreadActivateDummySlot(void);
 void TmThreadDeActivateDummySlot(void);
 
-int TmThreadsCheckFlag(ThreadVars *, uint8_t);
-void TmThreadsSetFlag(ThreadVars *, uint8_t);
-void TmThreadsUnsetFlag(ThreadVars *, uint8_t);
-void TmThreadWaitForFlag(ThreadVars *, uint8_t);
+int TmThreadsCheckFlag(ThreadVars *, uint16_t);
+void TmThreadsSetFlag(ThreadVars *, uint16_t);
+void TmThreadsUnsetFlag(ThreadVars *, uint16_t);
+void TmThreadWaitForFlag(ThreadVars *, uint16_t);
 
 TmEcode TmThreadsSlotVarRun (ThreadVars *tv, Packet *p, TmSlot *slot);