packet pool: attach to ThreadVars for multi-instance support

jasonish · claude · jasonish · commit 31bbe6501e3c · 2025-10-29T08:33:46.000-06:00
Move packet pools from thread-local storage to ThreadVars structure to enable multi-instance Suricata support. Thread-local storage is process-wide and cannot be isolated per instance. This implementation uses a simplified approach where packets are returned directly to their origin pool (via p->pool) with mutex locking. The previous batching optimizations that used thread-local pending pools have been removed in favor of code simplicity. Changes: - Add pkt_pool pointer to ThreadVars - Add tv back-pointer to PktPool - Update PacketPoolGetPacket/PacketPoolWait to accept ThreadVars parameter - Update PacketGetFromQueueOrAlloc to accept ThreadVars parameter - Update PacketDefragPktSetup to accept ThreadVars parameter - Update FlowPseudoPacketGet to accept ThreadVars parameter - Add PacketPoolInitThreaded/PacketPoolDestroyThreaded functions - Remove thread-local current_tv and pending pool batching logic - Update all call sites across source files 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
diff --git a/examples/plugins/ci-capture/source.c b/examples/plugins/ci-capture/source.c
@@ -65,7 +65,7 @@ static TmEcode ReceiveLoop(ThreadVars *tv, void *data, void *slot)
     /* Notify we are running and processing packets. */
     TmThreadsSetFlag(tv, THV_RUNNING);
 
-    PacketPoolWait();
+    PacketPoolWait(ptv->tv);
     Packet *p = PacketGetFromQueueOrAlloc();
     if (unlikely(p == NULL)) {
         return TM_ECODE_FAILED;
diff --git a/plugins/napatech/source-napatech.c b/plugins/napatech/source-napatech.c
@@ -847,7 +847,7 @@ TmEcode NapatechPacketLoop(ThreadVars *tv, void *data, void *slot)
     while (!(suricata_ctl_flags & SURICATA_STOP)) {
         /* make sure we have at least one packet in the packet pool, to prevent
          * us from alloc'ing packets at line rate */
-        PacketPoolWait();
+        PacketPoolWait(ptv->tv);
 
         /* Napatech returns packets 1 at a time */
         status = NT_NetRxGet(ntv->rx_stream, &packet_buffer, 1000);
diff --git a/plugins/pfring/source-pfring.c b/plugins/pfring/source-pfring.c
@@ -330,7 +330,7 @@ TmEcode ReceivePfringLoop(ThreadVars *tv, void *data, void *slot)
 
         /* make sure we have at least one packet in the packet pool, to prevent
          * us from alloc'ing packets at line rate */
-        PacketPoolWait();
+        PacketPoolWait(ptv->tv);
 
         p = PacketGetFromQueueOrAlloc();
         if (p == NULL) {
diff --git a/src/decode.c b/src/decode.c
@@ -290,10 +290,10 @@ void PacketFreeOrRelease(Packet *p)
  *
  *  \retval p packet, NULL on error
  */
-Packet *PacketGetFromQueueOrAlloc(void)
+Packet *PacketGetFromQueueOrAlloc(ThreadVars *tv)
 {
     /* try the pool first */
-    Packet *p = PacketPoolGetPacket();
+    Packet *p = PacketPoolGetPacket(tv);
 
     if (p == NULL) {
         /* non fatal, we're just not processing a packet then */
@@ -403,7 +403,7 @@ Packet *PacketTunnelPktSetup(ThreadVars *tv, DecodeThreadVars *dtv, Packet *pare
     }
 
     /* get us a packet */
-    Packet *p = PacketGetFromQueueOrAlloc();
+    Packet *p = PacketGetFromQueueOrAlloc(tv);
     if (unlikely(p == NULL)) {
         SCReturnPtr(NULL, "Packet");
     }
@@ -470,12 +470,12 @@ Packet *PacketTunnelPktSetup(ThreadVars *tv, DecodeThreadVars *dtv, Packet *pare
  *
  *  \retval p the pseudo packet or NULL if out of memory
  */
-Packet *PacketDefragPktSetup(Packet *parent, const uint8_t *pkt, uint32_t len, uint8_t proto)
+Packet *PacketDefragPktSetup(ThreadVars *tv, Packet *parent, const uint8_t *pkt, uint32_t len, uint8_t proto)
 {
     SCEnter();
 
     /* get us a packet */
-    Packet *p = PacketGetFromQueueOrAlloc();
+    Packet *p = PacketGetFromQueueOrAlloc(tv);
     if (unlikely(p == NULL)) {
         SCReturnPtr(NULL, "Packet");
     }
diff --git a/src/decode.h b/src/decode.h
@@ -1115,10 +1115,10 @@ enum DecodeTunnelProto {
 
 Packet *PacketTunnelPktSetup(ThreadVars *tv, DecodeThreadVars *dtv, Packet *parent,
                              const uint8_t *pkt, uint32_t len, enum DecodeTunnelProto proto);
-Packet *PacketDefragPktSetup(Packet *parent, const uint8_t *pkt, uint32_t len, uint8_t proto);
+Packet *PacketDefragPktSetup(ThreadVars *tv, Packet *parent, const uint8_t *pkt, uint32_t len, uint8_t proto);
 void PacketDefragPktSetupParent(Packet *parent);
 void DecodeRegisterPerfCounters(DecodeThreadVars *, ThreadVars *);
-Packet *PacketGetFromQueueOrAlloc(void);
+Packet *PacketGetFromQueueOrAlloc(ThreadVars *);
 Packet *PacketGetFromAlloc(void);
 void PacketDecodeFinalize(ThreadVars *tv, DecodeThreadVars *dtv, Packet *p);
 void PacketUpdateEngineEventCounters(ThreadVars *tv,
diff --git a/src/defrag.c b/src/defrag.c
@@ -263,7 +263,7 @@ Defrag4Reassemble(ThreadVars *tv, DefragTracker *tracker, Packet *p)
 
     /* Allocate a Packet for the reassembled packet.  On failure we
      * SCFree all the resources held by this tracker. */
-    rp = PacketDefragPktSetup(p, NULL, 0, IPV4_GET_RAW_IPPROTO(oip4h));
+    rp = PacketDefragPktSetup(tv, p, NULL, 0, IPV4_GET_RAW_IPPROTO(oip4h));
     if (rp == NULL) {
         goto error_remove_tracker;
     }
@@ -424,7 +424,7 @@ Defrag6Reassemble(ThreadVars *tv, DefragTracker *tracker, Packet *p)
     /* Allocate a Packet for the reassembled packet.  On failure we
      * SCFree all the resources held by this tracker. */
     rp = PacketDefragPktSetup(
-            p, (const uint8_t *)oip6h, IPV6_GET_RAW_PLEN(oip6h) + sizeof(IPV6Hdr), 0);
+            tv, p, (const uint8_t *)oip6h, IPV6_GET_RAW_PLEN(oip6h) + sizeof(IPV6Hdr), 0);
     if (rp == NULL) {
         goto error_remove_tracker;
     }
diff --git a/src/flow-timeout.c b/src/flow-timeout.c
@@ -261,10 +261,10 @@ static inline Packet *FlowPseudoPacketSetup(
     return NULL;
 }
 
-Packet *FlowPseudoPacketGet(int direction, Flow *f, const TcpSession *ssn)
+Packet *FlowPseudoPacketGet(ThreadVars *tv, int direction, Flow *f, const TcpSession *ssn)
 {
-    PacketPoolWait();
-    Packet *p = PacketPoolGetPacket();
+    PacketPoolWait(tv);
+    Packet *p = PacketPoolGetPacket(tv);
     if (p == NULL) {
         return NULL;
     }
diff --git a/src/flow-timeout.h b/src/flow-timeout.h
@@ -29,6 +29,6 @@
 void FlowSendToLocalThread(Flow *f);
 bool FlowNeedsReassembly(Flow *f);
 void FlowWorkToDoCleanup(void);
-Packet *FlowPseudoPacketGet(int direction, Flow *f, const TcpSession *ssn);
+Packet *FlowPseudoPacketGet(ThreadVars *tv, int direction, Flow *f, const TcpSession *ssn);
 
 #endif /* SURICATA_FLOW_TIMEOUT_H */
diff --git a/src/flow-worker.c b/src/flow-worker.c
@@ -122,7 +122,7 @@ static int FlowFinish(ThreadVars *tv, Flow *f, FlowWorkerThreadData *fw, void *d
 
     /* insert a pseudo packet in the toserver direction */
     if (client == STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_ONLY_DETECTION) {
-        Packet *p = FlowPseudoPacketGet(0, f, ssn);
+        Packet *p = FlowPseudoPacketGet(tv, 0, f, ssn);
         if (p != NULL) {
             PKT_SET_SRC(p, PKT_SRC_FFR);
             if (server == STREAM_HAS_UNPROCESSED_SEGMENTS_NONE) {
@@ -136,7 +136,7 @@ static int FlowFinish(ThreadVars *tv, Flow *f, FlowWorkerThreadData *fw, void *d
 
     /* handle toclient */
     if (server == STREAM_HAS_UNPROCESSED_SEGMENTS_NEED_ONLY_DETECTION) {
-        Packet *p = FlowPseudoPacketGet(1, f, ssn);
+        Packet *p = FlowPseudoPacketGet(tv, 1, f, ssn);
         if (p != NULL) {
             PKT_SET_SRC(p, PKT_SRC_FFR);
             p->flowflags |= FLOW_PKT_LAST_PSEUDO;
diff --git a/src/source-af-packet.c b/src/source-af-packet.c
@@ -926,7 +926,7 @@ static int AFPReadFromRing(AFPThreadVars *ptv)
         if (unlikely(AFPShouldIgnoreFrame(ptv, sll)))
             goto next_frame;
 
-        Packet *p = PacketGetFromQueueOrAlloc();
+        Packet *p = PacketGetFromQueueOrAlloc(ptv->tv);
         if (p == NULL) {
             return AFPSuriFailure(ptv, h);
         }
@@ -956,7 +956,7 @@ static inline void AFPFlushBlock(struct tpacket_block_desc *pbd)
 
 static inline int AFPParsePacketV3(AFPThreadVars *ptv, struct tpacket_block_desc *pbd, struct tpacket3_hdr *ppd)
 {
-    Packet *p = PacketGetFromQueueOrAlloc();
+    Packet *p = PacketGetFromQueueOrAlloc(ptv->tv);
     if (p == NULL) {
         SCReturnInt(AFP_SURI_FAILURE);
     }
@@ -1387,7 +1387,7 @@ TmEcode ReceiveAFPLoop(ThreadVars *tv, void *data, void *slot)
 
         /* make sure we have at least one packet in the packet pool, to prevent
          * us from alloc'ing packets at line rate */
-        PacketPoolWait();
+        PacketPoolWait(ptv->tv);
 
         StatsIncr(ptv->tv, ptv->capture_afp_poll);
 
diff --git a/src/source-af-xdp.c b/src/source-af-xdp.c
@@ -708,7 +708,7 @@ static TmEcode ReceiveAFXDPLoop(ThreadVars *tv, void *data, void *slot)
     // packets)
     TmThreadsSetFlag(tv, THV_RUNNING);
 
-    PacketPoolWait();
+    PacketPoolWait(ptv->tv);
     while (1) {
         /* Start by checking the state of our interface */
         if (unlikely(ptv->afxdp_state == AFXDP_STATE_DOWN)) {
@@ -777,7 +777,7 @@ static TmEcode ReceiveAFXDPLoop(ThreadVars *tv, void *data, void *slot)
         gettimeofday(&ts, NULL);
         ptv->pkts += rcvd;
         for (uint32_t i = 0; i < rcvd; i++) {
-            p = PacketGetFromQueueOrAlloc();
+            p = PacketGetFromQueueOrAlloc(ptv->tv);
             if (unlikely(p == NULL)) {
                 StatsIncr(ptv->tv, ptv->capture_afxdp_acquire_pkt_failed);
                 continue;
diff --git a/src/source-dpdk.c b/src/source-dpdk.c
@@ -353,7 +353,7 @@ static TmEcode ReceiveDPDKLoopInit(ThreadVars *tv, DPDKThreadVars *ptv)
     // Indicate that the thread is actually running its application level
     // code (i.e., it can poll packets)
     TmThreadsSetFlag(tv, THV_RUNNING);
-    PacketPoolWait();
+    PacketPoolWait(tv);
 
     rte_eth_stats_reset(ptv->port_id);
     rte_eth_xstats_reset(ptv->port_id);
diff --git a/src/source-erf-dag.c b/src/source-erf-dag.c
@@ -396,7 +396,7 @@ ProcessErfDagRecords(ErfDagThreadVars *ewtn, uint8_t *top, uint32_t *pkts_read)
 
         /* Make sure we have at least one packet in the packet pool,
          * to prevent us from alloc'ing packets at line rate. */
-        PacketPoolWait();
+        PacketPoolWait(ptv->tv);
 
         prec = (char *)ewtn->btm;
         dr = (dag_record_t*)prec;
diff --git a/src/source-erf-file.c b/src/source-erf-file.c
@@ -127,9 +127,9 @@ TmEcode ReceiveErfFileLoop(ThreadVars *tv, void *data, void *slot)
 
         /* Make sure we have at least one packet in the packet pool,
          * to prevent us from alloc'ing packets at line rate. */
-        PacketPoolWait();
+        PacketPoolWait(etv->tv);
 
-        p = PacketGetFromQueueOrAlloc();
+        p = PacketGetFromQueueOrAlloc(etv->tv);
         if (unlikely(p == NULL)) {
             SCLogError("Failed to allocate a packet.");
             EngineStop();
diff --git a/src/source-ipfw.c b/src/source-ipfw.c
@@ -281,7 +281,7 @@ TmEcode ReceiveIPFWLoop(ThreadVars *tv, void *data, void *slot)
 
         /* make sure we have at least one packet in the packet pool, to prevent
          * us from alloc'ing packets at line rate */
-        PacketPoolWait();
+        PacketPoolWait(ptv->tv);
 
         p = PacketGetFromQueueOrAlloc();
         if (p == NULL) {
diff --git a/src/source-netmap.c b/src/source-netmap.c
@@ -812,7 +812,7 @@ static TmEcode ReceiveNetmapLoop(ThreadVars *tv, void *data, void *slot)
 
         /* make sure we have at least one packet in the packet pool,
          * to prevent us from alloc'ing packets at line rate */
-        PacketPoolWait();
+        PacketPoolWait(ntv->tv);
 
         int r = poll(&fds, 1, POLL_TIMEOUT);
         if (r < 0) {
diff --git a/src/source-nfq.c b/src/source-nfq.c
@@ -551,7 +551,7 @@ static int NFQCallBack(struct nfq_q_handle *qh, struct nfgenmsg *nfmsg,
     int ret;
 
     /* grab a packet */
-    Packet *p = PacketGetFromQueueOrAlloc();
+    Packet *p = PacketGetFromQueueOrAlloc(tv);
     if (p == NULL) {
         return -1;
     }
diff --git a/src/source-pcap-file-helper.c b/src/source-pcap-file-helper.c
@@ -69,7 +69,7 @@ void PcapFileCallbackLoop(char *user, struct pcap_pkthdr *h, u_char *pkt)
     }
 #endif
     PcapFileFileVars *ptv = (PcapFileFileVars *)user;
-    Packet *p = PacketGetFromQueueOrAlloc();
+    Packet *p = PacketGetFromQueueOrAlloc(ptv->shared->tv);
 
     if (unlikely(p == NULL)) {
         SCReturn;
@@ -147,7 +147,7 @@ TmEcode PcapFileDispatch(PcapFileFileVars *ptv)
 
         /* make sure we have at least one packet in the packet pool, to prevent
          * us from alloc'ing packets at line rate */
-        PacketPoolWait();
+        PacketPoolWait(ptv->shared->tv);
 
         /* Right now we just support reading packets one at a time. */
         int r = pcap_dispatch(ptv->pcap_handle, packet_q_len,
diff --git a/src/source-pcap.c b/src/source-pcap.c
@@ -328,7 +328,7 @@ static void PcapCallbackLoop(char *user, struct pcap_pkthdr *h, u_char *pkt)
     SCEnter();
 
     PcapThreadVars *ptv = (PcapThreadVars *)user;
-    Packet *p = PacketGetFromQueueOrAlloc();
+    Packet *p = PacketGetFromQueueOrAlloc(ptv->tv);
 
     if (unlikely(p == NULL)) {
         SCReturn;
@@ -409,7 +409,7 @@ static TmEcode ReceivePcapLoop(ThreadVars *tv, void *data, void *slot)
 
         /* make sure we have at least one packet in the packet pool, to prevent
          * us from alloc'ing packets at line rate */
-        PacketPoolWait();
+        PacketPoolWait(ptv->tv);
 
         int r = pcap_dispatch(ptv->pcap_handle, packet_q_len,
                           (pcap_handler)PcapCallbackLoop, (u_char *)ptv);
diff --git a/src/source-windivert.c b/src/source-windivert.c
@@ -437,7 +437,7 @@ static TmEcode WinDivertRecvHelper(ThreadVars *tv, WinDivertThreadVars *wd_tv)
     /* make sure we have at least one packet in the packet pool, to prevent us
      * from alloc'ing packets at line rate
      */
-    PacketPoolWait();
+    PacketPoolWait(ptv->tv);
 
     /* obtain a packet buffer */
     Packet *p = PacketGetFromQueueOrAlloc();
diff --git a/src/stream-tcp.c b/src/stream-tcp.c
@@ -6855,7 +6855,7 @@ static void StreamTcpPseudoPacketCreateDetectLogFlush(ThreadVars *tv,
         SCReturn;
     }
 
-    Packet *np = PacketPoolGetPacket();
+    Packet *np = PacketPoolGetPacket(tv);
     if (np == NULL) {
         SCReturn;
     }
diff --git a/src/threadvars.h b/src/threadvars.h
@@ -139,6 +139,9 @@ typedef struct ThreadVars_ {
     /** Interface-specific thread affinity */
     char *iface_name;
 
+    /** Packet pool for this thread */
+    struct PktPool_ *pkt_pool;
+
     Storage storage[];
 } ThreadVars;
 
diff --git a/src/tm-threads.c b/src/tm-threads.c
@@ -493,7 +493,7 @@ static void *TmThreadsSlotVar(void *td)
         /* if we didn't get a packet see if we need to do some housekeeping */
         if (unlikely(p == NULL)) {
             if (tv->flow_queue && SC_ATOMIC_GET(tv->flow_queue->non_empty)) {
-                p = PacketGetFromQueueOrAlloc();
+                p = PacketGetFromQueueOrAlloc(tv);
                 if (p != NULL) {
                     p->flags |= PKT_PSEUDO_STREAM_END;
                     PKT_SET_SRC(p, PKT_SRC_CAPTURE_TIMEOUT);
diff --git a/src/tm-threads.h b/src/tm-threads.h
@@ -221,7 +221,7 @@ static inline void TmThreadsCaptureInjectPacket(ThreadVars *tv, Packet *p)
 {
     TmThreadsUnsetFlag(tv, THV_CAPTURE_INJECT_PKT);
     if (p == NULL)
-        p = PacketGetFromQueueOrAlloc();
+        p = PacketGetFromQueueOrAlloc(tv);
     if (p != NULL) {
         p->flags |= PKT_PSEUDO_STREAM_END;
         PKT_SET_SRC(p, PKT_SRC_CAPTURE_TIMEOUT);
diff --git a/src/tmqh-packetpool.c b/src/tmqh-packetpool.c
diff --git a/src/tmqh-packetpool.h b/src/tmqh-packetpool.h

Original file line number	Diff line number	Diff line change
`@@ -290,10 +290,10 @@ void PacketFreeOrRelease(Packet *p)`
`290`	`290`	`*`
`291`	`291`	`* \retval p packet, NULL on error`
`292`	`292`	`*/`
`293`		`-Packet *PacketGetFromQueueOrAlloc(void)`
	`293`	`+Packet PacketGetFromQueueOrAlloc(ThreadVars tv)`
`294`	`294`	`{`
`295`	`295`	`/* try the pool first */`
`296`		`- Packet *p = PacketPoolGetPacket();`
	`296`	`+ Packet *p = PacketPoolGetPacket(tv);`
`297`	`297`
`298`	`298`	`if (p == NULL) {`
`299`	`299`	`/* non fatal, we're just not processing a packet then */`
`@@ -403,7 +403,7 @@ Packet PacketTunnelPktSetup(ThreadVars tv, DecodeThreadVars dtv, Packet pare`
`403`	`403`	`}`
`404`	`404`
`405`	`405`	`/* get us a packet */`
`406`		`- Packet *p = PacketGetFromQueueOrAlloc();`
	`406`	`+ Packet *p = PacketGetFromQueueOrAlloc(tv);`
`407`	`407`	`if (unlikely(p == NULL)) {`
`408`	`408`	`SCReturnPtr(NULL, "Packet");`
`409`	`409`	`}`
`@@ -470,12 +470,12 @@ Packet PacketTunnelPktSetup(ThreadVars tv, DecodeThreadVars dtv, Packet pare`
`470`	`470`	`*`
`471`	`471`	`* \retval p the pseudo packet or NULL if out of memory`
`472`	`472`	`*/`
`473`		`-Packet PacketDefragPktSetup(Packet parent, const uint8_t *pkt, uint32_t len, uint8_t proto)`
	`473`	`+Packet PacketDefragPktSetup(ThreadVars tv, Packet parent, const uint8_t pkt, uint32_t len, uint8_t proto)`
`474`	`474`	`{`
`475`	`475`	`SCEnter();`
`476`	`476`
`477`	`477`	`/* get us a packet */`
`478`		`- Packet *p = PacketGetFromQueueOrAlloc();`
	`478`	`+ Packet *p = PacketGetFromQueueOrAlloc(tv);`
`479`	`479`	`if (unlikely(p == NULL)) {`
`480`	`480`	`SCReturnPtr(NULL, "Packet");`
`481`	`481`	`}`
Original file line number	Diff line number	Diff line change
`@@ -261,10 +261,10 @@ static inline Packet *FlowPseudoPacketSetup(`
`261`	`261`	`return NULL;`
`262`	`262`	`}`
`263`	`263`
`264`		`-Packet FlowPseudoPacketGet(int direction, Flow f, const TcpSession *ssn)`
	`264`	`+Packet FlowPseudoPacketGet(ThreadVars tv, int direction, Flow f, const TcpSession ssn)`
`265`	`265`	`{`
`266`		`- PacketPoolWait();`
`267`		`- Packet *p = PacketPoolGetPacket();`
	`266`	`+ PacketPoolWait(tv);`
	`267`	`+ Packet *p = PacketPoolGetPacket(tv);`
`268`	`268`	`if (p == NULL) {`
`269`	`269`	`return NULL;`
`270`	`270`	`}`