flow: make TCP reuse handling in flow engine optional

In case of autofp (or more general, when flow and stream engine
run in different threads) the flow engine should not trigger a flow
reuse as this can lead to race conditions between the flow and the
stream engine.

In such cases, the flow engine can be far ahead of the stream engine
as packets are in a queue between the threads.

Observed:

Flow engine tags packet 10 as start of new flow. Flow is tagged as
'reused'.

Stream engine evaluates packet 5 which belongs to the old flow. It
rejects the flow as it's tagged 'reused'. Attaches packet 5 to the
new flow which is wrong.

Solution:

This patch connects the flow engines handling of reuse cases to
the runmode. It hooks into the RunmodeSetFlowStreamAsync() call to
notify the flow engine that it shouldn't handle the reuse.

diff --git a/src/flow-hash.c b/src/flow-hash.c
index 3d609b197754a00a11d433142ee1c695cdac0e08..7a151199268aa6af8c004893ac7b48c2ba4cd1cf 100644 (file)
--- a/src/flow-hash.c
+++ b/src/flow-hash.c
@@ -52,6 +52,7 @@ SC_ATOMIC_EXTERN(unsigned int, flow_prune_idx);
 SC_ATOMIC_EXTERN(unsigned int, flow_flags);
 
 static Flow *FlowGetUsedFlow(ThreadVars *tv, DecodeThreadVars *dtv);
+static int handle_tcp_reuse = 1;
 
 #ifdef FLOW_DEBUG_STATS
 #define FLOW_DEBUG_STATS_PROTO_ALL      0
@@ -148,6 +149,11 @@ void FlowHashDebugDeinit(void)
 
 #endif /* FLOW_DEBUG_STATS */
 
+void FlowDisableTcpReuseHandling(void)
+{
+    handle_tcp_reuse = 0;
+}
+
 /** \brief compare two raw ipv6 addrs
  *
  *  \note we don't care about the real ipv6 ip's, this is just
@@ -405,14 +411,16 @@ static inline int FlowCompare(Flow *f, const Packet *p)
         if (f->flags & FLOW_TCP_REUSED)
             return 0;
 
-        /* lets see if we need to consider the existing session reuse */
-        if (unlikely(TcpSessionPacketSsnReuse(p, f, f->protoctx) == 1)) {
-            /* okay, we need to setup a new flow for this packet.
-             * Flag the flow that it's been replaced by a new one */
-            f->flags |= FLOW_TCP_REUSED;
-            SCLogDebug("flow obsolete: TCP reuse will use a new flow "
-                    "starting with packet %"PRIu64, p->pcap_cnt);
-            return 0;
+        if (handle_tcp_reuse == 1) {
+            /* lets see if we need to consider the existing session reuse */
+            if (unlikely(TcpSessionPacketSsnReuse(p, f, f->protoctx) == 1)) {
+                /* okay, we need to setup a new flow for this packet.
+                 * Flag the flow that it's been replaced by a new one */
+                f->flags |= FLOW_TCP_REUSED;
+                SCLogDebug("flow obsolete: TCP reuse will use a new flow "
+                        "starting with packet %"PRIu64, p->pcap_cnt);
+                return 0;
+            }
         }
         return 1;
     } else {

diff --git a/src/flow-hash.h b/src/flow-hash.h
index a5635b06ce19e64e7cd64dc0813582ab0313af9e..4272896b5a627bf8e91620053028d9b727fcaaaa 100644 (file)
--- a/src/flow-hash.h
+++ b/src/flow-hash.h
@@ -70,6 +70,8 @@ typedef struct FlowBucket_ {
 
 Flow *FlowGetFlowFromHash(ThreadVars *tv, DecodeThreadVars *dtv, const Packet *);
 
+void FlowDisableTcpReuseHandling(void);
+
 /** enable to print stats on hash lookups in flow-debug.log */
 //#define FLOW_DEBUG_STATS
 

diff --git a/src/util-runmodes.c b/src/util-runmodes.c
index 8f1c85964939de81bfa090f9e1380388f63c1fbd..26ef2b76c0dca8cd8b5cd9d4acd1eddfc9e75a80 100644 (file)
--- a/src/util-runmodes.c
+++ b/src/util-runmodes.c
@@ -47,6 +47,8 @@
 
 #include "util-runmodes.h"
 
+#include "flow-hash.h"
+
 /** set to true if flow engine and stream engine run in different
  *  threads. */
 static int runmode_flow_stream_async = 0;
@@ -54,6 +56,7 @@ static int runmode_flow_stream_async = 0;
 void RunmodeSetFlowStreamAsync(void)
 {
     runmode_flow_stream_async = 1;
+    FlowDisableTcpReuseHandling();
 }
 
 int RunmodeGetFlowStreamAsync(void)