threads/mutex: Ensure mutex held before signaling

Ensure that the mutex protecting the condition variable is held before
signaling it. This ensures that the thread(s) awaiting the signal are
notified.

Issue: 6569
(cherry picked from commit 2a1a70b3089751b30f623871063ce155451d4cbc)

diff --git a/src/tm-threads.c b/src/tm-threads.c
index b173cb84f442cdb15f07a3b22cb142920824e7cb..1853db6b035ea13b854b9cf6b5e30ec98e0960b0 100644 (file)
--- a/src/tm-threads.c
+++ b/src/tm-threads.c
@@ -1241,13 +1241,17 @@ static int TmThreadKillThread(ThreadVars *tv)
         }
         if (tv->inq != NULL) {
             for (int i = 0; i < (tv->inq->reader_cnt + tv->inq->writer_cnt); i++) {
+                SCMutexLock(&tv->inq->pq->mutex_q);
                 SCCondSignal(&tv->inq->pq->cond_q);
+                SCMutexUnlock(&tv->inq->pq->mutex_q);
             }
             SCLogDebug("signalled tv->inq->id %" PRIu32 "", tv->inq->id);
         }
 
         if (tv->ctrl_cond != NULL ) {
+            SCCtrlMutexLock(tv->ctrl_mutex);
             pthread_cond_broadcast(tv->ctrl_cond);
+            SCCtrlMutexUnlock(tv->ctrl_mutex);
         }
         return 0;
     }
@@ -1427,7 +1431,9 @@ again:
 
             if (tv->inq != NULL) {
                 for (int i = 0; i < (tv->inq->reader_cnt + tv->inq->writer_cnt); i++) {
+                    SCMutexLock(&tv->inq->pq->mutex_q);
                     SCCondSignal(&tv->inq->pq->cond_q);
+                    SCMutexUnlock(&tv->inq->pq->mutex_q);
                 }
                 SCLogDebug("signalled tv->inq->id %" PRIu32 "", tv->inq->id);
             }
@@ -1507,7 +1513,9 @@ again:
          * THV_KILL flag. */
         if (tv->inq != NULL) {
             for (int i = 0; i < (tv->inq->reader_cnt + tv->inq->writer_cnt); i++) {
+                SCMutexLock(&tv->inq->pq->mutex_q);
                 SCCondSignal(&tv->inq->pq->cond_q);
+                SCMutexUnlock(&tv->inq->pq->mutex_q);
             }
             SCLogDebug("signalled tv->inq->id %" PRIu32 "", tv->inq->id);
         }
@@ -2298,7 +2306,9 @@ void TmThreadsInjectFlowById(Flow *f, const int id)
 
     /* wake up listening thread(s) if necessary */
     if (tv->inq != NULL) {
+        SCMutexLock(&tv->inq->pq->mutex_q);
         SCCondSignal(&tv->inq->pq->cond_q);
+        SCMutexUnlock(&tv->inq->pq->mutex_q);
     } else if (tv->break_loop) {
         TmThreadsCaptureBreakLoop(tv);
     }

diff --git a/src/tmqh-simple.c b/src/tmqh-simple.c
index 47faed5702c5d0b37ff0236e7eca5b4f213e3a8a..0bfa173e5009ef0e904f79f6419d37b71c4dc94e 100644 (file)
--- a/src/tmqh-simple.c
+++ b/src/tmqh-simple.c
@@ -76,8 +76,11 @@ void TmqhInputSimpleShutdownHandler(ThreadVars *tv)
         return;
     }
 
-    for (i = 0; i < (tv->inq->reader_cnt + tv->inq->writer_cnt); i++)
+    for (i = 0; i < (tv->inq->reader_cnt + tv->inq->writer_cnt); i++) {
+        SCMutexLock(&tv->inq->pq->mutex_q);
         SCCondSignal(&tv->inq->pq->cond_q);
+        SCMutexUnlock(&tv->inq->pq->mutex_q);
+    }
 }
 
 void TmqhOutputSimple(ThreadVars *t, Packet *p)