flow timeout: prevent dead locks

The flow timeout mechanism called both from the flow manager at run time
and at shutdown creates pseudo packets. For this it has it's own packet
pool, which can be depleted if the timeout logic is faster than the packet
processing threads. In this case the flow timeout would enter a wait loop.
The problem however, is that this wait loop would happen while keeping a
flow locked. This could lead to a race condition when the packet thread(s)
are waiting for the lock that the flow manager has.

This patch introduces a new packet pool call 'PacketPoolWaitForN', meant
to make sure that the thread's packet pool has at least N available
packets. The flow timeout paths use this to make sure enough packets are
available *before* grabbing the flow lock. If there aren't enough packets
available yet, the wait happens before the lock as well.

This still means the wait can happen while the flow hash row is locked, so
we do make sure some more packets are available when entering that. But
perhaps in the future we need a more precise logic there as well.

diff --git a/src/flow-manager.c b/src/flow-manager.c
index 2d86b81c6a36cb0de92bc94749ea8545f5c30bbb..411dcbc88cee9874e72e595a581d7db6253de096 100644 (file)
--- a/src/flow-manager.c
+++ b/src/flow-manager.c
@@ -285,6 +285,10 @@ static uint32_t FlowManagerHashRowTimeout(Flow *f, struct timeval *ts,
             continue;
         }
 
+        /* before grabbing the flow lock, make sure we have at least
+         * 3 packets in the pool */
+        PacketPoolWaitForN(3);
+
         FLOWLOCK_WRLOCK(f);
 
         Flow *next_flow = f->hprev;
@@ -377,6 +381,10 @@ static uint32_t FlowTimeoutHash(struct timeval *ts, uint32_t try_cnt,
     for (idx = hash_min; idx < hash_max; idx++) {
         FlowBucket *fb = &flow_hash[idx];
 
+        /* before grabbing the row lock, make sure we have at least
+         * 9 packets in the pool */
+        PacketPoolWaitForN(9);
+
         if (FBLOCK_TRYLOCK(fb) != 0)
             continue;
 
@@ -1220,7 +1228,10 @@ static int FlowMgrTest04 (void)
 static int FlowMgrTest05 (void)
 {
     int result = 0;
+    extern intmax_t max_pending_packets;
+    max_pending_packets = 128;
 
+    PacketPoolInit();
     FlowInitConfig(FLOW_QUIET);
     FlowConfig backup;
     memcpy(&backup, &flow_config, sizeof(FlowConfig));
@@ -1258,7 +1269,7 @@ static int FlowMgrTest05 (void)
 
     memcpy(&flow_config, &backup, sizeof(FlowConfig));
     FlowShutdown();
-
+    PacketPoolDestroy();
     return result;
 }
 #endif /* UNITTESTS */

diff --git a/src/flow-timeout.c b/src/flow-timeout.c
index f2eb0899289c1d21fbf545060e3ede41df2c42d3..b18007939e3b9487676cca760115e70b20383f3c 100644 (file)
--- a/src/flow-timeout.c
+++ b/src/flow-timeout.c
@@ -515,6 +515,7 @@ static inline void FlowForceReassemblyForHash(void)
     for (idx = 0; idx < flow_config.hash_size; idx++) {
         FlowBucket *fb = &flow_hash[idx];
 
+        PacketPoolWaitForN(9);
         FBLOCK_LOCK(fb);
 
         /* get the topmost flow from the QUEUE */
@@ -522,6 +523,8 @@ static inline void FlowForceReassemblyForHash(void)
 
         /* we need to loop through all the flows in the queue */
         while (f != NULL) {
+            PacketPoolWaitForN(3);
+
             FLOWLOCK_WRLOCK(f);
 
             /* Get the tcp session for the flow */

diff --git a/src/tmqh-packetpool.c b/src/tmqh-packetpool.c
index 79103a4a01d927a92c3dc3658fe412f82f49ba13..4c4bf40691afa011cf8a7d86d1c143151b08c3dc 100644 (file)
--- a/src/tmqh-packetpool.c
+++ b/src/tmqh-packetpool.c
@@ -157,6 +157,57 @@ void PacketPoolWait(void)
         cc_barrier();
 }
 
+/** \brief Wait until we have the requested ammount of packets in the pool
+ *
+ *  In some cases waiting for packets is undesirable. Especially when
+ *  a wait would happen under a lock of some kind, other parts of the
+ *  engine could have to wait.
+ *
+ *  This function only returns when at least N packets are in our pool.
+ *
+ *  \param n number of packets needed
+ */
+void PacketPoolWaitForN(int n)
+{
+    PktPool *my_pool = GetThreadPacketPool();
+    Packet *p = NULL;
+
+    while (1) {
+        int i = 0;
+        PacketPoolWait();
+
+        /* count packets in our stack */
+        p = my_pool->head;
+        while (p != NULL) {
+            if (++i == n)
+                return;
+
+            p = p->next;
+        }
+
+        /* continue counting in the return stack */
+        if (my_pool->return_stack.head != NULL) {
+            SCMutexLock(&my_pool->return_stack.mutex);
+            p = my_pool->return_stack.head;
+            while (p != NULL) {
+                if (++i == n) {
+                    SCMutexUnlock(&my_pool->return_stack.mutex);
+                    return;
+                }
+                p = p->next;
+            }
+            SCMutexUnlock(&my_pool->return_stack.mutex);
+
+        /* or signal that we need packets and wait */
+        } else {
+            SCMutexLock(&my_pool->return_stack.mutex);
+            SC_ATOMIC_ADD(my_pool->return_stack.sync_now, 1);
+            SCCondWait(&my_pool->return_stack.cond, &my_pool->return_stack.mutex);
+            SCMutexUnlock(&my_pool->return_stack.mutex);
+        }
+    }
+}
+
 /** \brief a initialized packet
  *
  *  \warning Use *only* at init, not at packet runtime

diff --git a/src/tmqh-packetpool.h b/src/tmqh-packetpool.h
index 6da6fc6aa8555bfc4b4d2b21ef9527da7dc0dbb0..12a3f5cabdd48336a5a47fdc5a0caf83a359b22b 100644 (file)
--- a/src/tmqh-packetpool.h
+++ b/src/tmqh-packetpool.h
@@ -69,6 +69,7 @@ void TmqhReleasePacketsToPacketPool(PacketQueue *);
 void TmqhPacketpoolRegister(void);
 Packet *PacketPoolGetPacket(void);
 void PacketPoolWait(void);
+void PacketPoolWaitForN(int n);
 void PacketPoolReturnPacket(Packet *p);
 void PacketPoolInit(void);
 void PacketPoolInitEmpty(void);