flow/recycler: batch returns to spare pool

To reduce locking overhead in the spare pool, batch returns per
100 (spare pool block size).

src/flow-manager.c

@@ -1038,9 +1038,10 @@ static void Recycler(ThreadVars *tv, FlowRecyclerThreadData *ftd, Flow *f)
 
     FlowClearMemory(f, f->protomap);
     FLOWLOCK_UNLOCK(f);
-    FlowSparePoolReturnFlow(f);
 }
 
+extern uint32_t flow_spare_pool_block_size;
+
 /** \brief Thread that manages timed out flows.
  *
  *  \param td ThreadVars casted to void ptr
@@ -1051,6 +1052,7 @@ static TmEcode FlowRecycler(ThreadVars *th_v, void *thread_data)
     BUG_ON(ftd == NULL);
     const bool time_is_live = TimeModeIsLive();
     uint64_t recycled_cnt = 0;
+    FlowQueuePrivate ret_queue = { NULL, NULL, 0 };
 
     TmThreadsSetFlag(th_v, THV_RUNNING);
 
@@ -1077,6 +1079,15 @@ static TmEcode FlowRecycler(ThreadVars *th_v, void *thread_data)
         while ((f = FlowQueuePrivateGetFromTop(&list)) != NULL) {
             Recycler(th_v, ftd, f);
             cnt++;
+
+            /* for every full sized block, add it to the spare pool */
+            FlowQueuePrivateAppendFlow(&ret_queue, f);
+            if (ret_queue.len == flow_spare_pool_block_size) {
+                FlowSparePoolReturnFlows(&ret_queue);
+            }
+        }
+        if (ret_queue.len > 0) {
+            FlowSparePoolReturnFlows(&ret_queue);
         }
         if (cnt > 0) {
             recycled_cnt += cnt;

src/flow-spare-pool.c

@@ -1,4 +1,4 @@
-/* Copyright (C) 2007-2020 Open Information Security Foundation
+/* Copyright (C) 2007-2023 Open Information Security Foundation
  *
  * You can copy, redistribute or modify this Program under the terms of
  * the GNU General Public License version 2 as published by the Free
@@ -40,7 +40,7 @@ typedef struct FlowSparePool {
 } FlowSparePool;
 
 static uint32_t flow_spare_pool_flow_cnt = 0;
-static uint32_t flow_spare_pool_block_size = 100;
+uint32_t flow_spare_pool_block_size = 100;
 static FlowSparePool *flow_spare_pool = NULL;
 static SCMutex flow_spare_pool_m = SCMUTEX_INITIALIZER;
 
@@ -121,7 +121,18 @@ void FlowSparePoolReturnFlow(Flow *f)
 
 void FlowSparePoolReturnFlows(FlowQueuePrivate *fqp)
 {
+    FlowSparePool *p = FlowSpareGetPool();
+    DEBUG_VALIDATE_BUG_ON(p == NULL);
+    p->queue = *fqp;
 
+    SCMutexLock(&flow_spare_pool_m);
+    p->next = flow_spare_pool;
+    flow_spare_pool = p;
+    flow_spare_pool_flow_cnt += fqp->len;
+    SCMutexUnlock(&flow_spare_pool_m);
+
+    FlowQueuePrivate empty = { NULL, NULL, 0 };
+    *fqp = empty;
 }
 
 FlowQueuePrivate FlowSpareGetFromPool(void)