flow: fix time handling for non-TCP

Track per flow thread id for UDP and other non-TCP protocols. This
improves the timeout handling as the per thread timestamp is used in
offline mode.

Fixes: ada2bfe00966 ("flow/worker: improve flow timeout time accuracy")
Fixes: ef396f7509cc ("flow/manager: in offline mode, use owning threads time")
Bug #7687.

(cherry picked from commit c648abad0d7393135de0d547b0a8f03ce5af2693)

diff --git a/src/flow-hash.c b/src/flow-hash.c
index 3b221e2ffffc352f86a9fb374273ebcef53f91da..0b642ce5f969603c7134b801f2851409d0d861d1 100644 (file)
--- a/src/flow-hash.c
+++ b/src/flow-hash.c
@@ -758,7 +758,7 @@ static Flow *TcpReuseReplace(ThreadVars *tv, FlowLookupStruct *fls, FlowBucket *
     fb->head = f;
 
     /* initialize and return */
-    FlowInit(f, p);
+    FlowInit(tv, f, p);
     f->flow_hash = hash;
     f->fb = fb;
     FlowUpdateState(f, FLOW_STATE_NEW);
@@ -863,7 +863,7 @@ Flow *FlowGetFlowFromHash(ThreadVars *tv, FlowLookupStruct *fls, Packet *p, Flow
         fb->head = f;
 
         /* got one, now lock, initialize and return */
-        FlowInit(f, p);
+        FlowInit(tv, f, p);
         f->flow_hash = hash;
         f->fb = fb;
         FlowUpdateState(f, FLOW_STATE_NEW);
@@ -928,7 +928,7 @@ flow_removed:
             fb->head = f;
 
             /* initialize and return */
-            FlowInit(f, p);
+            FlowInit(tv, f, p);
             f->flow_hash = hash;
             f->fb = fb;
             FlowUpdateState(f, FLOW_STATE_NEW);

diff --git a/src/flow-util.c b/src/flow-util.c
index dc6a7103a6bd57238d69a727b33ffa3fd87b2854..9b575f399c1003a88e829b6cf2e038841707fa73 100644 (file)
--- a/src/flow-util.c
+++ b/src/flow-util.c
@@ -143,7 +143,7 @@ static inline void FlowSetICMPv6CounterPart(Flow *f)
 
 /* initialize the flow from the first packet
  * we see from it. */
-void FlowInit(Flow *f, const Packet *p)
+void FlowInit(ThreadVars *tv, Flow *f, const Packet *p)
 {
     SCEnter();
     SCLogDebug("flow %p", f);
@@ -152,6 +152,9 @@ void FlowInit(Flow *f, const Packet *p)
     f->recursion_level = p->recursion_level;
     memcpy(&f->vlan_id[0], &p->vlan_id[0], sizeof(f->vlan_id));
     f->vlan_idx = p->vlan_idx;
+
+    f->thread_id[0] = (FlowThreadId)tv->id;
+
     f->livedev = p->livedev;
 
     if (PKT_IS_IPV4(p)) {

diff --git a/src/flow-util.h b/src/flow-util.h
index 3d0d978b5a76bffcf9973e755cb1e1ac502812c3..098f5874737c432f3800302372240d8acfd08700 100644 (file)
--- a/src/flow-util.h
+++ b/src/flow-util.h
@@ -140,7 +140,7 @@
 Flow *FlowAlloc(void);
 void FlowFree(Flow *);
 uint8_t FlowGetProtoMapping(uint8_t);
-void FlowInit(Flow *, const Packet *);
+void FlowInit(ThreadVars *, Flow *, const Packet *);
 uint8_t FlowGetReverseProtoMapping(uint8_t rproto);
 
 /* flow end counter logic */

diff --git a/src/flow.c b/src/flow.c
index 9e910c4f05f920225de3080d47ee51ef41c2d836..2a37a1f60b1372327dca842a5fd4eedf06bd4678 100644 (file)
--- a/src/flow.c
+++ b/src/flow.c
@@ -492,6 +492,9 @@ void FlowHandlePacketUpdate(Flow *f, Packet *p, ThreadVars *tv, DecodeThreadVars
             FlowUpdateTtlTC(f, p, IPV6_GET_HLIM(p));
         }
     }
+    if (f->thread_id[pkt_dir] == 0) {
+        f->thread_id[pkt_dir] = (FlowThreadId)tv->id;
+    }
 
     if (f->flow_state == FLOW_STATE_ESTABLISHED) {
         SCLogDebug("pkt %p FLOW_PKT_ESTABLISHED", p);

diff --git a/src/stream-tcp.c b/src/stream-tcp.c
index f179b41bfcb450159fc0012ee28422ae620ed81e..34ee0108e1fbb6551b876a27484194d46c213def 100644 (file)
--- a/src/stream-tcp.c
+++ b/src/stream-tcp.c
@@ -5333,20 +5333,20 @@ static inline int StreamTcpStateDispatch(
     return 0;
 }
 
-static inline void HandleThreadId(ThreadVars *tv, Packet *p, StreamTcpThread *stt)
+static inline void CheckThreadId(ThreadVars *tv, Packet *p, StreamTcpThread *stt)
 {
     const int idx = (!(PKT_IS_TOSERVER(p)));
 
     /* assign the thread id to the flow */
-    if (unlikely(p->flow->thread_id[idx] == 0)) {
-        p->flow->thread_id[idx] = (FlowThreadId)tv->id;
-    } else if (unlikely((FlowThreadId)tv->id != p->flow->thread_id[idx])) {
-        SCLogDebug("wrong thread: flow has %u, we are %d", p->flow->thread_id[idx], tv->id);
-        if (p->pkt_src == PKT_SRC_WIRE) {
-            StatsIncr(tv, stt->counter_tcp_wrong_thread);
-            if ((p->flow->flags & FLOW_WRONG_THREAD) == 0) {
-                p->flow->flags |= FLOW_WRONG_THREAD;
-                StreamTcpSetEvent(p, STREAM_WRONG_THREAD);
+    if (likely(p->flow->thread_id[idx] != 0)) {
+        if (unlikely((FlowThreadId)tv->id != p->flow->thread_id[idx])) {
+            SCLogDebug("wrong thread: flow has %u, we are %d", p->flow->thread_id[idx], tv->id);
+            if (p->pkt_src == PKT_SRC_WIRE) {
+                StatsIncr(tv, stt->counter_tcp_wrong_thread);
+                if ((p->flow->flags & FLOW_WRONG_THREAD) == 0) {
+                    p->flow->flags |= FLOW_WRONG_THREAD;
+                    StreamTcpSetEvent(p, STREAM_WRONG_THREAD);
+                }
             }
         }
     }
@@ -5772,7 +5772,7 @@ TmEcode StreamTcp (ThreadVars *tv, Packet *p, void *data, PacketQueueNoLock *pq)
         return TM_ECODE_OK;
     }
 
-    HandleThreadId(tv, p, stt);
+    CheckThreadId(tv, p, stt);
 
     /* only TCP packets with a flow from here */