23338889526

Committed 20 Mar 2026 10:29AM UTC coverage: 76.331% (-3.0%) from 79.315%

Build # 23338889526

Build Type

Pull #15053

github

Committed by

web-flow

Commit Message

Merge 00ac1dd14 into 6587e363a

Pull Request Pull Request #15053: Flow queue/v3

Coverage Stats

106 of 127 new or added lines in 8 files covered. (83.46%)

9913 existing lines in 468 files now uncovered.

255689 of 334972 relevant lines covered (76.33%)

4170649.82 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.51

/src/tmqh-packetpool.c

/* Copyright (C) 2007-2022 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
 * Software Foundation.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * version 2 along with this program; if not, write to the Free Software
 * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
 * 02110-1301, USA.
 */

/**
 * \file
 *
 * \author Victor Julien <victor@inliniac.net>
 *
 * Packetpool queue handlers. Packet pool is implemented as a stack.
 */

#include "suricata-common.h"
#include "tmqh-packetpool.h"
#include "tm-queuehandlers.h"
#include "tm-threads.h"
#include "threads.h"
#include "decode.h"
#include "tm-modules.h"
#include "packet.h"
#include "util-profiling.h"
#include "util-validate.h"
#include "action-globals.h"

extern uint32_t max_pending_packets;

/* Number of freed packet to save for one pool before freeing them. */
#define MAX_PENDING_RETURN_PACKETS 32
static uint32_t max_pending_return_packets = MAX_PENDING_RETURN_PACKETS;

thread_local PktPool thread_pkt_pool;

static inline PktPool *GetThreadPacketPool(void)
{
    return &thread_pkt_pool;
}

static PacketQueueNoLock TmqhInputPacketpool(ThreadVars *);

/**
 * \brief TmqhPacketpoolRegister
 * \initonly
 */
void TmqhPacketpoolRegister (void)
{
    tmqh_table[TMQH_PACKETPOOL].name = "packetpool";
    tmqh_table[TMQH_PACKETPOOL].InHandler = TmqhInputPacketpool;
    tmqh_table[TMQH_PACKETPOOL].OutHandler = TmqhOutputPacketpool;
}

static void UpdateReturnThreshold(PktPool *pool)
{
    const float perc = (float)pool->cnt / (float)max_pending_packets;
    uint32_t threshold = (uint32_t)(perc * (float)max_pending_return_packets);
    if (threshold != SC_ATOMIC_GET(pool->return_stack.return_threshold)) {
        SC_ATOMIC_SET(pool->return_stack.return_threshold, threshold);
    }
}

void PacketPoolWait(void)
{
    PktPool *my_pool = GetThreadPacketPool();

    if (my_pool->head == NULL) {
        SC_ATOMIC_SET(my_pool->return_stack.return_threshold, 1);

        SCMutexLock(&my_pool->return_stack.mutex);
        int rc = 0;
        while (my_pool->return_stack.cnt == 0 && rc == 0) {
            rc = SCCondWait(&my_pool->return_stack.cond, &my_pool->return_stack.mutex);
        }
        SCMutexUnlock(&my_pool->return_stack.mutex);

        UpdateReturnThreshold(my_pool);
    }
}

/** \brief a initialized packet
 *
 *  \warning Use *only* at init, not at packet runtime
 */
static void PacketPoolStorePacket(Packet *p)
{
    p->pool = GetThreadPacketPool();
    p->ReleasePacket = PacketPoolReturnPacket;
    PacketPoolReturnPacket(p);
}

static void PacketPoolGetReturnedPackets(PktPool *pool)
{
    SCMutexLock(&pool->return_stack.mutex);
    /* Move all the packets from the locked return stack to the local stack. */
    pool->head = pool->return_stack.head;
    pool->return_stack.head = NULL;
    pool->cnt += pool->return_stack.cnt;
    pool->return_stack.cnt = 0;
    SCMutexUnlock(&pool->return_stack.mutex);
}

/** \brief Get a new packet from the packet pool
 *
 * Only allocates from the thread's local stack, or mallocs new packets.
 * If the local stack is empty, first move all the return stack packets to
 * the local stack.
 *  \retval Packet pointer, or NULL on failure.
 */
Packet *PacketPoolGetPacket(void)
{
    PktPool *pool = GetThreadPacketPool();
    DEBUG_VALIDATE_BUG_ON(pool->initialized == 0);
    DEBUG_VALIDATE_BUG_ON(pool->destroyed == 1);
    if (pool->head) {
        /* Stack is not empty. */
        Packet *p = pool->head;
        pool->head = p->next;
        pool->cnt--;
        p->pool = pool;
        PacketReinit(p);

        UpdateReturnThreshold(pool);
        SCLogDebug("pp: %0.2f cnt:%u max:%d threshold:%u",
                ((float)pool->cnt / (float)max_pending_packets) * (float)100, pool->cnt,
                max_pending_packets, SC_ATOMIC_GET(pool->return_stack.return_threshold));
        return p;
    }

    /* Local Stack is empty, so check the return stack, which requires
     * locking. */
    PacketPoolGetReturnedPackets(pool);

    /* Try to allocate again. Need to check for not empty again, since the
     * return stack might have been empty too.
     */
    if (pool->head) {
        /* Stack is not empty. */
        Packet *p = pool->head;
        pool->head = p->next;
        pool->cnt--;
        p->pool = pool;
        PacketReinit(p);

        UpdateReturnThreshold(pool);
        SCLogDebug("pp: %0.2f cnt:%u max:%d threshold:%u",
                ((float)pool->cnt / (float)max_pending_packets) * (float)100, pool->cnt,
                max_pending_packets, SC_ATOMIC_GET(pool->return_stack.return_threshold));
        return p;
    }

    /* Failed to allocate a packet, so return NULL. */
    /* Optionally, could allocate a new packet here. */
    return NULL;
}

/** \brief Return packet to Packet pool
 *
 */
void PacketPoolReturnPacket(Packet *p)
{
    PktPool *my_pool = GetThreadPacketPool();
    PktPool *pool = p->pool;
    if (pool == NULL) {
        PacketFree(p);
        return;
    }

    PacketReleaseRefs(p);

#ifdef DEBUG_VALIDATION
    BUG_ON(pool->initialized == 0);
    BUG_ON(pool->destroyed == 1);
    BUG_ON(my_pool->initialized == 0);
    BUG_ON(my_pool->destroyed == 1);
#endif /* DEBUG_VALIDATION */

    if (pool == my_pool) {
        /* Push back onto this thread's own stack, so no locking. */
        p->next = my_pool->head;
        my_pool->head = p;
        my_pool->cnt++;
    } else {
        PktPool *pending_pool = my_pool->pending_pool;
        if (pending_pool == NULL || pending_pool == pool) {
            if (pending_pool == NULL) {
                /* No pending packet, so store the current packet. */
                p->next = NULL;
                my_pool->pending_pool = pool;
                my_pool->pending_head = p;
                my_pool->pending_tail = p;
                my_pool->pending_count = 1;
            } else if (pending_pool == pool) {
                /* Another packet for the pending pool list. */
                p->next = my_pool->pending_head;
                my_pool->pending_head = p;
                my_pool->pending_count++;
            }

            const uint32_t threshold = SC_ATOMIC_GET(pool->return_stack.return_threshold);
            if (my_pool->pending_count >= threshold) {
                /* Return the entire list of pending packets. */
                SCMutexLock(&pool->return_stack.mutex);
                my_pool->pending_tail->next = pool->return_stack.head;
                pool->return_stack.head = my_pool->pending_head;
                pool->return_stack.cnt += my_pool->pending_count;
                SCCondSignal(&pool->return_stack.cond);
                SCMutexUnlock(&pool->return_stack.mutex);
                /* Clear the list of pending packets to return. */
                my_pool->pending_pool = NULL;
                my_pool->pending_head = NULL;
                my_pool->pending_tail = NULL;
                my_pool->pending_count = 0;
            }
        } else {
            /* Push onto return stack for this pool */
            SCMutexLock(&pool->return_stack.mutex);
            p->next = pool->return_stack.head;
            pool->return_stack.head = p;
            pool->return_stack.cnt++;
            SCCondSignal(&pool->return_stack.cond);
            SCMutexUnlock(&pool->return_stack.mutex);
        }
    }
}

void PacketPoolInit(void)
{
    PktPool *my_pool = GetThreadPacketPool();

#ifdef DEBUG_VALIDATION
    BUG_ON(my_pool->initialized);
    my_pool->initialized = 1;
    my_pool->destroyed = 0;
#endif /* DEBUG_VALIDATION */

    SCMutexInit(&my_pool->return_stack.mutex, NULL);
    SCCondInit(&my_pool->return_stack.cond, NULL);
    SC_ATOMIC_INIT(my_pool->return_stack.return_threshold);
    SC_ATOMIC_SET(my_pool->return_stack.return_threshold, 32);

    /* pre allocate packets */
    SCLogDebug("preallocating packets... packet size %" PRIuMAX "",
               (uintmax_t)SIZE_OF_PACKET);
    for (uint32_t i = 0; i < max_pending_packets; i++) {
        Packet *p = PacketGetFromAlloc();
        if (unlikely(p == NULL)) {
            FatalError("Fatal error encountered while allocating a packet. Exiting...");
        }
        PacketPoolStorePacket(p);
    }

    //SCLogInfo("preallocated %"PRIiMAX" packets. Total memory %"PRIuMAX"",
    //        max_pending_packets, (uintmax_t)(max_pending_packets*SIZE_OF_PACKET));
}

void PacketPoolDestroy(void)
{
    Packet *p = NULL;
    PktPool *my_pool = GetThreadPacketPool();

#ifdef DEBUG_VALIDATION
    BUG_ON(my_pool && my_pool->destroyed);
#endif /* DEBUG_VALIDATION */

    if (my_pool && my_pool->pending_pool != NULL) {
        p = my_pool->pending_head;
        while (p) {
            Packet *next_p = p->next;
            PacketFree(p);
            p = next_p;
            my_pool->pending_count--;
        }
#ifdef DEBUG_VALIDATION
        BUG_ON(my_pool->pending_count);
#endif /* DEBUG_VALIDATION */
        my_pool->pending_pool = NULL;
        my_pool->pending_head = NULL;
        my_pool->pending_tail = NULL;
    }

    while ((p = PacketPoolGetPacket()) != NULL) {
        PacketFree(p);
    }

#ifdef DEBUG_VALIDATION
    my_pool->initialized = 0;
    my_pool->destroyed = 1;
#endif /* DEBUG_VALIDATION */
}

static PacketQueueNoLock TmqhInputPacketpool(ThreadVars *tv)
{
    Packet *p = PacketPoolGetPacket();
    PacketQueueNoLock pq = { .top = p, .bot = p, .len = (p != NULL) };
    return pq;
}

void TmqhOutputPacketpool(ThreadVars *t, Packet *p)
{
    bool proot = false;

    SCEnter();
    SCLogDebug("Packet %p, p->root %p, alloced %s", p, p->root, BOOL2STR(p->pool == NULL));

    if (PacketIsTunnel(p)) {
        SCLogDebug("Packet %p is a tunnel packet: %s",
            p,p->root ? "upper layer" : "tunnel root");

        /* get a lock to access root packet fields */
        SCSpinlock *lock = p->root ? &p->root->persistent.tunnel_lock : &p->persistent.tunnel_lock;
        SCSpinLock(lock);

        if (PacketIsTunnelRoot(p)) {
            SCLogDebug("IS_TUNNEL_ROOT_PKT == TRUE");
            CaptureStatsUpdate(t, p); // TODO move out of lock

            const uint16_t outstanding = TUNNEL_PKT_TPR(p) - TUNNEL_PKT_RTV(p);
            SCLogDebug("root pkt: outstanding %u", outstanding);
            if (outstanding == 0) {
                SCLogDebug("no tunnel packets outstanding, no more tunnel "
                        "packet(s) depending on this root");
                /* if this packet is the root and there are no
                 * more tunnel packets to consider
                 *
                 * return it to the pool */
            } else {
                SCLogDebug("tunnel root Packet %p: outstanding > 0, so "
                        "packets are still depending on this root, setting "
                        "SET_TUNNEL_PKT_VERDICTED", p);
                /* if this is the root and there are more tunnel
                 * packets, return this to the pool. It's still referenced
                 * by the tunnel packets, and we will return it
                 * when we handle them */
                PacketTunnelSetVerdicted(p);

                PACKET_PROFILING_END(p);
                SCSpinUnlock(lock);
                SCReturn;
            }
        } else {
            SCLogDebug("NOT IS_TUNNEL_ROOT_PKT, so tunnel pkt");

            TUNNEL_INCR_PKT_RTV_NOLOCK(p);
            const uint16_t outstanding = TUNNEL_PKT_TPR(p) - TUNNEL_PKT_RTV(p);
            SCLogDebug("tunnel pkt: outstanding %u", outstanding);
            /* all tunnel packets are processed except us. Root already
             * processed. So return tunnel pkt and root packet to the
             * pool. */
            if (outstanding == 0 && p->root && PacketTunnelIsVerdicted(p->root)) {
                SCLogDebug("root verdicted == true && no outstanding");

                /* handle freeing the root as well*/
                SCLogDebug("setting proot = 1 for root pkt, p->root %p "
                        "(tunnel packet %p)", p->root, p);
                proot = true;

                /* fall through */

            } else {
                /* root not ready yet, or not the last tunnel packet,
                 * so get rid of the tunnel pkt only */

                SCLogDebug("NOT IS_TUNNEL_PKT_VERDICTED (%s) || "
                           "outstanding > 0 (%u)",
                        (p->root && PacketTunnelIsVerdicted(p->root)) ? "true" : "false",
                        outstanding);

                /* fall through */
            }
        }
        SCSpinUnlock(lock);

        SCLogDebug("tunnel stuff done, move on (proot %d)", proot);

    } else {
        CaptureStatsUpdate(t, p);
    }

    SCLogDebug("[packet %p][%s] %s", p,
            PacketIsTunnel(p) ? PacketIsTunnelRoot(p) ? "tunnel::root" : "tunnel::leaf"
                              : "no tunnel",
            (p->action & ACTION_DROP) ? "DROP" : "no drop");

    /* we're done with the tunnel root now as well */
    if (proot) {
        SCLogDebug("getting rid of root pkt... alloc'd %s", BOOL2STR(p->root->pool == NULL));

        PacketReleaseRefs(p->root);
        p->root->ReleasePacket(p->root);
        p->root = NULL;
    }

    PACKET_PROFILING_END(p);

    PacketReleaseRefs(p);
    p->ReleasePacket(p);

    SCReturn;
}

/**
 *  \brief Release all the packets in the queue back to the packetpool.  Mainly
 *         used by threads that have failed, and wants to return the packets back
 *         to the packetpool.
 *
 *  \param pq Pointer to the packetqueue from which the packets have to be
 *            returned back to the packetpool
 *
 *  \warning this function assumes that the pq does not use locking
 */
void TmqhReleasePacketsToPacketPool(PacketQueue *pq)
{
    Packet *p = NULL;

    if (pq == NULL)
        return;

    while ((p = PacketDequeue(pq)) != NULL) {
        DEBUG_VALIDATE_BUG_ON(p->flow != NULL);
        TmqhOutputPacketpool(NULL, p);
    }
}

/** number of packets to keep reserved when calculating the pending
 *  return packets count. This assumes we need at max 10 packets in one
 *  PacketPoolWaitForN call. The actual number is 9 now, so this has a
 *  bit of margin. */
#define RESERVED_PACKETS 10

/**
 *  \brief Set the max_pending_return_packets value
 *
 *  Set it to the max pending packets value, divided by the number
 *  of lister threads. Normally, in autofp these are the stream/detect/log
 *  worker threads.
 *
 *  The max_pending_return_packets value needs to stay below the packet
 *  pool size of the 'producers' (normally pkt capture threads but also
 *  flow timeout injection ) to avoid a deadlock where all the 'workers'
 *  keep packets in their return pools, while the capture thread can't
 *  continue because its pool is empty.
 */
void PacketPoolPostRunmodes(void)
{
    extern uint32_t max_pending_packets;
    uint32_t pending_packets = max_pending_packets;
    if (pending_packets < RESERVED_PACKETS) {
        FatalError("'max-pending-packets' setting "
                   "must be at least %d",
                RESERVED_PACKETS);
    }
    uint32_t threads = TmThreadCountThreadsByTmmFlags(TM_FLAG_FLOWWORKER_TM);
    if (threads == 0)
        return;

    uint32_t packets = (pending_packets / threads) - 1;
    if (packets < max_pending_return_packets)
        max_pending_return_packets = packets;

    /* make sure to have a margin in the return logic */
    if (max_pending_return_packets >= RESERVED_PACKETS)
        max_pending_return_packets -= RESERVED_PACKETS;

    SCLogDebug("detect threads %u, max packets %u, max_pending_return_packets %u",
            threads, packets, max_pending_return_packets);
}

1	/* Copyright (C) 2007-2022 Open Information Security Foundation
2	*
3	* You can copy, redistribute or modify this Program under the terms of
4	* the GNU General Public License version 2 as published by the Free
5	* Software Foundation.
6	*
7	* This program is distributed in the hope that it will be useful,
8	* but WITHOUT ANY WARRANTY; without even the implied warranty of
9	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
10	* GNU General Public License for more details.
11	*
12	* You should have received a copy of the GNU General Public License
13	* version 2 along with this program; if not, write to the Free Software
14	* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
15	* 02110-1301, USA.
16	*/
17
18	/**
19	* \file
20	*
21	* \author Victor Julien <victor@inliniac.net>
22	*
23	* Packetpool queue handlers. Packet pool is implemented as a stack.
24	*/
25
26	#include "suricata-common.h"
27	#include "tmqh-packetpool.h"
28	#include "tm-queuehandlers.h"
29	#include "tm-threads.h"
30	#include "threads.h"
31	#include "decode.h"
32	#include "tm-modules.h"
33	#include "packet.h"
34	#include "util-profiling.h"
35	#include "util-validate.h"
36	#include "action-globals.h"
37
38	extern uint32_t max_pending_packets;
39
40	/* Number of freed packet to save for one pool before freeing them. */
41	#define MAX_PENDING_RETURN_PACKETS 32
42	static uint32_t max_pending_return_packets = MAX_PENDING_RETURN_PACKETS;
43
44	thread_local PktPool thread_pkt_pool;
45
46	static inline PktPool *GetThreadPacketPool(void)
47	{	321,487,589✔
48	return &thread_pkt_pool;	321,487,589✔
49	}	321,487,589✔
50
51	static PacketQueueNoLock TmqhInputPacketpool(ThreadVars *);
52
53	/**
54	* \brief TmqhPacketpoolRegister
55	* \initonly
56	*/
57	void TmqhPacketpoolRegister (void)
58	{	2,215✔
59	tmqh_table[TMQH_PACKETPOOL].name = "packetpool";	2,215✔
60	tmqh_table[TMQH_PACKETPOOL].InHandler = TmqhInputPacketpool;	2,215✔
61	tmqh_table[TMQH_PACKETPOOL].OutHandler = TmqhOutputPacketpool;	2,215✔
62	}	2,215✔
63
64	static void UpdateReturnThreshold(PktPool *pool)
65	{	154,062,752✔
66	const float perc = (float)pool->cnt / (float)max_pending_packets;	154,062,752✔
67	uint32_t threshold = (uint32_t)(perc * (float)max_pending_return_packets);	154,062,752✔
68	if (threshold != SC_ATOMIC_GET(pool->return_stack.return_threshold)) {	154,062,752✔
69	SC_ATOMIC_SET(pool->return_stack.return_threshold, threshold);	368,097✔
70	}	368,097✔
71	}	154,062,752✔
72
73	void PacketPoolWait(void)
74	{	72,454✔
75	PktPool *my_pool = GetThreadPacketPool();	72,454✔
76
77	if (my_pool->head == NULL) {	72,454✔
78	SC_ATOMIC_SET(my_pool->return_stack.return_threshold, 1);	32,661✔
79
80	SCMutexLock(&my_pool->return_stack.mutex);	32,661✔
81	int rc = 0;	32,661✔
82	while (my_pool->return_stack.cnt == 0 && rc == 0) {	63,861✔
83	rc = SCCondWait(&my_pool->return_stack.cond, &my_pool->return_stack.mutex);	31,200✔
84	}	31,200✔
85	SCMutexUnlock(&my_pool->return_stack.mutex);	32,661✔
86
87	UpdateReturnThreshold(my_pool);	32,661✔
88	}	32,661✔
89	}	72,454✔
90
91	/** \brief a initialized packet
92	*
93	* \warning Use only at init, not at packet runtime
94	*/
95	static void PacketPoolStorePacket(Packet *p)
96	{	24,363,136✔
97	p->pool = GetThreadPacketPool();	24,363,136✔
98	p->ReleasePacket = PacketPoolReturnPacket;	24,363,136✔
99	PacketPoolReturnPacket(p);	24,363,136✔
100	}	24,363,136✔
101
102	static void PacketPoolGetReturnedPackets(PktPool *pool)
103	{	2,044,054✔
104	SCMutexLock(&pool->return_stack.mutex);	2,044,054✔
105	/* Move all the packets from the locked return stack to the local stack. */
106	pool->head = pool->return_stack.head;	2,044,054✔
107	pool->return_stack.head = NULL;	2,044,054✔
108	pool->cnt += pool->return_stack.cnt;	2,044,054✔
109	pool->return_stack.cnt = 0;	2,044,054✔
110	SCMutexUnlock(&pool->return_stack.mutex);	2,044,054✔
111	}	2,044,054✔
112
113	/** \brief Get a new packet from the packet pool
114	*
115	* Only allocates from the thread's local stack, or mallocs new packets.
116	* If the local stack is empty, first move all the return stack packets to
117	* the local stack.
118	* \retval Packet pointer, or NULL on failure.
119	*/
120	Packet *PacketPoolGetPacket(void)
121	{	151,010,406✔
122	PktPool *pool = GetThreadPacketPool();	151,010,406✔
123	DEBUG_VALIDATE_BUG_ON(pool->initialized == 0);	151,010,406✔
124	DEBUG_VALIDATE_BUG_ON(pool->destroyed == 1);	151,010,406✔
125	if (pool->head) {	153,744,136✔
126	/* Stack is not empty. */
127	Packet *p = pool->head;	151,700,189✔
128	pool->head = p->next;	151,700,189✔
129	pool->cnt--;	151,700,189✔
130	p->pool = pool;	151,700,189✔
131	PacketReinit(p);	151,700,189✔
132
133	UpdateReturnThreshold(pool);	151,700,189✔
134	SCLogDebug("pp: %0.2f cnt:%u max:%d threshold:%u",	151,700,189✔
135	((float)pool->cnt / (float)max_pending_packets) * (float)100, pool->cnt,	151,700,189✔
136	max_pending_packets, SC_ATOMIC_GET(pool->return_stack.return_threshold));	151,700,189✔
137	return p;	151,700,189✔
138	}	151,700,189✔
139
140	/* Local Stack is empty, so check the return stack, which requires
141	* locking. */
142	PacketPoolGetReturnedPackets(pool);	2,149,527,594✔
143
144	/* Try to allocate again. Need to check for not empty again, since the
145	* return stack might have been empty too.
146	*/
147	if (pool->head) {	2,149,527,594✔
148	/* Stack is not empty. */
149	Packet *p = pool->head;	115,672✔
150	pool->head = p->next;	115,672✔
151	pool->cnt--;	115,672✔
152	p->pool = pool;	115,672✔
153	PacketReinit(p);	115,672✔
154
155	UpdateReturnThreshold(pool);	115,672✔
156	SCLogDebug("pp: %0.2f cnt:%u max:%d threshold:%u",	115,672✔
157	((float)pool->cnt / (float)max_pending_packets) * (float)100, pool->cnt,	115,672✔
158	max_pending_packets, SC_ATOMIC_GET(pool->return_stack.return_threshold));	115,672✔
159	return p;	115,672✔
160	}	115,672✔
161
162	/* Failed to allocate a packet, so return NULL. */
163	/* Optionally, could allocate a new packet here. */
164	return NULL;	2,149,411,926✔
165	}	2,149,527,594✔
166
167	/** \brief Return packet to Packet pool
168	*
169	*/
170	void PacketPoolReturnPacket(Packet *p)
171	{	153,708,346✔
172	PktPool *my_pool = GetThreadPacketPool();	153,708,346✔
173	PktPool *pool = p->pool;	153,708,346✔
174	if (pool == NULL) {	153,708,346✔
175	PacketFree(p);	×
176	return;	×
177	}	×
178
179	PacketReleaseRefs(p);	153,708,346✔
180
181	#ifdef DEBUG_VALIDATION
182	BUG_ON(pool->initialized == 0);
183	BUG_ON(pool->destroyed == 1);
184	BUG_ON(my_pool->initialized == 0);
185	BUG_ON(my_pool->destroyed == 1);
186	#endif /* DEBUG_VALIDATION */
187
188	if (pool == my_pool) {	153,708,346✔
189	/* Push back onto this thread's own stack, so no locking. */
190	p->next = my_pool->head;	150,391,649✔
191	my_pool->head = p;	150,391,649✔
192	my_pool->cnt++;	150,391,649✔
193	} else {	150,391,649✔
194	PktPool *pending_pool = my_pool->pending_pool;	3,316,697✔
195	if (pending_pool == NULL \|\| pending_pool == pool) {	3,316,833✔
196	if (pending_pool == NULL) {	806,032✔
197	/* No pending packet, so store the current packet. */
198	p->next = NULL;	729,290✔
199	my_pool->pending_pool = pool;	729,290✔
200	my_pool->pending_head = p;	729,290✔
201	my_pool->pending_tail = p;	729,290✔
202	my_pool->pending_count = 1;	729,290✔
203	} else if (pending_pool == pool) {	729,575✔
204	/* Another packet for the pending pool list. */
205	p->next = my_pool->pending_head;	76,766✔
206	my_pool->pending_head = p;	76,766✔
207	my_pool->pending_count++;	76,766✔
208	}	76,766✔
209
210	const uint32_t threshold = SC_ATOMIC_GET(pool->return_stack.return_threshold);	806,032✔
211	if (my_pool->pending_count >= threshold) {	806,056✔
212	/* Return the entire list of pending packets. */
213	SCMutexLock(&pool->return_stack.mutex);	726,896✔
214	my_pool->pending_tail->next = pool->return_stack.head;	726,896✔
215	pool->return_stack.head = my_pool->pending_head;	726,896✔
216	pool->return_stack.cnt += my_pool->pending_count;	726,896✔
217	SCCondSignal(&pool->return_stack.cond);	726,896✔
218	SCMutexUnlock(&pool->return_stack.mutex);	726,896✔
219	/* Clear the list of pending packets to return. */
220	my_pool->pending_pool = NULL;	726,896✔
221	my_pool->pending_head = NULL;	726,896✔
222	my_pool->pending_tail = NULL;	726,896✔
223	my_pool->pending_count = 0;	726,896✔
224	}	726,896✔
225	} else {	4,297,478,095✔
226	/* Push onto return stack for this pool */
227	SCMutexLock(&pool->return_stack.mutex);	4,297,478,095✔
228	p->next = pool->return_stack.head;	4,297,478,095✔
229	pool->return_stack.head = p;	4,297,478,095✔
230	pool->return_stack.cnt++;	4,297,478,095✔
231	SCCondSignal(&pool->return_stack.cond);	4,297,478,095✔
232	SCMutexUnlock(&pool->return_stack.mutex);	4,297,478,095✔
233	}	4,297,478,095✔
234	}	3,316,697✔
235	}	153,708,346✔
236
237	void PacketPoolInit(void)
238	{	23,793✔
239	PktPool *my_pool = GetThreadPacketPool();	23,793✔
240
241	#ifdef DEBUG_VALIDATION
242	BUG_ON(my_pool->initialized);
243	my_pool->initialized = 1;
244	my_pool->destroyed = 0;
245	#endif /* DEBUG_VALIDATION */
246
247	SCMutexInit(&my_pool->return_stack.mutex, NULL);	23,793✔
248	SCCondInit(&my_pool->return_stack.cond, NULL);	23,793✔
249	SC_ATOMIC_INIT(my_pool->return_stack.return_threshold);	23,793✔
250	SC_ATOMIC_SET(my_pool->return_stack.return_threshold, 32);	23,793✔
251
252	/* pre allocate packets */
253	SCLogDebug("preallocating packets... packet size %" PRIuMAX "",	23,793✔
254	(uintmax_t)SIZE_OF_PACKET);	23,793✔
255	for (uint32_t i = 0; i < max_pending_packets; i++) {	24,386,929✔
256	Packet *p = PacketGetFromAlloc();	24,363,136✔
257	if (unlikely(p == NULL)) {	24,363,136✔
258	FatalError("Fatal error encountered while allocating a packet. Exiting...");	×
259	}	×
260	PacketPoolStorePacket(p);	24,363,136✔
261	}	24,363,136✔
262
263	//SCLogInfo("preallocated %"PRIiMAX" packets. Total memory %"PRIuMAX"",
264	// max_pending_packets, (uintmax_t)(max_pending_packets*SIZE_OF_PACKET));
265	}	23,793✔
266
267	void PacketPoolDestroy(void)
268	{	23,793✔
269	Packet *p = NULL;	23,793✔
270	PktPool *my_pool = GetThreadPacketPool();	23,793✔
271
272	#ifdef DEBUG_VALIDATION
273	BUG_ON(my_pool && my_pool->destroyed);
274	#endif /* DEBUG_VALIDATION */
275
276	if (my_pool && my_pool->pending_pool != NULL) {	23,793✔
277	p = my_pool->pending_head;	2,410✔
278	while (p) {	20,310✔
279	Packet *next_p = p->next;	17,900✔
280	PacketFree(p);	17,900✔
281	p = next_p;	17,900✔
282	my_pool->pending_count--;	17,900✔
283	}	17,900✔
284	#ifdef DEBUG_VALIDATION
285	BUG_ON(my_pool->pending_count);
286	#endif /* DEBUG_VALIDATION */
287	my_pool->pending_pool = NULL;	2,410✔
288	my_pool->pending_head = NULL;	2,410✔
289	my_pool->pending_tail = NULL;	2,410✔
290	}	2,410✔
291
292	while ((p = PacketPoolGetPacket()) != NULL) {	24,368,986✔
293	PacketFree(p);	24,345,193✔
294	}	24,345,193✔
295
296	#ifdef DEBUG_VALIDATION
297	my_pool->initialized = 0;
298	my_pool->destroyed = 1;
299	#endif /* DEBUG_VALIDATION */
300	}	23,793✔
301
302	static PacketQueueNoLock TmqhInputPacketpool(ThreadVars *tv)
303	{	×
NEW 304	Packet *p = PacketPoolGetPacket();	×
NEW 305	PacketQueueNoLock pq = { .top = p, .bot = p, .len = (p != NULL) };	×
NEW 306	return pq;	×
UNCOV 307	}	×
308
309	void TmqhOutputPacketpool(ThreadVars t, Packet p)
310	{	130,698,023✔
311	bool proot = false;	130,698,023✔
312
313	SCEnter();	130,698,023✔
314	SCLogDebug("Packet %p, p->root %p, alloced %s", p, p->root, BOOL2STR(p->pool == NULL));	130,698,023✔
315
316	if (PacketIsTunnel(p)) {	130,698,023✔
317	SCLogDebug("Packet %p is a tunnel packet: %s",	6,224✔
318	p,p->root ? "upper layer" : "tunnel root");	6,224✔
319
320	/* get a lock to access root packet fields */
321	SCSpinlock *lock = p->root ? &p->root->persistent.tunnel_lock : &p->persistent.tunnel_lock;	6,224✔
322	SCSpinLock(lock);	6,224✔
323
324	if (PacketIsTunnelRoot(p)) {	6,224✔
325	SCLogDebug("IS_TUNNEL_ROOT_PKT == TRUE");	3,101✔
326	CaptureStatsUpdate(t, p); // TODO move out of lock	3,101✔
327
328	const uint16_t outstanding = TUNNEL_PKT_TPR(p) - TUNNEL_PKT_RTV(p);	3,101✔
329	SCLogDebug("root pkt: outstanding %u", outstanding);	3,101✔
330	if (outstanding == 0) {	3,101✔
331	SCLogDebug("no tunnel packets outstanding, no more tunnel "	1,732✔
332	"packet(s) depending on this root");	1,732✔
333	/* if this packet is the root and there are no
334	* more tunnel packets to consider
335	*
336	* return it to the pool */
337	} else {	1,827✔
338	SCLogDebug("tunnel root Packet %p: outstanding > 0, so "	1,369✔
339	"packets are still depending on this root, setting "	1,369✔
340	"SET_TUNNEL_PKT_VERDICTED", p);	1,369✔
341	/* if this is the root and there are more tunnel
342	* packets, return this to the pool. It's still referenced
343	* by the tunnel packets, and we will return it
344	* when we handle them */
345	PacketTunnelSetVerdicted(p);	1,369✔
346
347	PACKET_PROFILING_END(p);	1,369✔
348	SCSpinUnlock(lock);	1,369✔
349	SCReturn;	1,369✔
350	}	1,369✔
351	} else {	3,123✔
352	SCLogDebug("NOT IS_TUNNEL_ROOT_PKT, so tunnel pkt");	3,123✔
353
354	TUNNEL_INCR_PKT_RTV_NOLOCK(p);	3,123✔
355	const uint16_t outstanding = TUNNEL_PKT_TPR(p) - TUNNEL_PKT_RTV(p);	3,123✔
356	SCLogDebug("tunnel pkt: outstanding %u", outstanding);	3,123✔
357	/* all tunnel packets are processed except us. Root already
358	* processed. So return tunnel pkt and root packet to the
359	* pool. */
360	if (outstanding == 0 && p->root && PacketTunnelIsVerdicted(p->root)) {	3,123✔
361	SCLogDebug("root verdicted == true && no outstanding");	1,369✔
362
363	/* handle freeing the root as well*/
364	SCLogDebug("setting proot = 1 for root pkt, p->root %p "	1,369✔
365	"(tunnel packet %p)", p->root, p);	1,369✔
366	proot = true;	1,369✔
367
368	/* fall through */
369
370	} else {	1,839✔
371	/* root not ready yet, or not the last tunnel packet,
372	* so get rid of the tunnel pkt only */
373
374	SCLogDebug("NOT IS_TUNNEL_PKT_VERDICTED (%s) \|\| "	1,754✔
375	"outstanding > 0 (%u)",	1,754✔
376	(p->root && PacketTunnelIsVerdicted(p->root)) ? "true" : "false",	1,754✔
377	outstanding);	1,754✔
378
379	/* fall through */
380	}	1,754✔
381	}	3,123✔
382	SCSpinUnlock(lock);	4,855✔
383
384	SCLogDebug("tunnel stuff done, move on (proot %d)", proot);	4,855✔
385
386	} else {	130,691,799✔
387	CaptureStatsUpdate(t, p);	130,691,799✔
388	}	130,691,799✔
389
390	SCLogDebug("[packet %p][%s] %s", p,	130,696,654✔
391	PacketIsTunnel(p) ? PacketIsTunnelRoot(p) ? "tunnel::root" : "tunnel::leaf"	130,696,654✔
392	: "no tunnel",	130,696,654✔
393	(p->action & ACTION_DROP) ? "DROP" : "no drop");	130,696,654✔
394
395	/* we're done with the tunnel root now as well */
396	if (proot) {	130,696,654✔
397	SCLogDebug("getting rid of root pkt... alloc'd %s", BOOL2STR(p->root->pool == NULL));	1,369✔
398
399	PacketReleaseRefs(p->root);	1,369✔
400	p->root->ReleasePacket(p->root);	1,369✔
401	p->root = NULL;	1,369✔
402	}	1,369✔
403
404	PACKET_PROFILING_END(p);	130,696,654✔
405
406	PacketReleaseRefs(p);	130,696,654✔
407	p->ReleasePacket(p);	130,696,654✔
408
409	SCReturn;	130,696,654✔
410	}	130,698,023✔
411
412	/**
413	* \brief Release all the packets in the queue back to the packetpool. Mainly
414	* used by threads that have failed, and wants to return the packets back
415	* to the packetpool.
416	*
417	* \param pq Pointer to the packetqueue from which the packets have to be
418	* returned back to the packetpool
419	*
420	* \warning this function assumes that the pq does not use locking
421	*/
422	void TmqhReleasePacketsToPacketPool(PacketQueue *pq)
423	{	×
424	Packet *p = NULL;	×
425
426	if (pq == NULL)	×
427	return;	×
428
429	while ((p = PacketDequeue(pq)) != NULL) {	×
430	DEBUG_VALIDATE_BUG_ON(p->flow != NULL);	×
431	TmqhOutputPacketpool(NULL, p);	×
432	}	×
433	}	×
434
435	/** number of packets to keep reserved when calculating the pending
436	* return packets count. This assumes we need at max 10 packets in one
437	* PacketPoolWaitForN call. The actual number is 9 now, so this has a
438	* bit of margin. */
439	#define RESERVED_PACKETS 10	8,884✔
440
441	/**
442	* \brief Set the max_pending_return_packets value
443	*
444	* Set it to the max pending packets value, divided by the number
445	* of lister threads. Normally, in autofp these are the stream/detect/log
446	* worker threads.
447	*
448	* The max_pending_return_packets value needs to stay below the packet
449	* pool size of the 'producers' (normally pkt capture threads but also
450	* flow timeout injection ) to avoid a deadlock where all the 'workers'
451	* keep packets in their return pools, while the capture thread can't
452	* continue because its pool is empty.
453	*/
454	void PacketPoolPostRunmodes(void)
455	{	3,427✔
456	extern uint32_t max_pending_packets;	3,427✔
457	uint32_t pending_packets = max_pending_packets;	3,427✔
458	if (pending_packets < RESERVED_PACKETS) {	3,427✔
459	FatalError("'max-pending-packets' setting "	×
460	"must be at least %d",	×
461	RESERVED_PACKETS);	×
462	}	×
463	uint32_t threads = TmThreadCountThreadsByTmmFlags(TM_FLAG_FLOWWORKER_TM);	3,427✔
464	if (threads == 0)	3,427✔
465	return;	2✔
466
467	uint32_t packets = (pending_packets / threads) - 1;	3,425✔
468	if (packets < max_pending_return_packets)	3,425✔
469	max_pending_return_packets = packets;	×
470
471	/* make sure to have a margin in the return logic */
472	if (max_pending_return_packets >= RESERVED_PACKETS)	3,425✔
473	max_pending_return_packets -= RESERVED_PACKETS;	2,032✔
474
475	SCLogDebug("detect threads %u, max packets %u, max_pending_return_packets %u",	3,425✔
476	threads, packets, max_pending_return_packets);	3,425✔
477	}	3,425✔

OISF / suricata / 23338889526

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous