threading: force break loop on flow inject

Track availability of break loop callback to avoid overhead.

(cherry picked from commit ff97d7c15da0a8a7b8ea1a0d461f4f56ca2052d6)

diff --git a/src/threadvars.h b/src/threadvars.h
index 1089dd132da69598d650cca74ee92093deb2000c..8b604445ed55f6b6d4675d3d04f999fd8b1ee602 100644 (file)
--- a/src/threadvars.h
+++ b/src/threadvars.h
@@ -132,6 +132,7 @@ typedef struct ThreadVars_ {
     SCCtrlCondT *ctrl_cond;
 
     struct FlowQueue_ *flow_queue;
+    bool break_loop;
 
 } ThreadVars;
 

diff --git a/src/tm-threads.c b/src/tm-threads.c
index 21b40bad6ebac40b48e330979d04aadd948278b0..83ce1e7ac7fc5bae38fd07465bc52d028cb62d71 100644 (file)
--- a/src/tm-threads.c
+++ b/src/tm-threads.c
@@ -668,6 +668,9 @@ void TmSlotSetFuncAppend(ThreadVars *tv, TmModule *tm, const void *data)
         slot->SlotFunc = tm->Func;
     } else if (tm->PktAcqLoop) {
         slot->PktAcqLoop = tm->PktAcqLoop;
+        if (tm->PktAcqBreakLoop) {
+            tv->break_loop = true;
+        }
     } else if (tm->Management) {
         slot->Management = tm->Management;
     }
@@ -2280,6 +2283,23 @@ uint16_t TmThreadsGetWorkerThreadMax()
     return thread_max;
 }
 
+static inline void ThreadBreakLoop(ThreadVars *tv)
+{
+    if ((tv->tmm_flags & TM_FLAG_RECEIVE_TM) == 0) {
+        return;
+    }
+    /* find the correct slot */
+    TmSlot *s = tv->tm_slots;
+    TmModule *tm = TmModuleGetById(s->tm_id);
+    if (tm->flags & TM_FLAG_RECEIVE_TM) {
+        /* if the method supports it, BreakLoop. Otherwise we rely on
+         * the capture method's recv timeout */
+        if (tm->PktAcqLoop && tm->PktAcqBreakLoop) {
+            tm->PktAcqBreakLoop(tv, SC_ATOMIC_GET(s->slot_data));
+        }
+    }
+}
+
 /**
  *  \retval r 1 if packet was accepted, 0 otherwise
  *  \note if packet was not accepted, it's still the responsibility
@@ -2308,6 +2328,8 @@ int TmThreadsInjectPacketsById(Packet **packets, const int id)
     /* wake up listening thread(s) if necessary */
     if (tv->inq != NULL) {
         SCCondSignal(&tv->inq->pq->cond_q);
+    } else if (tv->break_loop) {
+        ThreadBreakLoop(tv);
     }
     return 1;
 }
@@ -2330,5 +2352,7 @@ void TmThreadsInjectFlowById(Flow *f, const int id)
     /* wake up listening thread(s) if necessary */
     if (tv->inq != NULL) {
         SCCondSignal(&tv->inq->pq->cond_q);
+    } else if (tv->break_loop) {
+        ThreadBreakLoop(tv);
     }
 }