For PktPool add local pending freed packets list.

Better handle the autofp case where one thread allocates the majority
of the packets and other threads free those packets.

Add a list of locally pending packets. The first packet freed goes on the
pending list, then subsequent freed packets for the same Packet Pool are
added to this list until it hits a fixed number of packets, then the
entire list of packets is pushed onto the pool's return stack. If a freed
packet is not for the pending pool, it is freed immediately to its pool's
return stack, as before.

For the autofp case, since there is only one Packet Pool doing all the
allocation, every other thread will keep a list of pending packets for
that pool.

For the worker run mode, most packets are allocated and freed locally. For
the case where packets are being returned to a remote pool, a pending list
will be kept for one of those other threads, all others are returned as before.

Which remote pool for which to keep a pending list is changed each time the
pending list is returned. Since the return pending pool is cleared when it is
freed, then next packet to be freed chooses the new pending pool.

diff --git a/src/tmqh-packetpool.c b/src/tmqh-packetpool.c
index 38ef1aa9be94f72afb7c36284306c5196b557453..3653888deef43c0f91ed155633c1debaa54946e3 100644 (file)
--- a/src/tmqh-packetpool.c
+++ b/src/tmqh-packetpool.c
@@ -51,6 +51,9 @@
 /* TODO: Handle case without __thread */
 __thread PktPool thread_pkt_pool;
 
+/* Number of freed packet to save for one pool before freeing them. */
+#define MAX_PENDING_RETURN_PACKETS 32
+
 /**
  * \brief TmqhPacketpoolRegister
  * \initonly
@@ -64,7 +67,7 @@ void TmqhPacketpoolRegister (void) {
 static int PacketPoolIsEmpty(void)
 {
     /* Check local stack first. */
-    if (thread_pkt_pool.head || thread_pkt_pool.return_head)
+    if (thread_pkt_pool.head || thread_pkt_pool.return_stack.head)
         return 0;
 
     return 1;
@@ -111,18 +114,18 @@ Packet *PacketPoolGetPacket(void)
 
     /* Local Stack is empty, so check the return stack, which requires
      * locking. */
-    SCMutexLock(&pool->return_mutex);
+    SCMutexLock(&pool->return_stack.mutex);
     /* Move all the packets from the locked return stack to the local stack. */
-    pool->head = pool->return_head;
-    pool->return_head = NULL;
-    SCMutexUnlock(&pool->return_mutex);
+    pool->head = pool->return_stack.head;
+    pool->return_stack.head = NULL;
+    SCMutexUnlock(&pool->return_stack.mutex);
 
     /* 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;
+        Packet *p = pool->return_stack.head;
         pool->head = p->next;
         p->pool = pool;
         return p;
@@ -151,11 +154,34 @@ void PacketPoolReturnPacket(Packet *p)
         p->next = thread_pkt_pool.head;
         thread_pkt_pool.head = p;
     } else {
-        /* Push onto return stack for this pool */
-        SCMutexLock(&pool->return_mutex);
-        p->next = pool->return_head;
-        pool->return_head = p;
-        SCMutexUnlock(&pool->return_mutex);
+        PktPool *pending_pool = thread_pkt_pool.pending_pool;
+        if (pending_pool == NULL) {
+            /* No pending packet, so store the current packet. */
+            thread_pkt_pool.pending_pool = pool;
+            thread_pkt_pool.pending_head = p;
+            thread_pkt_pool.pending_tail = p;
+            thread_pkt_pool.pending_count = 1;
+        } else if (pending_pool == pool) {
+            /* Another packet for the pending pool list. */
+            p->next = thread_pkt_pool.pending_head;
+            thread_pkt_pool.pending_head->next = p;
+            thread_pkt_pool.pending_count++;
+            if (thread_pkt_pool.pending_count > MAX_PENDING_RETURN_PACKETS) {
+                /* Return the entire list of pending packets. */
+                SCMutexLock(&pool->return_stack.mutex);
+                thread_pkt_pool.pending_tail->next = pool->return_stack.head;
+                pool->return_stack.head = thread_pkt_pool.pending_head;
+                SCMutexUnlock(&pool->return_stack.mutex);
+                /* Clear the list of pending packets to return. */
+                thread_pkt_pool.pending_pool = NULL;
+            }
+        } else {
+            /* Push onto return stack for this pool */
+            SCMutexLock(&pool->return_stack.mutex);
+            p->next = pool->return_stack.head;
+            pool->return_stack.head = p;
+            SCMutexUnlock(&pool->return_stack.mutex);
+        }
     }
 }
 
@@ -163,10 +189,11 @@ void PacketPoolInit(void)
 {
     extern intmax_t max_pending_packets;
 
-    SCMutexInit(&thread_pkt_pool.return_mutex, NULL);
+    SCMutexInit(&thread_pkt_pool.return_stack.mutex, NULL);
 
     /* pre allocate packets */
-    SCLogDebug("preallocating packets... packet size %" PRIuMAX "", (uintmax_t)SIZE_OF_PACKET);
+    SCLogDebug("preallocating packets... packet size %" PRIuMAX "",
+               (uintmax_t)SIZE_OF_PACKET);
     int i = 0;
     for (i = 0; i < max_pending_packets; i++) {
         Packet *p = PacketGetFromAlloc();
@@ -298,6 +325,14 @@ void TmqhOutputPacketpool(ThreadVars *t, Packet *p)
         p->root = NULL;
     }
 
+    /* if p uses extended data, free them */
+    if (p->ext_pkt) {
+        if (!(p->flags & PKT_ZERO_COPY)) {
+            SCFree(p->ext_pkt);
+        }
+        p->ext_pkt = NULL;
+    }
+
     PACKET_PROFILING_END(p);
 
     p->ReleasePacket(p);

diff --git a/src/tmqh-packetpool.h b/src/tmqh-packetpool.h
index aa564f9d2d8b89ce2eee31563c95ef54c3a0c8a0..ddb7060e3c05420617f6bdcad94a9c0ae0914b87 100644 (file)
--- a/src/tmqh-packetpool.h
+++ b/src/tmqh-packetpool.h
@@ -25,17 +25,39 @@
 #define __TMQH_PACKETPOOL_H__
 
 #include "decode.h"
+#include "threads.h"
+
+    /* Return stack, onto which other threads free packets. */
+typedef struct PktPoolLockedStack_{
+    /* linked list of free packets. */
+    SCSpinlock mutex;
+    Packet *head;
+} __attribute__((aligned(CLS))) PktPoolLockedStack;
 
 typedef struct PktPool_ {
-    /* link listed of free packets local to this thread. */
+    /* link listed of free packets local to this thread. 
+     * No mutex is needed.
+     */
     Packet *head;
-  
-    /* Return stack, onto which other threads free packets. */
-    struct {
-        /* linked list of free packets. */
-        Packet *return_head;
-        SCMutex return_mutex;
-    } __attribute__((aligned(CLS)));
+    /* Packets waiting (pending) to be returned to the given Packet
+     * Pool. Accumulate packets for the same pool until a theshold is
+     * reached, then return them all at once.  Keep the head and tail
+     * to fast insertion of the entire list onto a return stack.
+     */
+    struct PktPool_ *pending_pool;
+    Packet *pending_head;
+    Packet *pending_tail;
+    uint32_t pending_count;
+    
+    /* All members above this point are accessed locally by only one thread, so 
+     * these should live on their own cache line.
+     */
+
+    /* Return stack, where other threads put packets that they free that belong
+     * to this thread.
+     */
+    PktPoolLockedStack return_stack;
+
 } PktPool;
 
 Packet *TmqhInputPacketpool(ThreadVars *);