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: ada2bfe009 ("flow/worker: improve flow timeout time accuracy")
Fixes: ef396f7509 ("flow/manager: in offline mode, use owning threads time")

Bug #7687.

src/flow-util.c

@@ -153,6 +153,9 @@ void FlowInit(ThreadVars *tv, 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 (PacketIsIPv4(p)) {

src/flow.c

@@ -482,6 +482,9 @@ void FlowHandlePacketUpdate(Flow *f, Packet *p, ThreadVars *tv, DecodeThreadVars
             FlowUpdateTtlTC(f, IPV6_GET_RAW_HLIM(ip6h));
         }
     }
+    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);

src/stream-tcp.c

@@ -5519,20 +5519,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)) {
+    if (likely(p->flow->thread_id[idx] != 0)) {
-        p->flow->thread_id[idx] = (FlowThreadId)tv->id;
+        if (unlikely((FlowThreadId)tv->id != p->flow->thread_id[idx])) {
-    } 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);
-        SCLogDebug("wrong thread: flow has %u, we are %d", p->flow->thread_id[idx], tv->id);
+            if (p->pkt_src == PKT_SRC_WIRE) {
-        if (p->pkt_src == PKT_SRC_WIRE) {
+                StatsIncr(tv, stt->counter_tcp_wrong_thread);
-            StatsIncr(tv, stt->counter_tcp_wrong_thread);
+                if ((p->flow->flags & FLOW_WRONG_THREAD) == 0) {
-            if ((p->flow->flags & FLOW_WRONG_THREAD) == 0) {
+                    p->flow->flags |= FLOW_WRONG_THREAD;
-                p->flow->flags |= FLOW_WRONG_THREAD;
+                    StreamTcpSetEvent(p, STREAM_WRONG_THREAD);
-                StreamTcpSetEvent(p, STREAM_WRONG_THREAD);
+                }
             }
         }
     }
@@ -5959,7 +5959,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 */