flow: timeout check on flow lookup

src/flow-hash.c

@@ -413,16 +413,6 @@ static inline int FlowCompare(Flow *f, const Packet *p)
 {
     if (p->proto == IPPROTO_ICMP) {
         return FlowCompareICMPv4(f, p);
-    } else if (p->proto == IPPROTO_TCP) {
-        if (CmpFlowPacket(f, p) == 0)
-            return 0;
-
-        /* if this session is 'reused', we don't return it anymore,
-         * so return false on the compare */
-        if (f->flags & FLOW_TCP_REUSED)
-            return 0;
-
-        return 1;
     } else {
         return CmpFlowPacket(f, p);
     }
@@ -638,36 +628,44 @@ Flow *FlowGetFlowFromHash(ThreadVars *tv, DecodeThreadVars *dtv, const Packet *p
     f = fb->head;
     do {
         if (FlowCompare(f, p) != 0) {
-            /* we found our flow, lets put it on top of the
-             * hash list -- this rewards active flows */
-            if (f->hprev) {
-                if (f->hnext) {
-                    f->hnext->hprev = f->hprev;
-                }
-                f->hprev->hnext = f->hnext;
-                if (f == fb->tail) {
-                    fb->tail = f->hprev;
-                }
-
-                f->hnext = fb->head;
-                f->hprev = NULL;
-                fb->head->hprev = f;
-                fb->head = f;
-            }
-            /* found our flow, lock & return */
             FLOWLOCK_WRLOCK(f);
-            if (unlikely(TcpSessionPacketSsnReuse(p, f, f->protoctx) == 1)) {
-                f = TcpReuseReplace(tv, dtv, fb, f, hash, p);
-                if (f == NULL) {
+            if ((f->flags & (FLOW_TCP_REUSED|FLOW_TIMED_OUT)) == 0) {
+                uint32_t timeout = FlowGetFlowTimeout(f, SC_ATOMIC_GET(f->flow_state));
+                int32_t flow_times_out_at = (int32_t)(f->lastts.tv_sec + timeout);
+                /* do the timeout check */
+                if (flow_times_out_at >= p->ts.tv_sec) {
+                    /* we found our flow, lets put it on top of the
+                     * hash list -- this rewards active flows */
+                    if (f->hprev) {
+                        if (f->hnext) {
+                            f->hnext->hprev = f->hprev;
+                        }
+                        f->hprev->hnext = f->hnext;
+                        if (f == fb->tail) {
+                            fb->tail = f->hprev;
+                        }
+
+                        f->hnext = fb->head;
+                        f->hprev = NULL;
+                        fb->head->hprev = f;
+                        fb->head = f;
+                    }
+                    if (unlikely(TcpSessionPacketSsnReuse(p, f, f->protoctx) == 1)) {
+                        f = TcpReuseReplace(tv, dtv, fb, f, hash, p);
+                        if (f == NULL) {
+                            FBLOCK_UNLOCK(fb);
+                            return NULL;
+                        }
+                    }
+
+                    FlowReference(dest, f);
+
                     FBLOCK_UNLOCK(fb);
-                    return NULL;
+                    return f;
                 }
+                f->flags |= FLOW_TIMED_OUT;
             }
-
-            FlowReference(dest, f);
-
-            FBLOCK_UNLOCK(fb);
-            return f;
+            FLOWLOCK_UNLOCK(f);
         }
         if (f->hnext == NULL) {
             pf = f;

src/flow-manager.c

@@ -92,9 +92,6 @@ SCCtrlMutex flow_manager_ctrl_mutex;
 SCCtrlCondT flow_recycler_ctrl_cond;
 SCCtrlMutex flow_recycler_ctrl_mutex;
 
-typedef FlowProtoTimeout *FlowProtoTimeoutPtr;
-SC_ATOMIC_DECLARE(FlowProtoTimeoutPtr, flow_timeouts);
-
 void FlowTimeoutsInit(void)
 {
     SC_ATOMIC_SET(flow_timeouts, flow_timeouts_normal);
@@ -192,41 +189,6 @@ again:
     return;
 }
 
-/** \internal
- *  \brief get timeout for flow
- *
- *  \param f flow
- *  \param state flow state
- *
- *  \retval timeout timeout in seconds
- */
-static inline uint32_t FlowGetFlowTimeout(const Flow *f, enum FlowState state)
-{
-    uint32_t timeout;
-    FlowProtoTimeoutPtr flow_timeouts = SC_ATOMIC_GET(flow_timeouts);
-    switch(state) {
-        default:
-        case FLOW_STATE_NEW:
-            timeout = flow_timeouts[f->protomap].new_timeout;
-            break;
-        case FLOW_STATE_ESTABLISHED:
-            timeout = flow_timeouts[f->protomap].est_timeout;
-            break;
-        case FLOW_STATE_CLOSED:
-            timeout = flow_timeouts[f->protomap].closed_timeout;
-            break;
-#ifdef CAPTURE_OFFLOAD
-        case FLOW_STATE_CAPTURE_BYPASSED:
-            timeout = FLOW_BYPASSED_TIMEOUT;
-            break;
-#endif
-        case FLOW_STATE_LOCAL_BYPASSED:
-            timeout = flow_timeouts[f->protomap].bypassed_timeout;
-            break;
-    }
-    return timeout;
-}
-
 /** \internal
  *  \brief check if a flow is timed out
  *

src/flow-private.h

@@ -100,5 +100,44 @@ extern FlowConfig flow_config;
 /** flow memuse counter (atomic), for enforcing memcap limit */
 SC_ATOMIC_EXTERN(uint64_t, flow_memuse);
 
+typedef FlowProtoTimeout *FlowProtoTimeoutPtr;
+SC_ATOMIC_DECLARE(FlowProtoTimeoutPtr, flow_timeouts);
+
+/** \internal
+ *  \brief get timeout for flow
+ *
+ *  \param f flow
+ *  \param state flow state
+ *
+ *  \retval timeout timeout in seconds
+ */
+static inline uint32_t FlowGetFlowTimeout(const Flow *f, enum FlowState state)
+{
+    uint32_t timeout;
+    FlowProtoTimeoutPtr flow_timeouts = SC_ATOMIC_GET(flow_timeouts);
+    switch(state) {
+        default:
+        case FLOW_STATE_NEW:
+            timeout = flow_timeouts[f->protomap].new_timeout;
+            break;
+        case FLOW_STATE_ESTABLISHED:
+            timeout = flow_timeouts[f->protomap].est_timeout;
+            break;
+        case FLOW_STATE_CLOSED:
+            timeout = flow_timeouts[f->protomap].closed_timeout;
+            break;
+#ifdef CAPTURE_OFFLOAD
+        case FLOW_STATE_CAPTURE_BYPASSED:
+            timeout = FLOW_BYPASSED_TIMEOUT;
+            break;
+#endif
+        case FLOW_STATE_LOCAL_BYPASSED:
+            timeout = flow_timeouts[f->protomap].bypassed_timeout;
+            break;
+    }
+    return timeout;
+}
+
+
 #endif /* __FLOW_PRIVATE_H__ */
 

src/flow.h

@@ -107,6 +107,8 @@ typedef struct AppLayerParserState_ AppLayerParserState;
 #define FLOW_DIR_REVERSED               BIT_U32(26)
 /** Indicate that the flow did trigger an expectation creation */
 #define FLOW_HAS_EXPECTATION            BIT_U32(27)
+/** Make sure flow is not 'found' during flow hash lookup. */
+#define FLOW_TIMED_OUT                  BIT_U32(28)
 
 /* File flags */