Changeset a49a9eb

Timestamp:

07/30/14 18:44:16 (7 years ago)

Author:

Richard Sanger <rsangerarj@…>

Branches:

4.0.1-hotfixes, cachetimestamps, develop, dpdk-ndag, etsilive, libtrace4, master, ndag_format, pfring, rc-4.0.1, rc-4.0.2, rc-4.0.3, rc-4.0.4, ringdecrementfix, ringperformance, ringtimestampfixes

Children:

be3f75b

Parents:

41148f2

Message:

Add an object cache with thread local caches
All packets used by a trace are put through this.
Adds bulk read/write operations to the ringbuffer (used by the object cache)
Replace semaphores with condition variables to support these bulk operations.
Internally use bulk read operations from a single threaded formats to reduce lock overhead.
Replaces the asserts around pthread_* functions with a version that will still run the command if NDEBUG

Files:

• lib/Makefile.am (modified) (1 diff)
• lib/data-struct/deque.c (modified) (13 diffs)
• lib/data-struct/message_queue.c (modified) (4 diffs)
• lib/data-struct/message_queue.h (modified) (1 diff)
• lib/data-struct/object_cache.c (added)
• lib/data-struct/object_cache.h (added)
• lib/data-struct/ring_buffer.c (modified) (10 diffs)
• lib/data-struct/ring_buffer.h (modified) (2 diffs)
• lib/data-struct/vector.c (modified) (10 diffs)
• lib/libtrace.h.in (modified) (2 diffs)
• lib/libtrace_int.h (modified) (2 diffs)
• lib/trace.c (modified) (7 diffs)
• lib/trace_parallel.c (modified) (58 diffs)
• test/test-datastruct-ringbuffer.c (modified) (3 diffs)

lib/Makefile.am

@@ -58 +58 @@
                 libtrace_arphrd.h \
                 data-struct/ring_buffer.c data-struct/vector.c data-struct/message_queue.c \
-                data-struct/deque.c data-struct/sliding_window.c \
+                data-struct/deque.c data-struct/sliding_window.c data-struct/object_cache.c \
                 hash_toeplitz.c
 

lib/data-struct/deque.c

@@ -25 +25 @@
         q->size = 0;
         q->element_size = element_size;
-        assert(pthread_mutex_init(&q->lock, NULL) == 0);
+        ASSERT_RET(pthread_mutex_init(&q->lock, NULL), == 0);
 }
 
@@ -36 +36 @@
         memcpy(&new_node->data, d, q->element_size);
         // Only ->prev is unknown at this stage to be completed in lock
-        assert(pthread_mutex_lock(&q->lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&q->lock), == 0);
         if (q->head == NULL) {
                 assert(q->tail == NULL && q->size == 0);
@@ -48 +48 @@
         }
         q->size++;
-        assert(pthread_mutex_unlock(&q->lock) == 0);
+        ASSERT_RET(pthread_mutex_unlock(&q->lock), == 0);
 }
 
@@ -59 +59 @@
         memcpy(&new_node->data, d, q->element_size);
         // Only ->next is unknown at this stage to be completed in lock
-        assert(pthread_mutex_lock(&q->lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&q->lock), == 0);
         if (q->head == NULL) {
                 assert(q->tail == NULL && q->size == 0);
@@ -72 +72 @@
         }
         q->size++;
-        assert(pthread_mutex_unlock(&q->lock) == 0);
+        ASSERT_RET(pthread_mutex_unlock(&q->lock), == 0);
 }
 
@@ -78 +78 @@
 {
         int ret = 1;
-        assert(pthread_mutex_lock(&q->lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&q->lock), == 0);
         if (q->head == NULL)
                 ret = 0;
         else
                 memcpy(d, &q->head->data, q->element_size);
-        assert(pthread_mutex_unlock(&q->lock) == 0);
+        ASSERT_RET(pthread_mutex_unlock(&q->lock), == 0);
         return ret;
 }
@@ -90 +90 @@
 {
         int ret = 1;
-        assert(pthread_mutex_lock(&q->lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&q->lock), == 0);
         if (q->tail == NULL)
                 ret = 0;
         else
                 memcpy(d, &q->tail->data, q->element_size);
-        assert(pthread_mutex_unlock(&q->lock) == 0);
+        ASSERT_RET(pthread_mutex_unlock(&q->lock), == 0);
         return ret;
 }
@@ -103 +103 @@
         int ret = 0;
         list_node_t * n = NULL;
-        assert(pthread_mutex_lock(&q->lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&q->lock), == 0);
         if (q->head != NULL) {
                 n = q->head;
@@ -114 +114 @@
                         q->tail = q->head;
         }
-        assert(pthread_mutex_unlock(&q->lock) == 0);
+        ASSERT_RET(pthread_mutex_unlock(&q->lock), == 0);
         // Unlock once we've removed it :)
         if (ret) {
@@ -127 +127 @@
         int ret = 0;
         list_node_t * n;
-        assert(pthread_mutex_lock(&q->lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&q->lock), == 0);
         if (q->tail != NULL) {
                 n = q->tail;
@@ -138 +138 @@
                         q->head = q->tail;
         }
-        assert(pthread_mutex_unlock(&q->lock) == 0);
+        ASSERT_RET(pthread_mutex_unlock(&q->lock), == 0);
         if (ret) {
                 memcpy(d, &n->data, q->element_size);
@@ -150 +150 @@
 #if RACE_SAFE
         size_t ret;
-        assert(pthread_mutex_lock(&q->lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&q->lock), == 0);
         ret = q->size;
-        assert(pthread_mutex_unlock(&q->lock) == 0);
+        ASSERT_RET(pthread_mutex_unlock(&q->lock), == 0);
         return ret;
 #else
@@ -169 +169 @@
         list_node_t *n;
         assert(q->element_size == sizeof(libtrace_result_t));
-        assert(pthread_mutex_lock(&q->lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&q->lock), == 0);
         n = q->head;
         for (n = q->head; n != NULL; n = n->next) {
                 (*fn)(&n->data);
         }
-        assert(pthread_mutex_unlock(&q->lock) == 0);
+        ASSERT_RET(pthread_mutex_unlock(&q->lock), == 0);
 }

lib/data-struct/message_queue.c

@@ -19 +19 @@
 {
         assert(message_len);
-        assert(pipe(mq->pipefd) != -1);
+        ASSERT_RET(pipe(mq->pipefd), != -1);
         mq->message_count = 0;
         if (message_len > PIPE_BUF)
@@ -45 +45 @@
         int ret;
         assert(mq->message_len);
-        assert(write(mq->pipefd[1], message, mq->message_len) == (int) mq->message_len);
+        ASSERT_RET(write(mq->pipefd[1], message, mq->message_len), == (int) mq->message_len);
         // Update after we've written
         pthread_spin_lock(&mq->spin);
@@ -73 +73 @@
         ret = mq->message_count--;
         pthread_spin_unlock(&mq->spin);
-        assert(read(mq->pipefd[0], message, mq->message_len) == (int) mq->message_len);
+        ASSERT_RET(read(mq->pipefd[0], message, mq->message_len), == (int) mq->message_len);
         return ret;
 }
@@ -102 +102 @@
                 ret = --mq->message_count;
                 // :( read(...) needs to be done within the *spin* lock otherwise blocking might steal our read
-                assert(read(mq->pipefd[0], message, mq->message_len) == (int) mq->message_len);
+                ASSERT_RET(read(mq->pipefd[0], message, mq->message_len), == (int) mq->message_len);
         } else {
                 ret = LIBTRACE_MQ_FAILED;

lib/data-struct/message_queue.h

@@ -1 +1 @@
 #include <pthread.h>
 #include <limits.h>
+#include "../libtrace.h"
 
 #ifndef LIBTRACE_MESSAGE_QUEUE

lib/data-struct/ring_buffer.c

@@ -7 +7 @@
 #include <stdlib.h>
 #include <assert.h>
+#include <string.h>
+#include <stdio.h>
 
 #define LOCK_TYPE_MUTEX 0 // Default if not defined
 #define LOCK_TYPE_SPIN 1
-#define LOCK_TYPE_SEMAPHORE 2
-#define LOCK_TYPE_NONE 3
+#define LOCK_TYPE_NONE 2
 
 // No major difference noticed here between mutex and spin, both have there
 // downsides.
 
-#define USE_MODULUS 1
 #define USE_CHECK_EARLY 1
 
-#define USE_LOCK_TYPE LOCK_TYPE_SPIN
+#define USE_LOCK_TYPE LOCK_TYPE_MUTEX
 #if USE_LOCK_TYPE == LOCK_TYPE_SPIN
-#       define LOCK(dir) assert(pthread_spin_lock(&rb->s ## dir ## lock) == 0)
-#       define UNLOCK(dir) assert(pthread_spin_unlock(&rb->s ## dir ## lock) == 0)
+#       define LOCK(dir) ASSERT_RET(pthread_spin_lock(&rb->s ## dir ## lock), == 0)
+#       define UNLOCK(dir) ASSERT_RET(pthread_spin_unlock(&rb->s ## dir ## lock), == 0)
 #       define TRY_LOCK(dir, action) if(pthread_spin_lock(&rb->s ## dir ## lock) != 0) { \
-                                                                action }
-#elif USE_LOCK_TYPE == LOCK_TYPE_SEMAPHORE
-#       define LOCK(dir) assert(sem_wait(&rb->sem ## dir ## lock) == 0)
-#       define UNLOCK(dir) assert(sem_post(&rb->sem ## dir ## lock) == 0)
-#       define TRY_LOCK(dir, action) if(sem_trywait(&rb->sem ## dir ## lock) != 0) { \
                                                                 action }
 #elif USE_LOCK_TYPE == LOCK_TYPE_NONE
@@ -35 +30 @@
 #       define TRY_LOCK(dir, action)
 #else // Mutex
-#       define LOCK(dir) assert(pthread_mutex_lock(&rb-> dir ## lock) == 0)
-#       define UNLOCK(dir) assert(pthread_mutex_unlock(&rb-> dir ## lock) == 0)
+#       define LOCK(dir) ASSERT_RET(pthread_mutex_lock(&rb-> dir ## lock), == 0)
+#       define UNLOCK(dir) ASSERT_RET(pthread_mutex_unlock(&rb-> dir ## lock), == 0)
 #       define TRY_LOCK(dir, action) if(pthread_mutex_lock(&rb-> dir ## lock) != 0) {\
                                                                 action }
@@ -64 +59 @@
         if (mode == LIBTRACE_RINGBUFFER_BLOCKING) {
                 /* The signaling part - i.e. release when data's ready to read */
-                assert(sem_init(&rb->fulls, 0, 0) == 0);
-                assert(sem_init(&rb->emptys, 0, size - 1) == 0); // REMEMBER the -1 here :) very important
+                pthread_cond_init(&rb->full_cond, NULL);
+                pthread_cond_init(&rb->empty_cond, NULL);
+                ASSERT_RET(pthread_mutex_init(&rb->empty_lock, NULL), == 0);
+                ASSERT_RET(pthread_mutex_init(&rb->full_lock, NULL), == 0);
         }
         /* The mutual exclusion part */
 #if USE_LOCK_TYPE == LOCK_TYPE_SPIN
 #warning "using spinners"
-        assert(pthread_spin_init(&rb->swlock, 0) == 0);
-        assert(pthread_spin_init(&rb->srlock, 0) == 0);
-#elif USE_LOCK_TYPE == LOCK_TYPE_SEMAPHORE
-#warning "using semaphore"
-        assert(sem_init(&rb->semrlock, 0, 1) != -1);
-        assert(sem_init(&rb->semwlock, 0, 1) != -1);
+        ASSERT_RET(pthread_spin_init(&rb->swlock, 0), == 0);
+        ASSERT_RET(pthread_spin_init(&rb->srlock, 0), == 0);
 #elif USE_LOCK_TYPE == LOCK_TYPE_NONE
 #warning "No locking used"
-#else /* USE_LOCK_TYPE == LOCK_TYPE_MUTEX */
-        assert(pthread_mutex_init(&rb->wlock, NULL) == 0);
-        assert(pthread_mutex_init(&rb->rlock, NULL) == 0);
+#else
+        ASSERT_RET(pthread_mutex_init(&rb->wlock, NULL), == 0);
+        ASSERT_RET(pthread_mutex_init(&rb->rlock, NULL), == 0);
 #endif
 }
@@ -90 +83 @@
 DLLEXPORT void libtrace_ringbuffer_destroy(libtrace_ringbuffer_t * rb) {
 #if USE_LOCK_TYPE == LOCK_TYPE_SPIN
-        assert(pthread_spin_destroy(&rb->swlock) == 0);
-        assert(pthread_spin_destroy(&rb->srlock) == 0);
-#elif USE_LOCK_TYPE == LOCK_TYPE_SEMAPHORE
-        assert(sem_destroy(&rb->semrlock) != -1);
-        assert(sem_destroy(&rb->semwlock) != -1);
+        ASSERT_RET(pthread_spin_destroy(&rb->swlock), == 0);
+        ASSERT_RET(pthread_spin_destroy(&rb->srlock), == 0);
 #elif USE_LOCK_TYPE == LOCK_TYPE_NONE
-#else /* USE_LOCK_TYPE == LOCK_TYPE_MUTEX */
-        assert(pthread_mutex_destroy(&rb->wlock) == 0);
-        assert(pthread_mutex_destroy(&rb->rlock) == 0);
-#endif
+#endif
+        ASSERT_RET(pthread_mutex_destroy(&rb->wlock), == 0);
+        ASSERT_RET(pthread_mutex_destroy(&rb->rlock), == 0);
         if (rb->mode == LIBTRACE_RINGBUFFER_BLOCKING) {
-                assert(sem_destroy(&rb->fulls) == 0);
-                assert(sem_destroy(&rb->emptys) == 0);
+                pthread_cond_destroy(&rb->full_cond);
+                pthread_cond_destroy(&rb->empty_cond);
         }
         rb->size = 0;
@@ -126 +115 @@
  */
 DLLEXPORT int libtrace_ringbuffer_is_full(const libtrace_ringbuffer_t * rb) {
-#if USE_MODULUS
         return rb->start == ((rb->end + 1) % rb->size);
-#else
-        return rb->start == ((rb->end + 1 < rb->size) ? rb->end + 1 : 0);
-#endif
-}
-
-/**
- * Performs a blocking write to the buffer, upon return the value will be
- * stored. This will not clobber old values.
- * 
- * This assumes only one thread writing at once. Use 
- * libtrace_ringbuffer_swrite for a thread safe version.
- * 
- * @param rb a pointer to libtrace_ringbuffer structure
- * @param value the value to store
- */
-DLLEXPORT void libtrace_ringbuffer_write(libtrace_ringbuffer_t * rb, void* value) {
+}
+
+static inline size_t libtrace_ringbuffer_nb_full(const libtrace_ringbuffer_t *rb) {
+        if (rb->end < rb->start)
+                return rb->end + rb->size - rb->start;
+        else
+                return rb->end - rb->start;
+        // return (rb->end + rb->size - rb->start) % rb->size;
+}
+
+static inline size_t libtrace_ringbuffer_nb_empty(const libtrace_ringbuffer_t *rb) {
+        if (rb->start <= rb->end)
+                return rb->start + rb->size - rb->end - 1;
+        else
+                return rb->start - rb->end - 1;
+        // return (rb->start + rb->size - rb->end - 1) % rb->size;
+}
+
+/**
+ * Waits for a empty slot, that we can write to.
+ * @param rb The ringbuffer
+ */
+static inline void wait_for_empty(libtrace_ringbuffer_t *rb) {
         /* Need an empty to start with */
-        if (rb->mode == LIBTRACE_RINGBUFFER_BLOCKING)
-                assert(sem_wait(&rb->emptys) == 0);
-        else 
+        if (rb->mode == LIBTRACE_RINGBUFFER_BLOCKING) {
+                pthread_mutex_lock(&rb->empty_lock);
+                while (libtrace_ringbuffer_is_full(rb))
+                        pthread_cond_wait(&rb->empty_cond, &rb->empty_lock);
+                pthread_mutex_unlock(&rb->empty_lock);
+        } else {
                 while (libtrace_ringbuffer_is_full(rb))
                         /* Yield our time, why?, we tried and failed to write an item
@@ -154 +152 @@
                          * burst to write without blocking */
                         sched_yield();//_mm_pause();
-
+        }
+}
+
+/**
+ * Waits for a full slot, that we read from.
+ * @param rb The ringbuffer
+ */
+static inline void wait_for_full(libtrace_ringbuffer_t *rb) {
+        /* Need an empty to start with */
+        if (rb->mode == LIBTRACE_RINGBUFFER_BLOCKING) {
+                pthread_mutex_lock(&rb->full_lock);
+                while (libtrace_ringbuffer_is_empty(rb))
+                        pthread_cond_wait(&rb->full_cond, &rb->full_lock);
+                pthread_mutex_unlock(&rb->full_lock);
+        } else {
+                while (libtrace_ringbuffer_is_empty(rb))
+                        /* Yield our time, why?, we tried and failed to write an item
+                         * to the buffer - so we should give up our time in the hope
+                         * that the reader thread can empty the buffer giving us a good
+                         * burst to write without blocking */
+                        sched_yield();//_mm_pause();
+        }
+}
+
+/**
+ * Notifies we have created a full slot, after a write.
+ * @param rb The ringbuffer
+ */
+static inline void notify_full(libtrace_ringbuffer_t *rb) {
+        /* Need an empty to start with */
+        if (rb->mode == LIBTRACE_RINGBUFFER_BLOCKING) {
+                pthread_mutex_lock(&rb->full_lock);
+                pthread_cond_broadcast(&rb->full_cond);
+                pthread_mutex_unlock(&rb->full_lock);
+        }
+}
+
+/**
+ * Notifies we have created an empty slot, after a read.
+ * @param rb The ringbuffer
+ */
+static inline void notify_empty(libtrace_ringbuffer_t *rb) {
+        /* Need an empty to start with */
+        if (rb->mode == LIBTRACE_RINGBUFFER_BLOCKING) {
+                pthread_mutex_lock(&rb->empty_lock);
+                pthread_cond_broadcast(&rb->empty_cond);
+                pthread_mutex_unlock(&rb->empty_lock);
+        }
+}
+
+/**
+ * Performs a blocking write to the buffer, upon return the value will be
+ * stored. This will not clobber old values.
+ * 
+ * This assumes only one thread writing at once. Use 
+ * libtrace_ringbuffer_swrite for a thread safe version.
+ * 
+ * @param rb a pointer to libtrace_ringbuffer structure
+ * @param value the value to store
+ */
+DLLEXPORT void libtrace_ringbuffer_write(libtrace_ringbuffer_t * rb, void* value) {
+        /* Need an empty to start with */
+        wait_for_empty(rb);
         rb->elements[rb->end] = value;
-#if USE_MODULUS
         rb->end = (rb->end + 1) % rb->size;
-#else
-        rb->end = (rb->end + 1 < rb->size) ? rb->end + 1 : 0;
-#endif
-        /* This check is bad we can easily lose our time slice, and the reader
-         * can catch up before it should, in this case spin locking is used */
-        //if (libtrace_ringbuffer_is_empty(rb))
-        //      assert(0 == 1);
-        /* Now we've made another full */
-        if (rb->mode == LIBTRACE_RINGBUFFER_BLOCKING)
-                assert(sem_post(&rb->fulls) == 0);
+        notify_full(rb);
+}
+
+/**
+ * Performs a blocking write to the buffer, upon return the value will be
+ * stored. This will not clobber old values.
+ * 
+ * This assumes only one thread writing at once. Use 
+ * libtrace_ringbuffer_swrite for a thread safe version.
+ *
+ * Packets are written out from start to end in order, if only some packets are
+ * written those at the end of the array will be still be unwritten.
+ *
+ * @param rb a pointer to libtrace_ringbuffer structure
+ * @param values A pointer to a memory address read in
+ * @param nb_buffer The maximum buffers to write i.e. the length of values
+ * @param min_nb_buffers The minimum number of buffers to write
+ * @param value the value to store
+ */
+DLLEXPORT size_t libtrace_ringbuffer_write_bulk(libtrace_ringbuffer_t * rb, void *values[], size_t nb_buffers, size_t min_nb_buffers) {
+        size_t nb_ready;
+        size_t i = 0;
+        
+        assert(min_nb_buffers <= nb_buffers);
+        if (!min_nb_buffers && libtrace_ringbuffer_is_full(rb))
+                return 0;
+
+        do {
+                register size_t end;
+                wait_for_empty(rb);
+                nb_ready = libtrace_ringbuffer_nb_empty(rb);
+                nb_ready = MIN(nb_ready, nb_buffers-i);
+                nb_ready += i;
+                // TODO consider optimising into at most 2 memcpys??
+                end = rb->end;
+                for (; i < nb_ready; i++) {
+                        rb->elements[end] = values[i];
+                        end = (end + 1) % rb->size;
+                }
+                rb->end = end;
+                notify_full(rb);
+        } while (i < min_nb_buffers);
+        return i;
 }
 
@@ -198 +290 @@
         
         /* We need a full slot */
-        if (rb->mode == LIBTRACE_RINGBUFFER_BLOCKING)
-                assert(sem_wait(&rb->fulls) == 0);
-        else
-                while (libtrace_ringbuffer_is_empty(rb)) 
-                        /* Yield our time, why?, we tried and failed to read an item
-                         * from the buffer - so we should give up our time in the hope
-                         * that the writer thread can fill the buffer giving us a good
-                         * burst to read without blocking etc */
-                        sched_yield();//_mm_pause();
+        wait_for_full(rb);
+        value = rb->elements[rb->start];
+        rb->start = (rb->start + 1) % rb->size;
+        /* Now that's an empty slot */
+        notify_empty(rb);
+        return value;
+}
+
+/**
+ * Waits and reads from the supplied buffer, note this will block forever.
+ * Attempts to read the requested number of packets, however will return
+ * with only the number that are currently ready.
+ *
+ * Set min_nb_buffers to 0 to 'try' read packets.
+ *
+ * The buffer is filled from start to finish i.e. if 2 is returned [0] and [1]
+ * are valid.
+ * 
+ * @param rb a pointer to libtrace_ringbuffer structure
+ * @param values A pointer to a memory address where the returned item would be placed
+ * @param nb_buffer The maximum buffers to read i.e. the length of values
+ * @param min_nb_buffers The minimum number of buffers to read
+ * @return The number of packets read
+ */
+DLLEXPORT size_t libtrace_ringbuffer_read_bulk(libtrace_ringbuffer_t *rb, void *values[], size_t nb_buffers, size_t min_nb_buffers) {
+        size_t nb_ready;
+        size_t i = 0;
         
-        value = rb->elements[rb->start];
-#if USE_MODULUS
-        rb->start = (rb->start + 1) % rb->size;
-#else
-        rb->start = (rb->start + 1 < rb->size) ? rb->start + 1 : 0;
-#endif
-        /* Now that's a empty slot */
-        if (rb->mode == LIBTRACE_RINGBUFFER_BLOCKING)
-                assert(sem_post(&rb->emptys) == 0);
-        return value;
-}
+        assert(min_nb_buffers <= nb_buffers);
+
+        if (!min_nb_buffers && libtrace_ringbuffer_is_empty(rb))
+                return 0;
+
+        do {
+                register size_t start;
+                /* We need a full slot */
+                wait_for_full(rb);
+
+                nb_ready = libtrace_ringbuffer_nb_full(rb);
+                nb_ready = MIN(nb_ready, nb_buffers-i);
+                // Additional to the i we've already read
+                nb_ready += i;
+                start = rb->start;
+                for (; i < nb_ready; i++) {
+                        values[i] = rb->elements[start];
+                        start = (start + 1) % rb->size;
+                }
+                rb->start = start;
+                /* Now that's an empty slot */
+                notify_empty(rb);
+        } while (i < min_nb_buffers);
+        return i;
+}
+
 
 /**
@@ -242 +367 @@
         libtrace_ringbuffer_write(rb, value);
         UNLOCK(w);
+}
+
+/**
+ * A thread safe version of libtrace_ringbuffer_write_bulk
+ */
+DLLEXPORT size_t libtrace_ringbuffer_swrite_bulk(libtrace_ringbuffer_t * rb, void *values[], size_t nb_buffers, size_t min_nb_buffers) {
+        size_t ret;
+#if USE_CHECK_EARLY
+        if (!min_nb_buffers && libtrace_ringbuffer_is_full(rb)) // Check early
+                return 0;
+#endif
+        LOCK(w);
+        ret = libtrace_ringbuffer_write_bulk(rb, values, nb_buffers, min_nb_buffers);
+        UNLOCK(w);
+        return ret;
 }
 
@@ -290 +430 @@
 
 /**
+ * A thread safe version of libtrace_ringbuffer_read_bulk
+ */
+DLLEXPORT size_t libtrace_ringbuffer_sread_bulk(libtrace_ringbuffer_t * rb, void *values[], size_t nb_buffers, size_t min_nb_buffers) {
+        size_t ret;
+#if USE_CHECK_EARLY
+        if (!min_nb_buffers && libtrace_ringbuffer_is_empty(rb)) // Check early
+                return 0;
+#endif
+        LOCK(r);
+        ret = libtrace_ringbuffer_read_bulk(rb, values, nb_buffers, min_nb_buffers);
+        UNLOCK(r);
+        return ret;
+}
+
+/**
  * A thread safe version of libtrace_ringbuffer_try_write
  */
@@ -330 +485 @@
         rb->elements = NULL;
 }
+
+

lib/data-struct/ring_buffer.h

@@ -20 +20 @@
         pthread_spinlock_t swlock;
         pthread_spinlock_t srlock;
-        sem_t semrlock;
-        sem_t semwlock;
-        sem_t emptys;
-        sem_t fulls;
+        // We need to ensure that broadcasts dont get lost hence
+        // these locks below
+        // We avoid using semaphores since they don't allow
+        // multiple releases.
+        pthread_mutex_t empty_lock;
+        pthread_mutex_t full_lock;
+        pthread_cond_t empty_cond; // Signal when empties are ready
+        pthread_cond_t full_cond; // Signal when fulls are ready
         // Aim to get this on a separate cache line to start - important if spinning
         volatile size_t end;
@@ -46 +50 @@
 DLLEXPORT int libtrace_ringbuffer_try_sread_bl(libtrace_ringbuffer_t *rb, void ** value);
 
+
+
+DLLEXPORT size_t libtrace_ringbuffer_write_bulk(libtrace_ringbuffer_t *rb, void *values[], size_t nb_buffers, size_t min_nb_buffers);
+DLLEXPORT size_t libtrace_ringbuffer_read_bulk(libtrace_ringbuffer_t *rb, void *values[], size_t nb_buffers, size_t min_nb_buffers);
+DLLEXPORT size_t libtrace_ringbuffer_sread_bulk(libtrace_ringbuffer_t *rb, void *values[], size_t nb_buffers, size_t min_nb_buffers);
+DLLEXPORT size_t libtrace_ringbuffer_swrite_bulk(libtrace_ringbuffer_t *rb, void *values[], size_t nb_buffers, size_t min_nb_buffers);
+
 #endif

lib/data-struct/vector.c

@@ -10 +10 @@
         v->max_size = 128; // Pick a largish size to begin with
         v->elements = malloc(v->max_size * v->element_size);
-        assert(pthread_mutex_init(&v->lock, NULL) == 0);
+        ASSERT_RET(pthread_mutex_init(&v->lock, NULL), == 0);
 }
 
 DLLEXPORT void libtrace_vector_destroy(libtrace_vector_t *v) {
-        assert(pthread_mutex_destroy(&v->lock) == 0);
+        ASSERT_RET(pthread_mutex_destroy(&v->lock), == 0);
         free(v->elements);
         // Be safe make sure we wont work any more
@@ -24 +24 @@
 
 DLLEXPORT void libtrace_vector_push_back(libtrace_vector_t *v, void *d) {
-        assert(pthread_mutex_lock(&v->lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&v->lock), == 0);
         if (v->size >= v->max_size) {
                 /* Resize */
@@ -33 +33 @@
         memcpy(&v->elements[v->size*v->element_size], d, v->element_size);
         v->size++;
-        assert(pthread_mutex_unlock(&v->lock) == 0);
+        ASSERT_RET(pthread_mutex_unlock(&v->lock), == 0);
 }
 
@@ -41 +41 @@
 
 DLLEXPORT int libtrace_vector_get(libtrace_vector_t *v, size_t location, void *d) {
-        assert(pthread_mutex_lock(&v->lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&v->lock), == 0);
         if (location >= v->size) {
-                assert(pthread_mutex_unlock(&v->lock) == 0);
+                ASSERT_RET(pthread_mutex_unlock(&v->lock), == 0);
                 return 0;
         }
         memcpy(d, &v->elements[location*v->element_size], v->element_size);
-        assert(pthread_mutex_unlock(&v->lock) == 0);
+        ASSERT_RET(pthread_mutex_unlock(&v->lock), == 0);
         return 1;
 }
@@ -53 +53 @@
 DLLEXPORT int libtrace_vector_remove_front(libtrace_vector_t *v) {
         size_t i;
-        assert(pthread_mutex_lock(&v->lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&v->lock), == 0);
         if (!v->size) {
-                assert(pthread_mutex_unlock(&v->lock) == 0);
+                ASSERT_RET(pthread_mutex_unlock(&v->lock), == 0);
                 return 0;
         }
@@ -62 +62 @@
         for (i = 0; i < v->size * v->element_size; i++)
                 v->elements[i] = v->elements[i+v->element_size];
-        assert(pthread_mutex_unlock(&v->lock) == 0);
+        ASSERT_RET(pthread_mutex_unlock(&v->lock), == 0);
         return 1;
 }
@@ -82 +82 @@
         if (src->size == 0) // Nothing to do if this is the case
                 return;
-        assert(pthread_mutex_lock(&dest->lock) == 0);
-        assert(pthread_mutex_lock(&src->lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&dest->lock), == 0);
+        ASSERT_RET(pthread_mutex_lock(&src->lock), == 0);
         if (src->size == 0) // Double check now we've got the locks - Nothing to do if this is the case
                 goto unlock;
@@ -104 +104 @@
         }
 unlock:
-        assert(pthread_mutex_unlock(&src->lock) == 0);
-        assert(pthread_mutex_unlock(&dest->lock) == 0);
+        ASSERT_RET(pthread_mutex_unlock(&src->lock), == 0);
+        ASSERT_RET(pthread_mutex_unlock(&dest->lock), == 0);
 }
 
@@ -117 +117 @@
 
 DLLEXPORT void libtrace_vector_empty(libtrace_vector_t *v) {
-        assert(pthread_mutex_lock(&v->lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&v->lock), == 0);
         v->size = 0;
-        assert(pthread_mutex_unlock(&v->lock) == 0);
+        ASSERT_RET(pthread_mutex_unlock(&v->lock), == 0);
 }
 
@@ -126 +126 @@
 {
         size_t cur;
-        assert(pthread_mutex_lock(&v->lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&v->lock), == 0);
         for (cur = 0; cur < v->size; cur++) {
                 (*fn)(&v->elements[cur*v->element_size]);
         }
-        assert(pthread_mutex_unlock(&v->lock) == 0);
+        ASSERT_RET(pthread_mutex_unlock(&v->lock), == 0);
 }

lib/libtrace.h.in

@@ -115 +115 @@
 /** DAG driver version installed on the current system */
 #define DAG_DRIVER_V "@DAG_VERSION_NUM@"
+
+/**
+  * A version of assert that always runs the first argument even
+  * when not debugging, however only asserts the condition if debugging
+  * Intended for use mainly with pthread locks etc. which have error
+  * returns but *should* never actually fail.
+  */
+#ifdef NDEBUG
+#define ASSERT_RET(run, cond) run
+#else
+#define ASSERT_RET(run, cond) assert(run cond)
+#endif
     
 #ifdef __cplusplus 
@@ -3188 +3200 @@
 
 DLLEXPORT int trace_pstart(libtrace_t *libtrace, void* global_blob, fn_per_pkt per_pkt, fn_reducer reducer);
-DLLEXPORT int trace_pread_packet(libtrace_t *libtrace, libtrace_thread_t *t, libtrace_packet_t **packet);
 DLLEXPORT int trace_ppause(libtrace_t *libtrace);
 DLLEXPORT int trace_pstop(libtrace_t *libtrace);

lib/libtrace_int.h

@@ -149 +149 @@
 
 #include "data-struct/ring_buffer.h"
+#include "data-struct/object_cache.h"
 #include "data-struct/vector.h"
 #include "data-struct/message_queue.h"
@@ -317 +318 @@
         int packet_freelist_size;
         /** The actual freelist */
-        libtrace_ringbuffer_t packet_freelist;
+        libtrace_ocache_t packet_freelist;
         /** The number of packets that can queue per thread - XXX consider deadlocks with non malloc()'d packets that need to be released */
         int perpkt_buffer_size;

lib/trace.c

@@ -273 +273 @@
         libtrace->perpkt_buffer_size = 0;
         libtrace->expected_key = 0;
-        libtrace_zero_ringbuffer(&libtrace->packet_freelist);
+        libtrace_zero_ocache(&libtrace->packet_freelist);
         libtrace_zero_thread(&libtrace->hasher_thread);
         libtrace_zero_thread(&libtrace->reducer_thread);
@@ -395 +395 @@
         libtrace->packet_freelist_size = 0;
         libtrace->perpkt_buffer_size = 0;
-        libtrace_zero_ringbuffer(&libtrace->packet_freelist);
+        libtrace_zero_ocache(&libtrace->packet_freelist);
         libtrace_zero_thread(&libtrace->hasher_thread);
         libtrace_zero_thread(&libtrace->reducer_thread);
@@ -651 +651 @@
                 }
                 free(libtrace->first_packets.packets);
-                assert(pthread_spin_destroy(&libtrace->first_packets.lock) == 0);
+                ASSERT_RET(pthread_spin_destroy(&libtrace->first_packets.lock), == 0);
         }
 
@@ -666 +666 @@
         /* Empty any packet memory */
         if (libtrace->state != STATE_NEW) {
-                libtrace_packet_t * packet;
-                while (libtrace_ringbuffer_try_read(&libtrace->packet_freelist,(void **) &packet))
-                        trace_destroy_packet(packet);
-                
-                libtrace_ringbuffer_destroy(&libtrace->packet_freelist);
+                // This has all of our packets
+                libtrace_ocache_destroy(&libtrace->packet_freelist);
                 
                 for (i = 0; i < libtrace->perpkt_thread_count; ++i) {
@@ -721 +718 @@
 }
 
-DLLEXPORT libtrace_packet_t *trace_create_packet(void) 
-{
-        libtrace_packet_t *packet = 
+DLLEXPORT libtrace_packet_t *trace_create_packet(void)
+{
+        libtrace_packet_t *packet =
                 (libtrace_packet_t*)calloc((size_t)1,sizeof(libtrace_packet_t));
 
@@ -1453 +1450 @@
 #if HAVE_LLVM
         if (!filter->jitfilter) {
-                assert(pthread_mutex_lock(&mutex) == 0);
+                ASSERT_RET(pthread_mutex_lock(&mutex), == 0);
                 /* Again double check here like the bpf filter */
                 if(filter->jitfilter) 
@@ -1462 +1459 @@
                  * as such lock here anyways */
                         filter->jitfilter = compile_program(filter->filter.bf_insns, filter->filter.bf_len);
-                assert(pthread_mutex_unlock(&mutex) == 0);
+                ASSERT_RET(pthread_mutex_unlock(&mutex), == 0);
         }
 #endif

lib/trace_parallel.c

@@ -103 +103 @@
 #define VERBOSE_DEBBUGING 0
 
+
+static size_t trace_pread_packet(libtrace_t *libtrace, libtrace_thread_t *t, libtrace_packet_t *packets[], size_t nb_packets);
+
 extern int libtrace_parallel;
 
@@ -109 +112 @@
         uint64_t wait_for_fill_complete_hits;
 } contention_stats[1024];
+
+struct mem_stats {
+        struct memfail {
+           uint64_t cache_hit;
+           uint64_t ring_hit;
+           uint64_t miss;
+           uint64_t recycled;
+        } readbulk, read, write, writebulk;
+};
+
+// Grrr gcc wants this spelt out
+__thread struct mem_stats mem_hits = {{0},{0},{0},{0}};
+
+static void print_memory_stats() {
+        char t_name[50];
+        uint64_t total;
+        pthread_getname_np(pthread_self(), t_name, sizeof(t_name));
+
+        fprintf(stderr, "Thread ID#%d - %s\n", (int) pthread_self(), t_name);
+
+        total = mem_hits.read.cache_hit + mem_hits.read.ring_hit + mem_hits.read.miss;
+        if (total) {
+                fprintf(stderr, "\tRead:\n\t---CHits=%"PRIu64"\n\t---RHits=%"PRIu64"\n\t---Misses=%"PRIu64"\n\t---Recycled=%"PRIu64"\n",
+                                mem_hits.read.cache_hit, mem_hits.read.ring_hit, mem_hits.read.miss, mem_hits.read.recycled);
+                fprintf(stderr, "\t---Total=%"PRIu64"\n\t---Miss %%=%f\n",
+                                total, (double) mem_hits.read.miss / (double) total * 100.0);
+        }
+
+        total = mem_hits.readbulk.cache_hit + mem_hits.readbulk.ring_hit + mem_hits.readbulk.miss;
+        if (total) {
+                fprintf(stderr, "\tReadbulk:\n\t---CHits=%"PRIu64"\n\t---RHits=%"PRIu64"\n\t---Misses=%"PRIu64"\n\t---Recycled=%"PRIu64"\n",
+                                mem_hits.readbulk.cache_hit, mem_hits.readbulk.ring_hit, mem_hits.readbulk.miss, mem_hits.readbulk.recycled);
+
+
+                fprintf(stderr, "\t---Total=%"PRIu64"\n\t---Miss %%=%f\n",
+                                total, (double) mem_hits.readbulk.miss / (double) total * 100.0);
+        }
+
+        total = mem_hits.write.cache_hit + mem_hits.write.ring_hit + mem_hits.write.miss;
+        if (total) {
+                fprintf(stderr, "\tWrite:\n\t---CHits=%"PRIu64"\n\t---RHits=%"PRIu64"\n\t---Misses=%"PRIu64"\n\t---Recycled=%"PRIu64"\n",
+                                mem_hits.write.cache_hit, mem_hits.write.ring_hit, mem_hits.write.miss, mem_hits.write.recycled);
+
+                fprintf(stderr, "\t---Total=%"PRIu64"\n\t---Miss %%=%f\n",
+                                total, (double) mem_hits.write.miss / (double) total * 100.0);
+        }
+
+        total = mem_hits.writebulk.cache_hit + mem_hits.writebulk.ring_hit + mem_hits.writebulk.miss;
+        if (total) {
+                fprintf(stderr, "\tWritebulk:\n\t---CHits=%"PRIu64"\n\t---RHits=%"PRIu64"\n\t---Misses=%"PRIu64"\n\t---Recycled=%"PRIu64"\n",
+                                mem_hits.writebulk.cache_hit, mem_hits.writebulk.ring_hit, mem_hits.writebulk.miss, mem_hits.writebulk.recycled);
+
+                fprintf(stderr, "\t---Total=%"PRIu64"\n\t---Miss %%=%f\n",
+                                total, (double) mem_hits.writebulk.miss / (double) total * 100.0);
+        }
+
+}
 
 /**
@@ -165 +225 @@
         const enum trace_state new_state, const bool need_lock)
 {
-        enum trace_state prev_state;
+        UNUSED enum trace_state prev_state;
         if (need_lock)
                 pthread_mutex_lock(&trace->libtrace_lock);
@@ -293 +353 @@
 static void trace_thread_pause(libtrace_t *trace, libtrace_thread_t *t) {
         trace_make_results_packets_safe(trace);
-        assert(pthread_mutex_lock(&trace->libtrace_lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&trace->libtrace_lock), == 0);
         thread_change_state(trace, t, THREAD_PAUSED, false);
         while (trace->state == STATE_PAUSED || trace->state == STATE_PAUSING) {
-                assert(pthread_cond_wait(&trace->perpkt_cond, &trace->libtrace_lock) == 0);
+                ASSERT_RET(pthread_cond_wait(&trace->perpkt_cond, &trace->libtrace_lock), == 0);
         }
         thread_change_state(trace, t, THREAD_RUNNING, false);
-        assert(pthread_mutex_unlock(&trace->libtrace_lock) == 0);
-}
+        ASSERT_RET(pthread_mutex_unlock(&trace->libtrace_lock), == 0);
+}
+
+#define PACKETQUEUES 10
 
 /**
@@ -309 +371 @@
         libtrace_thread_t * t;
         libtrace_message_t message = {0};
-        libtrace_packet_t *packet = NULL;
-
+        libtrace_packet_t *packets[PACKETQUEUES] = {NULL};
+        size_t nb_packets;
+        size_t i;
 
         // Force this thread to wait until trace_pstart has been completed
-        assert(pthread_mutex_lock(&trace->libtrace_lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&trace->libtrace_lock), == 0);
         t = get_thread_table(trace);
         assert(t);
@@ -320 +383 @@
                 trace->format->pregister_thread(trace, t, !trace_has_dedicated_hasher(trace));
         }
-        assert(pthread_mutex_unlock(&trace->libtrace_lock) == 0);
+        ASSERT_RET(pthread_mutex_unlock(&trace->libtrace_lock), == 0);
 
         /* ~~~~~~~~~~~ Setup complete now we loop ~~~~~~~~~~~~~~~ */
@@ -333 +396 @@
 
         for (;;) {
-                int psize;
 
                 if (libtrace_message_queue_try_get(&t->messages, &message) != LIBTRACE_MQ_FAILED) {
@@ -346 +408 @@
                                                 // The hasher has stopped by this point, so the queue shouldn't be filling
                                                 while(!libtrace_ringbuffer_is_empty(&t->rbuffer)) {
-                                                        psize = trace_pread_packet(trace, t, &packet);
-                                                        if (psize > 0) {
-                                                                packet = (*trace->per_pkt)(trace, packet, NULL, t);
+                                                        nb_packets = trace_pread_packet(trace, t, packets, 1);
+                                                        if (nb_packets == 1) {
+                                                                if (packets[0]->error > 0)
+                                                                        packets[0] = (*trace->per_pkt)(trace, packets[0], NULL, t);
                                                         } else {
-                                                                fprintf(stderr, "Psize=%d empty=%d this is probably EOF or message waiting, but if this continues we have a bug!!!\n", psize, libtrace_ringbuffer_is_empty(&t->rbuffer));
+                                                                fprintf(stderr, "Psize=%d empty=%d this is probably EOF or message waiting, but if this continues we have a bug!!!\n", packets[0]->error, libtrace_ringbuffer_is_empty(&t->rbuffer));
                                                         }
                                                 }
@@ -369 +432 @@
 
                 if (trace->perpkt_thread_count == 1) {
-                        if (!packet) {
-                                if (!libtrace_ringbuffer_try_sread_bl(&trace->packet_freelist, (void **) &packet))
-                                        packet = trace_create_packet();
+                        if (!packets[0]) {
+                                libtrace_ocache_alloc(&trace->packet_freelist, (void **) &packets[0], 1, 1);
                         }
-                        assert(packet);
-                        if ((psize = trace_read_packet(trace, packet)) <1) {
-                                break;
+                        assert(packets[0]);
+                        packets[0]->error = trace_read_packet(trace, packets[0]);
+                        nb_packets = 1;
+                } else {
+                        nb_packets = trace_pread_packet(trace, t, packets, PACKETQUEUES);
+                }
+                // Loop through the packets we just read
+                for (i = 0; i < nb_packets; ++i) {
+                        
+                        if (packets[i]->error > 0) {
+                                packets[i] = (*trace->per_pkt)(trace, packets[i], NULL, t);
+                        } else if (packets[i]->error != -2) {
+                                // An error this should be the last packet we read
+                                size_t z;
+                                for (z = i ; z < nb_packets; ++z)
+                                        fprintf(stderr, "i=%d nb_packet=%d err=%d\n", (int) z, (int) nb_packets, packets[i]->error);
+                                assert (i == nb_packets-1);
+                                goto stop;
                         }
-                } else {
-                        psize = trace_pread_packet(trace, t, &packet);
-                }
-
-                if (psize > 0) {
-                        packet = (*trace->per_pkt)(trace, packet, NULL, t);
-                        continue;
-                }
-
-                if (psize == -2)
-                        continue; // We have a message
-
-                if (psize < 1) { // consider sending a message
-                        break;
-                }
-
+                        // -2 is a message its not worth checking now just finish this lot and we'll check
+                        // when we loop next
+                }
         }
 
@@ -404 +468 @@
         (*trace->per_pkt)(trace, NULL, &message, t);
 
-        // Free our last packet
-        if (packet)
-                trace_destroy_packet(packet);
+        // Free any remaining packets
+        for (i = 0; i < PACKETQUEUES; i++) {
+                if (packets[i]) {
+                        libtrace_ocache_free(&trace->packet_freelist, (void **) &packets[i], 1, 1);
+                        packets[i] = NULL;
+                }
+        }
 
         
@@ -416 +484 @@
         trace_send_message_to_reducer(trace, &message);
 
-        assert(pthread_mutex_lock(&trace->libtrace_lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&trace->libtrace_lock), == 0);
         if (trace->format->punregister_thread) {
                 trace->format->punregister_thread(trace, t);
         }
-        assert(pthread_mutex_unlock(&trace->libtrace_lock) == 0);
+        print_memory_stats();
+
+        ASSERT_RET(pthread_mutex_unlock(&trace->libtrace_lock), == 0);
 
         pthread_exit(NULL);
@@ -439 +509 @@
         assert(trace_has_dedicated_hasher(trace));
         /* Wait until all threads are started and objects are initialised (ring buffers) */
-        assert(pthread_mutex_lock(&trace->libtrace_lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&trace->libtrace_lock), == 0);
         t = &trace->hasher_thread;
         assert(t->type == THREAD_HASHER && pthread_equal(pthread_self(), t->tid));
@@ -446 +516 @@
                 trace->format->pregister_thread(trace, t, true);
         }
-        assert(pthread_mutex_unlock(&trace->libtrace_lock) == 0);
+        ASSERT_RET(pthread_mutex_unlock(&trace->libtrace_lock), == 0);
         int pkt_skipped = 0;
         /* Read all packets in then hash and queue against the correct thread */
         while (1) {
                 int thread;
-                if (!pkt_skipped && !libtrace_ringbuffer_try_sread_bl(&trace->packet_freelist, (void **) &packet))
-                        packet = trace_create_packet();
+                if (!pkt_skipped)
+                        libtrace_ocache_alloc(&trace->packet_freelist, (void **) &packet, 1, 1);
                 assert(packet);
 
@@ -462 +532 @@
                         switch(message.code) {
                                 case MESSAGE_DO_PAUSE:
-                                        assert(pthread_mutex_lock(&trace->libtrace_lock) == 0);
+                                        ASSERT_RET(pthread_mutex_lock(&trace->libtrace_lock), == 0);
                                         thread_change_state(trace, t, THREAD_PAUSED, false);
                                         pthread_cond_broadcast(&trace->perpkt_cond);
                                         while (trace->state == STATE_PAUSED || trace->state == STATE_PAUSING) {
-                                                assert(pthread_cond_wait(&trace->perpkt_cond, &trace->libtrace_lock) == 0);
+                                                ASSERT_RET(pthread_cond_wait(&trace->perpkt_cond, &trace->libtrace_lock), == 0);
                                         }
                                         thread_change_state(trace, t, THREAD_RUNNING, false);
                                         pthread_cond_broadcast(&trace->perpkt_cond);
-                                        assert(pthread_mutex_unlock(&trace->libtrace_lock) == 0);
+                                        ASSERT_RET(pthread_mutex_unlock(&trace->libtrace_lock), == 0);
                                         break;
                                 case MESSAGE_DO_STOP:
@@ -506 +576 @@
                         bcast = packet;
                 } else {
-                        bcast = trace_create_packet();
+                        libtrace_ocache_alloc(&trace->packet_freelist, (void **) &bcast, 1, 1);
                         bcast->error = packet->error;
                 }
-                assert(pthread_mutex_lock(&trace->libtrace_lock) == 0);
+                ASSERT_RET(pthread_mutex_lock(&trace->libtrace_lock), == 0);
                 if (trace->perpkt_threads[i].state != THREAD_FINISHED) {
-                        assert(pthread_mutex_unlock(&trace->libtrace_lock) == 0);
+                        ASSERT_RET(pthread_mutex_unlock(&trace->libtrace_lock), == 0);
                         // Unlock early otherwise we could deadlock
                         libtrace_ringbuffer_write(&trace->perpkt_threads[i].rbuffer, bcast);
                 } else {
-                        assert(pthread_mutex_unlock(&trace->libtrace_lock) == 0);
+                        ASSERT_RET(pthread_mutex_unlock(&trace->libtrace_lock), == 0);
                 }
         }
@@ -526 +596 @@
         message.additional.uint64 = 0;
         trace_send_message_to_reducer(trace, &message);
-        assert(pthread_mutex_lock(&trace->libtrace_lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&trace->libtrace_lock), == 0);
         if (trace->format->punregister_thread) {
                 trace->format->punregister_thread(trace, t);
         }
-        assert(pthread_mutex_unlock(&trace->libtrace_lock) == 0);
+        print_memory_stats();
+        ASSERT_RET(pthread_mutex_unlock(&trace->libtrace_lock), == 0);
 
         // TODO remove from TTABLE t sometime
@@ -553 +624 @@
 }
 
+/**
+ * @brief Move NULLs to the end of an array.
+ * @param values
+ * @param len
+ * @return The location the first NULL, aka the number of non NULL elements
+ */
+static inline size_t move_nulls_back(void *arr[], size_t len) {
+        size_t fr=0, en = len-1;
+        // Shift all non NULL elements to the front of the array, and NULLs to the
+        // end, traverses every element at most once
+        for (;fr < en; ++fr) {
+                if (arr[fr] == NULL) {
+                        for (;en > fr; --en) {
+                                if(arr[en]) {
+                                        arr[fr] = arr[en];
+                                        arr[en] = NULL;
+                                        break;
+                                }
+                        }
+                }
+        }
+        // This is the index of the first NULL
+        en = MIN(fr, en);
+        // Or the end of the array if this special case
+        if (arr[en])
+                en++;
+        return en;
+}
+
+/** returns the number of packets successfully allocated in the final array
+ these will all be at the front of the array */
+inline static size_t fill_array_with_empty_packets(libtrace_t *libtrace, libtrace_packet_t *packets[], size_t nb_packets) {
+        size_t nb;
+        nb = move_nulls_back((void **) packets, nb_packets);
+        mem_hits.read.recycled += nb;
+        nb += libtrace_ocache_alloc(&libtrace->packet_freelist, (void **) &packets[nb], nb_packets - nb, nb_packets - nb);
+        assert(nb_packets == nb);
+        return nb;
+}
+
+
+inline static size_t empty_array_of_packets(libtrace_t *libtrace, libtrace_packet_t *packets[], size_t nb_packets) {
+        size_t nb;
+        nb = move_nulls_back((void **) packets, nb_packets);
+        mem_hits.write.recycled += nb_packets - nb;
+        nb += nb_packets - libtrace_ocache_free(&libtrace->packet_freelist, (void **)packets, nb, nb);
+        memset(packets, 0, nb); // XXX make better, maybe do this in ocache??
+        return nb;
+}
+
 /* Our simplest case when a thread becomes ready it can obtain an exclusive
- * lock to read a packet from the underlying trace.
- */
-inline static int trace_pread_packet_first_in_first_served(libtrace_t *libtrace, libtrace_thread_t *t, libtrace_packet_t **packet)
-{
-        // We need this to fill the 'first' packet table
-        if (!*packet) {
-                if (!libtrace_ringbuffer_try_sread_bl(&libtrace->packet_freelist, (void **) packet))
-                        *packet = trace_create_packet();
-        }
-        assert(*packet);
-        assert(pthread_mutex_lock(&libtrace->libtrace_lock) == 0);
-        /* Read a packet */
-        (*packet)->error = trace_read_packet(libtrace, *packet);
-        // Doing this inside the lock ensures the first packet is always
-        // recorded first
-        if ((*packet)->error > 0)
-                store_first_packet(libtrace, *packet, t);
-
-        assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
-        return (*packet)->error;
+ * lock to read packets from the underlying trace.
+ */
+inline static size_t trace_pread_packet_first_in_first_served(libtrace_t *libtrace, libtrace_thread_t *t, libtrace_packet_t *packets[], size_t nb_packets)
+{
+        size_t i = 0;
+
+        nb_packets = fill_array_with_empty_packets(libtrace, packets, nb_packets);
+
+        ASSERT_RET(pthread_mutex_lock(&libtrace->libtrace_lock), == 0);
+        /* Read nb_packets */
+        for (i = 0; i < nb_packets; ++i) {
+                packets[i]->error = trace_read_packet(libtrace, packets[i]);
+                // Doing this inside the lock ensures the first packet is always
+                // recorded first
+                if (packets[i]->error <= 0) {
+                        ++i;
+                        break;
+                }
+        }
+        if (packets[0]->error > 0) {
+                store_first_packet(libtrace, packets[0], t);
+        }
+        ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
+        return i;
 }
 
@@ -581 +706 @@
  * 2. Move that into the packet provided (packet)
  */
-inline static int trace_pread_packet_hasher_thread(libtrace_t *libtrace, libtrace_thread_t *t, libtrace_packet_t **packet)
-{
-        if (*packet) // Recycle the old get the new
-                if (!libtrace_ringbuffer_try_swrite_bl(&libtrace->packet_freelist, (void *) *packet))
-                        trace_destroy_packet(*packet);
-        *packet = libtrace_ringbuffer_read(&t->rbuffer);
-
-        if (*packet) {
+inline static size_t trace_pread_packet_hasher_thread(libtrace_t *libtrace, libtrace_thread_t *t, libtrace_packet_t **packets, size_t nb_packets)
+{
+        size_t i;
+
+        // Always grab at least one
+        if (packets[0]) // Recycle the old get the new
+                libtrace_ocache_free(&libtrace->packet_freelist, (void **) packets, 1, 1);
+        packets[0] = libtrace_ringbuffer_read(&t->rbuffer);
+
+
+        if (packets[0] == NULL) {
+                libtrace_ocache_alloc(&libtrace->packet_freelist, (void **) packets, 1, 1);
+                packets[0]->error = -2;
+        }
+
+        if (packets[0]->error < 0)
+                return 1;
+
+        for (i = 1; i < nb_packets; i++) {
+                if (packets[i]) // Recycle the old get the new
+                        libtrace_ocache_free(&libtrace->packet_freelist, (void **) &packets[i], 1, 1);
+                if (!libtrace_ringbuffer_try_read(&t->rbuffer, (void **) &packets[i])) {
+                        packets[i] = NULL;
+                        break;
+                }
+                // Message wating
+                if (packets[i] == NULL) {
+                        libtrace_ocache_alloc(&libtrace->packet_freelist, (void **) &packets[i], 1, 1);
+                        packets[i]->error = -2;
+                        ++i;
+                        break;
+                }
+        }
+        
+        return i;
+        /*if (*packet) {
                 return (*packet)->error;
         } else {
@@ -597 +750 @@
                 fprintf(stderr, "Got a NULL packet from hasher therefore message waiting\n");
                 return -2;
-        }
+        }*/
 }
 
@@ -613 +766 @@
 
                 if (*packet) // Recycle the old get the new
-                        if (!libtrace_ringbuffer_try_swrite_bl(&libtrace->packet_freelist, (void *) *packet))
-                                trace_destroy_packet(*packet);
+                        libtrace_ocache_free(&libtrace->packet_freelist, (void **) packet, 1, 1);
                 *packet = retrived_packet;
                 *ret = (*packet)->error;
@@ -635 +787 @@
         do {
                 // Wait for a thread to end
-                assert(pthread_mutex_lock(&libtrace->libtrace_lock) == 0);
+                ASSERT_RET(pthread_mutex_lock(&libtrace->libtrace_lock), == 0);
 
                 // Check before
                 if (libtrace->perpkt_thread_states[THREAD_FINISHING] == libtrace->perpkt_thread_count) {
                         complete = true;
-                        assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
+                        ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
                         continue;
                 }
 
-                assert(pthread_cond_wait(&libtrace->perpkt_cond, &libtrace->libtrace_lock) == 0);
+                ASSERT_RET(pthread_cond_wait(&libtrace->perpkt_cond, &libtrace->libtrace_lock), == 0);
 
                 // Check after
                 if (libtrace->perpkt_thread_states[THREAD_FINISHING] == libtrace->perpkt_thread_count) {
                         complete = true;
-                        assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
+                        ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
                         continue;
                 }
 
-                assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
+                ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
 
                 // Always trying to keep our buffer empty for the unlikely case more threads than buffer space want to write into our queue
@@ -696 +848 @@
                         return ret;
                 // Can still block here if another thread is writing to a full queue
-                assert(pthread_mutex_lock(&libtrace->libtrace_lock) == 0);
+                ASSERT_RET(pthread_mutex_lock(&libtrace->libtrace_lock), == 0);
 
                 // Its impossible for our own queue to overfill, because no one can write
                 // when we are in the lock
                 if(try_waiting_queue(libtrace, t, packet, &ret)) {
-                        assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
+                        ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
                         return ret;
                 }
@@ -708 +860 @@
                 if (libtrace->perpkt_queue_full) {
                         contention_stats[t->perpkt_num].wait_for_fill_complete_hits++;
-                        assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
+                        ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
                         continue;
                 }
 
-                if (!*packet) {
-                        if (!libtrace_ringbuffer_try_sread_bl(&libtrace->packet_freelist, (void **) packet))
-                                *packet = trace_create_packet();
-                }
+                if (!*packet)
+                        libtrace_ocache_alloc(&libtrace->packet_freelist, (void **) packet, 1, 1);
                 assert(*packet);
 
                 // If we fail here we can guarantee that our queue is empty (and no new data will be added because we hold the lock)
                 if (libtrace_halt || ((*packet)->error = trace_read_packet(libtrace, *packet)) <1 /*&& psize != LIBTRACE_MESSAGE_WAITING*/) {
-                        assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
+                        ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
                         if (libtrace_halt)
                                 return 0;
@@ -731 +881 @@
                 if (thread == t->perpkt_num) {
                         // If it's this thread we must be in order because we checked the buffer once we got the lock
-                        assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
+                        ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
                         return (*packet)->error;
                 }
@@ -738 +888 @@
                         while (!libtrace_ringbuffer_try_swrite_bl(&libtrace->perpkt_threads[thread].rbuffer, *packet)) {
                                 libtrace->perpkt_queue_full = true;
-                                assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
+                                ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
                                 contention_stats[t->perpkt_num].full_queue_hits++;
-                                assert(pthread_mutex_lock(&libtrace->libtrace_lock) == 0);
+                                ASSERT_RET(pthread_mutex_lock(&libtrace->libtrace_lock), == 0);
                         }
                         *packet = NULL;
@@ -748 +898 @@
                         assert (!"packet_hash_locked() The user terminated the trace in a abnormal manner");
                 }
-                assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
+                ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
         }
 }
@@ -779 +929 @@
                 // We limit the number of packets we get to the size of the sliding window
                 // such that it is impossible for any given thread to fail to store a packet
-                assert(sem_wait(&libtrace->sem) == 0);
+                ASSERT_RET(sem_wait(&libtrace->sem), == 0);
                 /*~~~~Single threaded read of a packet~~~~*/
-                assert(pthread_mutex_lock(&libtrace->libtrace_lock) == 0);
+                ASSERT_RET(pthread_mutex_lock(&libtrace->libtrace_lock), == 0);
 
                 /* Re-check our queue things we might have data waiting */
                 if(try_waiting_queue(libtrace, t, packet, &ret)) {
-                        assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
-                        assert(sem_post(&libtrace->sem) == 0);
+                        ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
+                        ASSERT_RET(sem_post(&libtrace->sem), == 0);
                         return ret;
                 }
@@ -792 +942 @@
                 // TODO put on *proper* condition variable
                 if (libtrace->perpkt_queue_full) {
-                        assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
-                        assert(sem_post(&libtrace->sem) == 0);
+                        ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
+                        ASSERT_RET(sem_post(&libtrace->sem), == 0);
                         contention_stats[t->perpkt_num].wait_for_fill_complete_hits++;
                         continue;
                 }
 
-                if (!*packet) {
-                        if (!libtrace_ringbuffer_try_sread_bl(&libtrace->packet_freelist, (void **) packet))
-                                *packet = trace_create_packet();
-                }
+                if (!*packet)
+                        libtrace_ocache_alloc(&libtrace->packet_freelist, (void **) packet, 1, 1);
                 assert(*packet);
 
                 if (libtrace_halt || ((*packet)->error = trace_read_packet(libtrace, *packet)) <1 /*&& psize != LIBTRACE_MESSAGE_WAITING*/) {
-                        assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
-                        assert(sem_post(&libtrace->sem) == 0);
+                        ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
+                        ASSERT_RET(sem_post(&libtrace->sem), == 0);
                         // Finish this thread ensuring that any data written later by another thread is retrieved also
                         if (libtrace_halt)
@@ -813 +961 @@
                                 return trace_finish_perpkt(libtrace, packet, t);
                 }
-                assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
+                ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
 
                 /* ~~~~Multiple threads can run the hasher~~~~ */
@@ -819 +967 @@
 
                 /* Yes this is correct opposite read lock for a write operation */
-                assert(pthread_rwlock_rdlock(&libtrace->window_lock) == 0);
+                ASSERT_RET(pthread_rwlock_rdlock(&libtrace->window_lock), == 0);
                 if (!libtrace_slidingwindow_try_write(&libtrace->sliding_window, trace_packet_get_order(*packet), *packet))
                         assert(!"Semaphore should stop us from ever overfilling the sliding window");
-                assert(pthread_rwlock_unlock(&libtrace->window_lock) == 0);
+                ASSERT_RET(pthread_rwlock_unlock(&libtrace->window_lock), == 0);
                 *packet = NULL;
 
                 // Always try read any data from the sliding window
                 while (libtrace_slidingwindow_read_ready(&libtrace->sliding_window)) {
-                        assert(pthread_rwlock_wrlock(&libtrace->window_lock) == 0);
+                        ASSERT_RET(pthread_rwlock_wrlock(&libtrace->window_lock), == 0);
                         if (libtrace->perpkt_queue_full) {
                                 // I might be the holdup in which case if I can read my queue I should do that and return
                                 if(try_waiting_queue(libtrace, t, packet, &ret)) {
-                                        assert(pthread_rwlock_unlock(&libtrace->window_lock) == 0);
+                                        ASSERT_RET(pthread_rwlock_unlock(&libtrace->window_lock), == 0);
                                         return ret;
                                 }
-                                assert(pthread_rwlock_unlock(&libtrace->window_lock) == 0);
+                                ASSERT_RET(pthread_rwlock_unlock(&libtrace->window_lock), == 0);
                                 continue;
                         }
@@ -851 +999 @@
                                                                 // We must be able to write this now 100% without fail
                                                                 libtrace_ringbuffer_write(&libtrace->perpkt_threads[thread].rbuffer, *packet);
-                                                                assert(sem_post(&libtrace->sem) == 0);
-                                                                assert(pthread_rwlock_unlock(&libtrace->window_lock) == 0);
+                                                                ASSERT_RET(sem_post(&libtrace->sem), == 0);
+                                                                ASSERT_RET(pthread_rwlock_unlock(&libtrace->window_lock), == 0);
                                                                 return ret;
                                                         } else {
@@ -860 +1008 @@
                                                 // Not us we have to give the other threads a chance to write there packets then
                                                 libtrace->perpkt_queue_full = true;
-                                                assert(pthread_rwlock_unlock(&libtrace->window_lock) == 0);
+                                                ASSERT_RET(pthread_rwlock_unlock(&libtrace->window_lock), == 0);
                                                 for (i = 0; i < libtrace->perpkt_thread_count-1; i++) // Release all other threads to read there packets
-                                                        assert(sem_post(&libtrace->sem) == 0);
+                                                        ASSERT_RET(sem_post(&libtrace->sem), == 0);
 
                                                 contention_stats[t->perpkt_num].full_queue_hits++;
-                                                assert(pthread_rwlock_wrlock(&libtrace->window_lock) == 0);
+                                                ASSERT_RET(pthread_rwlock_wrlock(&libtrace->window_lock), == 0);
                                                 // Grab these back
                                                 for (i = 0; i < libtrace->perpkt_thread_count-1; i++) // Release all other threads to read there packets
-                                                        assert(sem_wait(&libtrace->sem) == 0);
+                                                        ASSERT_RET(sem_wait(&libtrace->sem), == 0);
                                                 libtrace->perpkt_queue_full = false;
                                         }
-                                        assert(sem_post(&libtrace->sem) == 0);
+                                        ASSERT_RET(sem_post(&libtrace->sem), == 0);
                                         *packet = NULL;
                                 } else {
@@ -879 +1027 @@
                                 }
                         }
-                        assert(pthread_rwlock_unlock(&libtrace->window_lock) == 0);
+                        ASSERT_RET(pthread_rwlock_unlock(&libtrace->window_lock), == 0);
                 }
                 // Now we go back to checking our queue anyways
@@ -902 +1050 @@
                 gettimeofday(&tv, NULL);
                 dup = trace_copy_packet(packet);
-                assert(pthread_spin_lock(&libtrace->first_packets.lock) == 0);
+                ASSERT_RET(pthread_spin_lock(&libtrace->first_packets.lock), == 0);
                 libtrace->first_packets.packets[t->perpkt_num].packet = dup;
                 //printf("Stored first packet time=%f\n", trace_get_seconds(dup));
@@ -918 +1066 @@
                                 libtrace->first_packets.first = t->perpkt_num;
                 }
-                assert(pthread_spin_unlock(&libtrace->first_packets.lock) == 0);
+                ASSERT_RET(pthread_spin_unlock(&libtrace->first_packets.lock), == 0);
                 libtrace_message_t mesg = {0};
                 mesg.code = MESSAGE_FIRST_PACKET;
@@ -936 +1084 @@
 {
         int ret = 0;
-        assert(pthread_spin_lock(&libtrace->first_packets.lock) == 0);
+        ASSERT_RET(pthread_spin_lock(&libtrace->first_packets.lock), == 0);
         if (libtrace->first_packets.count) {
                 *packet = libtrace->first_packets.packets[libtrace->first_packets.first].packet;
@@ -956 +1104 @@
                 *tv = NULL;
         }
-        assert(pthread_spin_unlock(&libtrace->first_packets.lock) == 0);
+        ASSERT_RET(pthread_spin_unlock(&libtrace->first_packets.lock), == 0);
         return ret;
 }
@@ -1110 +1258 @@
 
 /**
- * Read a packet from the parallel trace
- */
-DLLEXPORT int trace_pread_packet(libtrace_t *libtrace, libtrace_thread_t *t, libtrace_packet_t **packet)
-{
-        int ret;
-
-        // Cleanup the packet passed back
-        if (*packet)
-                trace_fin_packet(*packet);
+ * Read packets from the parallel trace
+ * @return the number of packets read, null packets indicate messages. Check packet->error before
+ * assuming a packet is valid.
+ */
+static size_t trace_pread_packet(libtrace_t *libtrace, libtrace_thread_t *t, libtrace_packet_t *packets[], size_t nb_packets)
+{
+        size_t ret;
+        size_t i;
+        assert(nb_packets);
+
+        for (i = 0; i < nb_packets; i++) {
+                // Cleanup the packet passed back
+                if (packets[i])
+                        trace_fin_packet(packets[i]);
+        }
 
         if (trace_supports_parallel(libtrace) && !trace_has_dedicated_hasher(libtrace)) {
-                if (!*packet)
-                        *packet = trace_create_packet();
-                ret = trace_pread_packet_wrapper(libtrace, t, *packet);
+                if (!packets[0])
+                        libtrace_ocache_alloc(&libtrace->packet_freelist, (void **)packets, 1, 1);
+                packets[0]->error = trace_pread_packet_wrapper(libtrace, t, *packets);
+                ret = 1;
         } else if (trace_has_dedicated_hasher(libtrace)) {
-                ret = trace_pread_packet_hasher_thread(libtrace, t, packet);
+                ret = trace_pread_packet_hasher_thread(libtrace, t, packets, nb_packets);
         } else if (!trace_has_dedicated_hasher(libtrace)) {
                 /* We don't care about which core a packet goes to */
-                ret = trace_pread_packet_first_in_first_served(libtrace, t, packet);
+                ret = trace_pread_packet_first_in_first_served(libtrace, t, packets, nb_packets);
         } /* else {
                 ret = trace_pread_packet_hash_locked(libtrace, packet);
@@ -1135 +1290 @@
         // Formats can also optionally do this internally to ensure the first
         // packet is always reported correctly
-        if (ret > 0) {
-                store_first_packet(libtrace, *packet, t);
+        assert(ret);
+        assert(ret <= nb_packets);
+        if (packets[0]->error > 0) {
+                store_first_packet(libtrace, packets[0], t);
                 if (libtrace->tracetime)
-                        delay_tracetime(libtrace, *packet, t);
+                        delay_tracetime(libtrace, packets[0], t);
         }
 
@@ -1149 +1306 @@
 static inline int trace_start_perpkt_threads (libtrace_t *libtrace) {
         int i;
-
+        char name[16];
         for (i = 0; i < libtrace->perpkt_thread_count; i++) {
                 libtrace_thread_t *t = &libtrace->perpkt_threads[i];
-                assert(pthread_create(&t->tid, NULL, perpkt_threads_entry, (void *) libtrace) == 0);
+                ASSERT_RET(pthread_create(&t->tid, NULL, perpkt_threads_entry, (void *) libtrace), == 0);
+                snprintf(name, 16, "perpkt-%d", i);
+                pthread_setname_np(t->tid, name);
         }
         return libtrace->perpkt_thread_count;
@@ -1168 +1327 @@
 {
         int i;
+        char name[16];
         sigset_t sig_before, sig_block_all;
         assert(libtrace);
@@ -1177 +1337 @@
         if (libtrace->state != STATE_NEW) {
                 int err = 0;
-                assert(pthread_mutex_lock(&libtrace->libtrace_lock) == 0);
+                ASSERT_RET(pthread_mutex_lock(&libtrace->libtrace_lock), == 0);
                 if (libtrace->state != STATE_PAUSED) {
                         trace_set_err(libtrace, TRACE_ERR_BAD_STATE,
                                 "The trace(%s) has already been started and is not paused!!", libtrace->uridata);
-                        assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
+                        ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
                         return -1;
                 }
@@ -1207 +1367 @@
                         libtrace_change_state(libtrace, STATE_RUNNING, false);
                 }
-                assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
+                ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
                 return err;
         }
@@ -1219 +1379 @@
         libtrace->reducer = reducer;
 
-        assert(pthread_mutex_init(&libtrace->libtrace_lock, NULL) == 0);
-        assert(pthread_cond_init(&libtrace->perpkt_cond, NULL) == 0);
-        assert(pthread_rwlock_init(&libtrace->window_lock, NULL) == 0);
+        ASSERT_RET(pthread_mutex_init(&libtrace->libtrace_lock, NULL), == 0);
+        ASSERT_RET(pthread_cond_init(&libtrace->perpkt_cond, NULL), == 0);
+        ASSERT_RET(pthread_rwlock_init(&libtrace->window_lock, NULL), == 0);
         // Grab the lock
-        assert(pthread_mutex_lock(&libtrace->libtrace_lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&libtrace->libtrace_lock), == 0);
 
         // Set default buffer sizes
@@ -1250 +1410 @@
         sigemptyset(&sig_block_all);
 
-        assert(pthread_sigmask(SIG_SETMASK, &sig_block_all, &sig_before) == 0);
+        ASSERT_RET(pthread_sigmask(SIG_SETMASK, &sig_block_all, &sig_before), == 0);
 
         // If we are using a hasher start it
@@ -1261 +1421 @@
                 t->state = THREAD_RUNNING;
                 libtrace_message_queue_init(&t->messages, sizeof(libtrace_message_t));
-                assert(pthread_create(&t->tid, NULL, hasher_start, (void *) libtrace) == 0);
+                ASSERT_RET(pthread_create(&t->tid, NULL, hasher_start, (void *) libtrace), == 0);
+                snprintf(name, sizeof(name), "hasher-thread");
+                pthread_setname_np(t->tid, name);
         } else {
                 libtrace->hasher_thread.type = THREAD_EMPTY;
         }
-        libtrace_ringbuffer_init(&libtrace->packet_freelist, libtrace->packet_freelist_size, LIBTRACE_RINGBUFFER_BLOCKING);
+        //libtrace_ocache_init(&libtrace->packet_freelist, trace_create_packet, trace_destroy_packet, 64, libtrace->packet_freelist_size * 4, true);
+        libtrace_ocache_init(&libtrace->packet_freelist,
+                                                 (void* (*)()) trace_create_packet,
+                                                 (void (*)(void *))trace_destroy_packet,
+                                                 64,
+                                                 libtrace->packet_freelist_size * 4,
+                                                 true);
         //libtrace_slidingwindow_init(&libtrace->sliding_window, libtrace->packet_freelist_size, 0);
-        assert(sem_init(&libtrace->sem, 0, libtrace->packet_freelist_size) == 0);
+        ASSERT_RET(sem_init(&libtrace->sem, 0, libtrace->packet_freelist_size), == 0);
         // This will be applied to every new thread that starts, i.e. they will block all signals
         // Lets start a fixed number of reading threads
@@ -1281 +1449 @@
         libtrace->first_packets.first = 0;
         libtrace->first_packets.count = 0;
-        assert(pthread_spin_init(&libtrace->first_packets.lock, 0) == 0);
+        ASSERT_RET(pthread_spin_init(&libtrace->first_packets.lock, 0), == 0);
         libtrace->first_packets.packets = calloc(libtrace->perpkt_thread_count, sizeof(struct  __packet_storage_magic_type));
 
@@ -1297 +1465 @@
                 t->perpkt_num = i;
                 if (libtrace->hasher)
-                        libtrace_ringbuffer_init(&t->rbuffer, libtrace->perpkt_buffer_size, LIBTRACE_RINGBUFFER_BLOCKING);
+                        libtrace_ringbuffer_init(&t->rbuffer, libtrace->perpkt_buffer_size, LIBTRACE_RINGBUFFER_POLLING);
                 // Depending on the mode vector or deque might be chosen
                 libtrace_vector_init(&t->vector, sizeof(libtrace_result_t));
@@ -1305 +1473 @@
                 t->tmp_data = NULL;
                 t->recorded_first = false;
-                assert(pthread_spin_init(&t->tmp_spinlock, 0) == 0);
+                ASSERT_RET(pthread_spin_init(&t->tmp_spinlock, 0), == 0);
                 t->tracetime_offset_usec = 0;;
         }
@@ -1326 +1494 @@
                 libtrace->keepalive_thread.state = THREAD_RUNNING;
                 libtrace_message_queue_init(&libtrace->keepalive_thread.messages, sizeof(libtrace_message_t));
-                assert(pthread_create(&libtrace->keepalive_thread.tid, NULL, keepalive_entry, (void *) libtrace) == 0);
+                ASSERT_RET(pthread_create(&libtrace->keepalive_thread.tid, NULL, keepalive_entry, (void *) libtrace), == 0);
         }
 
@@ -1335 +1503 @@
 
         // Revert back - Allow signals again
-        assert(pthread_sigmask(SIG_SETMASK, &sig_before, NULL) == 0);
-        assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
+        ASSERT_RET(pthread_sigmask(SIG_SETMASK, &sig_before, NULL), == 0);
+        ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
 
         if (threads_started < 0)
@@ -1368 +1536 @@
         t = get_thread_table(libtrace);
         // Check state from within the lock if we are going to change it
-        assert(pthread_mutex_lock(&libtrace->libtrace_lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&libtrace->libtrace_lock), == 0);
         if (!libtrace->started || libtrace->state != STATE_RUNNING) {
                 fprintf(stderr, "pause failed started=%d state=%s (%d)\n", libtrace->started, get_trace_state_name(libtrace->state), libtrace->state);
                 trace_set_err(libtrace,TRACE_ERR_BAD_STATE, "You must call trace_start() before calling trace_ppause()");
-                assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
+                ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
                 return -1;
         }
 
         libtrace_change_state(libtrace, STATE_PAUSING, false);
-        assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
+        ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
 
         // Special case handle the hasher thread case
@@ -1386 +1554 @@
                 trace_send_message_to_thread(libtrace, &libtrace->hasher_thread, &message);
                 // Wait for it to pause
-                assert(pthread_mutex_lock(&libtrace->libtrace_lock) == 0);
+                ASSERT_RET(pthread_mutex_lock(&libtrace->libtrace_lock), == 0);
                 while (libtrace->hasher_thread.state == THREAD_RUNNING) {
-                        assert(pthread_cond_wait(&libtrace->perpkt_cond, &libtrace->libtrace_lock) == 0);
-                }
-                assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
+                        ASSERT_RET(pthread_cond_wait(&libtrace->perpkt_cond, &libtrace->libtrace_lock), == 0);
+                }
+                ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
         }
 
@@ -1425 +1593 @@
 
         // Wait for all threads to pause
-        assert(pthread_mutex_lock(&libtrace->libtrace_lock) == 0);
+        ASSERT_RET(pthread_mutex_lock(&libtrace->libtrace_lock), == 0);
         while(libtrace->perpkt_thread_states[THREAD_RUNNING]) {
-                assert(pthread_cond_wait(&libtrace->perpkt_cond, &libtrace->libtrace_lock) == 0);
-        }
-        assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
+                ASSERT_RET(pthread_cond_wait(&libtrace->perpkt_cond, &libtrace->libtrace_lock), == 0);
+        }
+        ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
 
         fprintf(stderr, "Threads have paused\n");
@@ -1559 +1727 @@
         for (i=0; i< libtrace->perpkt_thread_count; i++) {
                 //printf("Waiting to join with perpkt #%d\n", i);
-                assert(pthread_join(libtrace->perpkt_threads[i].tid, NULL) == 0);
+                ASSERT_RET(pthread_join(libtrace->perpkt_threads[i].tid, NULL), == 0);
                 //printf("Joined with perpkt #%d\n", i);
                 // So we must do our best effort to empty the queue - so
@@ -1607 +1775 @@
         
         libtrace_change_state(libtrace, STATE_JOINED, true);
+        print_memory_stats();
 }
 
@@ -1977 +2146 @@
 DLLEXPORT libtrace_packet_t* trace_result_packet(libtrace_t * libtrace, libtrace_packet_t * packet) {
         libtrace_packet_t* result;
-        if (!libtrace_ringbuffer_try_sread_bl(&libtrace->packet_freelist, (void **) &result))
-                result = trace_create_packet();
+        libtrace_ocache_alloc(&libtrace->packet_freelist, (void **) &result, 1, 1);
         assert(result);
         swap_packets(result, packet); // Move the current packet into our copy
@@ -1984 +2152 @@
 }
 
-DLLEXPORT void trace_free_result_packet(libtrace_t * libtrace, libtrace_packet_t * packet) {
+DLLEXPORT void trace_free_result_packet(libtrace_t *libtrace, libtrace_packet_t *packet) {
         // Try write back the packet
         assert(packet);
         // Always release any resources this might be holding such as a slot in a ringbuffer
         trace_fin_packet(packet);
-        if (!libtrace_ringbuffer_try_swrite_bl(&libtrace->packet_freelist, packet)) {
-                /* We couldn't, oh well lets just destroy it - XXX consider non managed formats i.e. rings buffers loosing packets and jamming up :( */
-                //assert(1 == 90);
-                trace_destroy_packet(packet);
-        }
+        libtrace_ocache_free(&libtrace->packet_freelist, (void **) &packet, 1, 1);
 }
 

test/test-datastruct-ringbuffer.c

@@ -26 +26 @@
 }
 
+static void * producer_bulk(void * a) {
+        libtrace_ringbuffer_t * rb = (libtrace_ringbuffer_t *) a;
+        char * i;
+        for (i = NULL; i < TEST_SIZE; i++) {
+                assert(libtrace_ringbuffer_write_bulk(rb, (void **) &i, 1, 1) == 1);
+        }
+        return 0;
+}
+
+static void * consumer_bulk(void * a) {
+        libtrace_ringbuffer_t * rb = (libtrace_ringbuffer_t *) a;
+        char *i;
+        void *value;
+        for (i = NULL; i < TEST_SIZE; i++) {
+                assert (libtrace_ringbuffer_read_bulk(rb, &value, 1, 1) == 1);
+                assert(value == i);
+        }
+        return 0;
+}
+
+
 /**
  * Tests the ringbuffer data structure, first this establishes that single
@@ -39 +60 @@
 
         libtrace_ringbuffer_init(&rb_block, (size_t) RINGBUFFER_SIZE, LIBTRACE_RINGBUFFER_BLOCKING);
-        libtrace_ringbuffer_init(&rb_polling, (size_t) RINGBUFFER_SIZE, LIBTRACE_RINGBUFFER_BLOCKING);
+        libtrace_ringbuffer_init(&rb_polling, (size_t) RINGBUFFER_SIZE, LIBTRACE_RINGBUFFER_POLLING);
         assert(libtrace_ringbuffer_is_empty(&rb_block));
         assert(libtrace_ringbuffer_is_empty(&rb_polling));
@@ -105 +126 @@
         assert(libtrace_ringbuffer_is_empty(&rb_polling));
 
+        pthread_create(&t[0], NULL, &producer_bulk, (void *) &rb_block);
+        pthread_create(&t[1], NULL, &consumer_bulk, (void *) &rb_block);
+        pthread_join(t[0], NULL);
+        pthread_join(t[1], NULL);
+        assert(libtrace_ringbuffer_is_empty(&rb_block));
+
+        pthread_create(&t[0], NULL, &producer_bulk, (void *) &rb_polling);
+        pthread_create(&t[1], NULL, &consumer_bulk, (void *) &rb_polling);
+        pthread_join(t[0], NULL);
+        pthread_join(t[1], NULL);
+        assert(libtrace_ringbuffer_is_empty(&rb_polling));
+
         return 0;
 }