From 7426a9c64504ac2731c36fae23f54a1a3639b9cc Mon Sep 17 00:00:00 2001
From: Victor Julien <victor@inliniac.net>
Date: Wed, 18 Feb 2015 08:50:01 +0100
Subject: [PATCH] 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.
---
 src/flow-hash.c     | 24 ++++++++++++++++--------
 src/flow-hash.h     |  2 ++
 src/util-runmodes.c |  3 +++
 3 files changed, 21 insertions(+), 8 deletions(-)

diff --git a/src/flow-hash.c b/src/flow-hash.c
index 3d609b1977..7a15119926 100644
--- 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 a5635b06ce..4272896b5a 100644
--- 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 8f1c859649..26ef2b76c0 100644
--- 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)
-- 
2.47.2