flow: lockless flow manager checks

Until this point, the flow manager would check for timed out flows
by walking the flow hash, locking first the hash row and then each
individual flow to get it's state and timestamp. To not be too
intrusive trylocks were used so that a busy flow wouldn't cause the
flow manager to wait for a long time while holding the hash row lock.

Building on the changes in handling of the flow state and lastts
fields, this patch changes the flow managers behavior.

It can now get a flows state atomically and the lastts can be safely
read while holding just the flow hash row lock. This allows the flow
manager to do the basic time out check much more cheaply:

1. it doesn't have to wait for getting a lock
2. it doesn't interupt the packet path

As a consequence the trylock is now also gone. A flow that returns
'true' on timeout is pretty much certainly not going to be busy so
we can safely lock it unconditionally. This also means the flow
manager now walks the entire row unconditionally and is guaranteed
to inspect each flow in the row.

To make sure the functions called before the flow lock don't
accidentally change the flow (which would require a lock) the args
to these flows are changed to const pointers.

diff --git a/src/flow-manager.c b/src/flow-manager.c
index 72d6ce535f080d05ab1c58b846c57acb487da67d..217681c609bc7372c9a97580b39ce62cf337dd0c 100644 (file)
--- a/src/flow-manager.c
+++ b/src/flow-manager.c
@@ -163,7 +163,7 @@ void FlowKillFlowManagerThread(void)
  *
  *  \retval timeout timeout in seconds
  */
-static inline uint32_t FlowGetFlowTimeout(Flow *f, int state, int emergency)
+static inline uint32_t FlowGetFlowTimeout(const Flow *f, int state, int emergency)
 {
     uint32_t timeout;
 
@@ -208,7 +208,7 @@ static inline uint32_t FlowGetFlowTimeout(Flow *f, int state, int emergency)
  *  \retval 0 not timed out
  *  \retval 1 timed out
  */
-static int FlowManagerFlowTimeout(Flow *f, int state, struct timeval *ts, int emergency)
+static int FlowManagerFlowTimeout(const Flow *f, int state, struct timeval *ts, int emergency)
 {
     /* set the timeout value according to the flow operating mode,
      * flow's state and protocol.*/
@@ -273,22 +273,23 @@ static uint32_t FlowManagerHashRowTimeout(Flow *f, struct timeval *ts,
     uint32_t cnt = 0;
 
     do {
-        if (FLOWLOCK_TRYWRLOCK(f) != 0) {
-            f = f->hprev;
-            continue;
-        }
-
-        Flow *next_flow = f->hprev;
+        /* check flow timeout based on lastts and state. Both can be
+         * accessed w/o Flow lock as we do have the hash row lock (so flow
+         * can't disappear) and flow_state is atomic. lastts can only
+         * be modified when we have both the flow and hash row lock */
 
         int state = SC_ATOMIC_GET(f->flow_state);
 
         /* timeout logic goes here */
         if (FlowManagerFlowTimeout(f, state, ts, emergency) == 0) {
-            FLOWLOCK_UNLOCK(f);
             f = f->hprev;
             continue;
         }
 
+        FLOWLOCK_WRLOCK(f);
+
+        Flow *next_flow = f->hprev;
+
         /* check if the flow is fully timed out and
          * ready to be discarded. */
         if (FlowManagerFlowTimedOut(f, ts) == 1) {