From 2a1a70b3089751b30f623871063ce155451d4cbc Mon Sep 17 00:00:00 2001
From: Jeff Lucovsky <jlucovsky@oisf.net>
Date: Sun, 4 Feb 2024 09:44:44 -0500
Subject: [PATCH] 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
---
 src/tm-threads.c  | 10 ++++++++++
 src/tmqh-simple.c |  5 ++++-
 2 files changed, 14 insertions(+), 1 deletion(-)

diff --git a/src/tm-threads.c b/src/tm-threads.c
index 10ef45da27..e1eee3b641 100644
--- a/src/tm-threads.c
+++ b/src/tm-threads.c
@@ -1239,13 +1239,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;
     }
@@ -1425,7 +1429,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);
             }
@@ -1505,7 +1511,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);
         }
@@ -2296,7 +2304,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 47faed5702..0bfa173e50 100644
--- 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)