flow/worker: refresh detect thread during housekeeping

During housekeeping multiple flows are processed. If a rule reload happens
at that time, we need to use the new detect thread as soon as possible.

Bug: #5969.

diff --git a/src/flow-worker.c b/src/flow-worker.c
index 4b0b7baf89684911c3b9637baec9881dfe40c0f8..b6676e6aa7e62f3a7c292569d2413342829a40ae 100644 (file)
--- a/src/flow-worker.c
+++ b/src/flow-worker.c
@@ -170,9 +170,8 @@ static int FlowFinish(ThreadVars *tv, Flow *f, FlowWorkerThreadData *fw, void *d
 }
 
 /** \param[in] max_work Max flows to process. 0 if unlimited. */
-static void CheckWorkQueue(ThreadVars *tv, FlowWorkerThreadData *fw,
-        void *detect_thread, // TODO proper type?
-        FlowTimeoutCounters *counters, FlowQueuePrivate *fq, const uint32_t max_work)
+static void CheckWorkQueue(ThreadVars *tv, FlowWorkerThreadData *fw, FlowTimeoutCounters *counters,
+        FlowQueuePrivate *fq, const uint32_t max_work)
 {
     uint32_t i = 0;
     Flow *f;
@@ -183,6 +182,8 @@ static void CheckWorkQueue(ThreadVars *tv, FlowWorkerThreadData *fw,
         if (f->proto == IPPROTO_TCP) {
             if (!(f->flags & FLOW_TIMEOUT_REASSEMBLY_DONE) && !FlowIsBypassed(f) &&
                     FlowForceReassemblyNeedReassembly(f) == 1 && f->ffr != 0) {
+                /* read detect thread in case we're doing a reload */
+                void *detect_thread = SC_ATOMIC_GET(fw->detect_thread);
                 int cnt = FlowFinish(tv, f, fw, detect_thread);
                 counters->flows_aside_pkt_inject += cnt;
                 counters->flows_aside_needs_work++;
@@ -481,8 +482,7 @@ static inline void FlowWorkerProcessInjectedFlows(
 /** \internal
  *  \brief process flows set aside locally during flow lookup
  */
-static inline void FlowWorkerProcessLocalFlows(ThreadVars *tv,
-        FlowWorkerThreadData *fw, Packet *p, void *detect_thread)
+static inline void FlowWorkerProcessLocalFlows(ThreadVars *tv, FlowWorkerThreadData *fw, Packet *p)
 {
     uint32_t max_work = 2;
     if (PKT_IS_PSEUDOPKT(p))
@@ -493,7 +493,7 @@ static inline void FlowWorkerProcessLocalFlows(ThreadVars *tv,
         StatsAddUI64(tv, fw->cnt.flows_removed, (uint64_t)fw->fls.work_queue.len);
 
         FlowTimeoutCounters counters = { 0, 0, };
-        CheckWorkQueue(tv, fw, detect_thread, &counters, &fw->fls.work_queue, max_work);
+        CheckWorkQueue(tv, fw, &counters, &fw->fls.work_queue, max_work);
         UpdateCounters(tv, fw, &counters);
     }
     FLOWWORKER_PROFILING_END(p, PROFILE_FLOWWORKER_FLOW_EVICTED);
@@ -612,7 +612,7 @@ housekeeping:
     FlowWorkerProcessInjectedFlows(tv, fw, p);
 
     /* process local work queue */
-    FlowWorkerProcessLocalFlows(tv, fw, p, detect_thread);
+    FlowWorkerProcessLocalFlows(tv, fw, p);
 
     return TM_ECODE_OK;
 }