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

@@ -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 {

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
 

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)