Changes in / [92d5f15:4631115]

Files:

• README (modified) (1 diff)
• configure.in (modified) (3 diffs)
• lib/Makefile.am (modified) (4 diffs)
• lib/combiner_ordered.c (added)
• lib/combiner_sorted.c (added)
• lib/combiner_unordered.c (added)
• lib/combiners.h (added)
• lib/data-struct/deque.c (added)
• lib/data-struct/deque.h (added)
• lib/data-struct/linked_list.c (added)
• lib/data-struct/linked_list.h (added)
• lib/data-struct/message_queue.c (added)
• lib/data-struct/message_queue.h (added)
• lib/data-struct/object_cache.c (added)
• lib/data-struct/object_cache.h (added)
• lib/data-struct/ring_buffer.c (added)
• lib/data-struct/ring_buffer.h (added)
• lib/data-struct/sliding_window.c (added)
• lib/data-struct/sliding_window.h (added)
• lib/data-struct/vector.c (added)
• lib/data-struct/vector.h (added)
• lib/format_atmhdr.c (modified) (1 diff)
• lib/format_bpf.c (modified) (2 diffs)
• lib/format_dag24.c (modified) (1 diff)
• lib/format_dag25.c (modified) (54 diffs)
• lib/format_dpdk.c (modified) (67 diffs)
• lib/format_duck.c (modified) (1 diff)
• lib/format_erf.c (modified) (2 diffs)
• lib/format_legacy.c (modified) (4 diffs)
• lib/format_linux.c (deleted)
• lib/format_linux_common.c (added)
• lib/format_linux_common.h (added)
• lib/format_linux_int.c (added)
• lib/format_linux_ring.c (added)
• lib/format_pcap.c (modified) (2 diffs)
• lib/format_pcapfile.c (modified) (1 diff)
• lib/format_rt.c (modified) (2 diffs)
• lib/format_tsh.c (modified) (2 diffs)
• lib/hash_toeplitz.c (added)
• lib/hash_toeplitz.h (added)
• lib/libtrace.h.in (modified) (5 diffs)
• lib/libtrace_int.h (modified) (8 diffs)
• lib/trace.c (modified) (28 diffs)
• lib/trace_parallel.c (added)
• test/Makefile (modified) (3 diffs)
• test/compare-speeds.sh (added)
• test/do-tests-datastruct.sh (added)
• test/do-tests-parallel.sh (added)
• test/test-datastruct-deque.c (added)
• test/test-datastruct-ringbuffer.c (added)
• test/test-datastruct-vector.c (added)
• test/test-format-parallel-hasher.c (added)
• test/test-format-parallel-reporter.c (added)
• test/test-format-parallel-singlethreaded-hasher.c (added)
• test/test-format-parallel-singlethreaded.c (added)
• test/test-format-parallel-stressthreads.c (added)
• test/test-format-parallel.c (added)
• tools/traceanon/Makefile.am (modified) (3 diffs)
• tools/traceanon/ipenc.c (modified) (2 diffs)
• tools/traceanon/panon.c (modified) (2 diffs)
• tools/traceanon/rijndael.c (modified) (1 diff)
• tools/traceanon/traceanon_parallel.c (added)
• tools/tracertstats/Makefile.am (modified) (2 diffs)
• tools/tracertstats/tracertstats_parallel.c (added)
• tools/tracestats/Makefile.am (modified) (2 diffs)
• tools/tracestats/tracestats_parallel.c (added)

README

@@ +1 @@
+This fork of Libtrace aims to support parallel packet processing.
+
+This is still work in progress and is full of bugs, some of the original
+Libtrace functions might not function correctly breaking the supplied tools.
+
 libtrace 3.0.22
 

configure.in

@@ -392 +392 @@
 fi
 
+# If we use DPDK we might be able to use libnuma
+AC_CHECK_LIB(numa, numa_node_to_cpus, have_numa=1, have_numa=0)
+
 # Checks for various "optional" libraries
 AC_CHECK_LIB(pthread, pthread_create, have_pthread=1, have_pthread=0)
@@ -411 +414 @@
 AC_CHECK_LIB(rt, clock_gettime, have_clock_gettime=1, have_clock_gettime=0)
 LIBS=
+
+if test "$have_numa" = 1; then
+        LIBTRACE_LIBS="$LIBTRACE_LIBS -lnuma"
+        AC_DEFINE(HAVE_LIBNUMA, 1, [Set to 1 if libnuma is found supported])
+        with_numa=yes
+else
+        AC_DEFINE(HAVE_LIBNUMA, 0, [Set to 1 if libnuma is found supported])
+        with_numa=no
+fi
 
 if test "$dlfound" = 0; then
@@ -688 +700 @@
 if test x"$libtrace_dpdk" = xtrue; then
     AC_MSG_NOTICE([Compiled with DPDK live capture support: Yes])
+    reportopt "Compiled with DPDK trace NUMA support" $with_numa
 elif test x"$want_dpdk" != "xno"; then
 #   We don't officially support DPDK so only report failure if the user

lib/Makefile.am

@@ -2 +2 @@
 include_HEADERS = libtrace.h dagformat.h lt_inttypes.h daglegacy.h rt_protocol.h erftypes.h  
 
-AM_CFLAGS=@LIBCFLAGS@ @CFLAG_VISIBILITY@
-AM_CXXFLAGS=@LIBCXXFLAGS@ @CFLAG_VISIBILITY@
+AM_CFLAGS=@LIBCFLAGS@ @CFLAG_VISIBILITY@ -pthread
+AM_CXXFLAGS=@LIBCXXFLAGS@ @CFLAG_VISIBILITY@ -pthread
 
 extra_DIST = format_template.c
-NATIVEFORMATS=format_linux.c
+NATIVEFORMATS=format_linux_common.c format_linux_ring.c format_linux_int.c format_linux_common.h
 BPFFORMATS=format_bpf.c
 
@@ -29 +29 @@
 NATIVEFORMATS+= format_dpdk.c
 # So we also make libtrace.mk in dpdk otherwise automake tries to expand
-# it to early which I cannot seem to stop unless we use a path that
+# it too early which I cannot seem to stop unless we use a path that
 # doesn't exist currently
 export RTE_SDK=@RTE_SDK@
@@ -44 +44 @@
 endif
 
-libtrace_la_SOURCES = trace.c common.h \
+libtrace_la_SOURCES = trace.c trace_parallel.c common.h \
                 format_erf.c format_pcap.c format_legacy.c \
                 format_rt.c format_helper.c format_helper.h format_pcapfile.c \
@@ -57 +57 @@
                 $(DAGSOURCE) format_erf.h \
                 $(BPFJITSOURCE) \
-                libtrace_arphrd.h
+                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/object_cache.c \
+                data-struct/linked_list.c hash_toeplitz.c combiner_ordered.c \
+                combiner_sorted.c combiner_unordered.c
 
 if DAG2_4

lib/format_atmhdr.c

@@ -231 +231 @@
         trace_event_trace,              /* trace_event */
         NULL,                           /* help */
-        NULL                            /* next pointer */
+        NULL,                            /* next pointer */
+        NON_PARALLEL(false)
 };
         

lib/format_bpf.c

@@ -614 +614 @@
         trace_event_device,     /* trace_event */
         bpf_help,               /* help */
-        NULL
+        NULL,                   /* next pointer */
+        NON_PARALLEL(true)
 };
 #else   /* HAVE_DECL_BIOCSETIF */
@@ -663 +664 @@
         NULL,                   /* trace_event */
         bpf_help,               /* help */
-        NULL
+        NULL,                   /* next pointer */
+        NON_PARALLEL(true)
 };
 #endif  /* HAVE_DECL_BIOCSETIF */

lib/format_dag24.c

@@ -558 +558 @@
         trace_event_dag,                /* trace_event */
         dag_help,                       /* help */
-        NULL                            /* next pointer */
+        NULL,                            /* next pointer */
+    NON_PARALLEL(true)
 };
 

lib/format_dag25.c

@@ -79 +79 @@
  */
 
-
 #define DATA(x) ((struct dag_format_data_t *)x->format_data)
 #define DATA_OUT(x) ((struct dag_format_data_out_t *)x->format_data)
+#define STREAM_DATA(x) ((struct dag_per_stream_t *)x->data)
 
 #define FORMAT_DATA DATA(libtrace)
@@ -87 +87 @@
 
 #define DUCK FORMAT_DATA->duck
+
+#define FORMAT_DATA_HEAD FORMAT_DATA->per_stream->head
+#define FORMAT_DATA_FIRST ((struct dag_per_stream_t *)FORMAT_DATA_HEAD->data)
+
 static struct libtrace_format_t dag;
 
@@ -118 +122 @@
 };
 
+/* Data that is stored against each input stream */
+struct dag_per_stream_t {
+        /* DAG device */
+        struct dag_dev_t *device;
+        /* DAG stream number */
+        uint16_t dagstream;
+        /* Pointer to the last unread byte in the DAG memory */
+        uint8_t *top;
+        /* Pointer to the first unread byte in the DAG memory */
+        uint8_t *bottom;
+        /* Amount of data processed from the bottom pointer */
+        uint32_t processed;
+        /* Number of packets seen by the stream */
+        uint64_t pkt_count;
+        /* Drop count for this particular stream */
+        uint64_t drops;
+        /* Boolean values to indicate if a particular interface has been seen
+         * or not. This is limited to four interfaces, which is enough to
+         * support all current DAG cards */
+        uint8_t seeninterface[4];
+};
+
 /* "Global" data that is stored for each DAG input trace */
 struct dag_format_data_t {
-
         /* Data required for regular DUCK reporting */
+        /* TODO: This doesn't work with the 10X2S card! I don't know how
+         * DUCK stuff works and don't know how to fix it */
         struct {
                 /* Timestamp of the last DUCK report */
                 uint32_t last_duck;
                 /* The number of seconds between each DUCK report */
-                uint32_t duck_freq;
+                uint32_t duck_freq;
                 /* Timestamp of the last packet read from the DAG card */
-                uint32_t last_pkt;
+                uint32_t last_pkt;
                 /* Dummy trace to ensure DUCK packets are dealt with using the
                  * DUCK format functions */
-                libtrace_t *dummy_duck;
-        } duck;
+                libtrace_t *dummy_duck;
+        } duck;
 
         /* String containing the DAG device name */
         char *device_name;
-        /* The DAG device that we are reading from */
-        struct dag_dev_t *device;
-        /* The DAG stream that we are reading from */
-        unsigned int dagstream;
-        /* Boolean flag indicating whether the stream is currently attached */
+        /* Boolean flag indicating whether the trace is currently attached */
         int stream_attached;
-        /* Pointer to the first unread byte in the DAG memory hole */
-        uint8_t *bottom;
-        /* Pointer to the last unread byte in the DAG memory hole */
-        uint8_t *top;
-        /* The amount of data processed thus far from the bottom pointer */
-        uint32_t processed;
-        /* The number of packets that have been dropped */
-        uint64_t drops;
-
-        uint8_t seeninterface[4];
+
+        /* Data stored against each DAG input stream */
+        libtrace_list_t *per_stream;
 };
 
@@ -207 +223 @@
 
 /* Initialises the DAG output data structure */
-static void dag_init_format_out_data(libtrace_out_t *libtrace) {
-        libtrace->format_data = (struct dag_format_data_out_t *) malloc(sizeof(struct dag_format_data_out_t));
+static void dag_init_format_out_data(libtrace_out_t *libtrace)
+{
+        libtrace->format_data = (struct dag_format_data_out_t *)
+                malloc(sizeof(struct dag_format_data_out_t));
         // no DUCK on output
         FORMAT_DATA_OUT->stream_attached = 0;
@@ -219 +237 @@
 
 /* Initialises the DAG input data structure */
-static void dag_init_format_data(libtrace_t *libtrace) {
+static void dag_init_format_data(libtrace_t *libtrace)
+{
+        struct dag_per_stream_t stream_data;
+
         libtrace->format_data = (struct dag_format_data_t *)
                 malloc(sizeof(struct dag_format_data_t));
         DUCK.last_duck = 0;
-        DUCK.duck_freq = 0;
-        DUCK.last_pkt = 0;
-        DUCK.dummy_duck = NULL;
-        FORMAT_DATA->stream_attached = 0;
-        FORMAT_DATA->drops = 0;
+        DUCK.duck_freq = 0;
+        DUCK.last_pkt = 0;
+        DUCK.dummy_duck = NULL;
+
+        FORMAT_DATA->per_stream =
+                libtrace_list_init(sizeof(stream_data));
+        assert(FORMAT_DATA->per_stream != NULL);
+
+        /* We'll start with just one instance of stream_data, and we'll
+         * add more later if we need them */
+        memset(&stream_data, 0, sizeof(stream_data));
+        libtrace_list_push_back(FORMAT_DATA->per_stream, &stream_data);
         FORMAT_DATA->device_name = NULL;
-        FORMAT_DATA->device = NULL;
-        FORMAT_DATA->dagstream = 0;
-        FORMAT_DATA->processed = 0;
-        FORMAT_DATA->bottom = NULL;
-        FORMAT_DATA->top = NULL;
-        memset(FORMAT_DATA->seeninterface, 0, sizeof(FORMAT_DATA->seeninterface));
 }
 
@@ -241 +263 @@
  *
  * NOTE: This function assumes the open_dag_mutex is held by the caller */
-static struct dag_dev_t *dag_find_open_device(char *dev_name) {
+static struct dag_dev_t *dag_find_open_device(char *dev_name)
+{
         struct dag_dev_t *dag_dev;
 
@@ -252 +275 @@
                         dag_dev->ref_count ++;
                         return dag_dev;
-
                 }
                 dag_dev = dag_dev->next;
         }
         return NULL;
-
-
 }
 
@@ -267 +287 @@
  *
  * NOTE: This function assumes the open_dag_mutex is held by the caller */
-static void dag_close_device(struct dag_dev_t *dev) {
+static void dag_close_device(struct dag_dev_t *dev)
+{
         /* Need to remove from the device list */
-
         assert(dev->ref_count == 0);
 
@@ -292 +312 @@
  *
  * NOTE: this function should only be called when opening a DAG device for
- * writing - there is little practical difference between this and the 
+ * writing - there is little practical difference between this and the
  * function below that covers the reading case, but we need the output trace
- * object to report errors properly so the two functions take slightly 
+ * object to report errors properly so the two functions take slightly
  * different arguments. This is really lame and there should be a much better
  * way of doing this.
  *
- * NOTE: This function assumes the open_dag_mutex is held by the caller 
+ * NOTE: This function assumes the open_dag_mutex is held by the caller
  */
-static struct dag_dev_t *dag_open_output_device(libtrace_out_t *libtrace, char *dev_name) {
+static struct dag_dev_t *dag_open_output_device(libtrace_out_t *libtrace,
+                                                char *dev_name)
+{
         struct stat buf;
         int fd;
@@ -309 +331 @@
                 trace_set_err_out(libtrace,errno,"stat(%s)",dev_name);
                 return NULL;
-}
+        }
 
         /* Make sure it is the appropriate type of device */
@@ -346 +368 @@
  *
  * NOTE: this function should only be called when opening a DAG device for
- * reading - there is little practical difference between this and the 
+ * reading - there is little practical difference between this and the
  * function above that covers the writing case, but we need the input trace
- * object to report errors properly so the two functions take slightly 
+ * object to report errors properly so the two functions take slightly
  * different arguments. This is really lame and there should be a much better
  * way of doing this.
@@ -359 +381 @@
 
         /* Make sure the device exists */
-        if (stat(dev_name, &buf) == -1) {
-                trace_set_err(libtrace,errno,"stat(%s)",dev_name);
-                return NULL;
-        }
+        if (stat(dev_name, &buf) == -1) {
+                trace_set_err(libtrace,errno,"stat(%s)",dev_name);
+                return NULL;
+        }
 
         /* Make sure it is the appropriate type of device */
@@ -368 +390 @@
                 /* Try opening the DAG device */
                 if((fd = dag_open(dev_name)) < 0) {
-                        trace_set_err(libtrace,errno,"Cannot open DAG %s",
-                                        dev_name);
-                        return NULL;
-                }
+                        trace_set_err(libtrace,errno,"Cannot open DAG %s",
+                                      dev_name);
+                        return NULL;
+                }
         } else {
                 trace_set_err(libtrace,errno,"Not a valid dag device: %s",
-                                dev_name);
-                return NULL;
-        }
+                              dev_name);
+                return NULL;
+        }
 
         /* Add the device to our device list - it is just a doubly linked
@@ -397 +419 @@
 
 /* Creates and initialises a DAG output trace */
-static int dag_init_output(libtrace_out_t *libtrace) {
+static int dag_init_output(libtrace_out_t *libtrace)
+{
         char *scan = NULL;
         struct dag_dev_t *dag_device = NULL;
         int stream = 1;
-        
+
         /* XXX I don't know if this is important or not, but this function
          * isn't present in all of the driver releases that this code is
@@ -411 +434 @@
 
         dag_init_format_out_data(libtrace);
-        /* Grab the mutex while we're likely to be messing with the device 
+        /* Grab the mutex while we're likely to be messing with the device
          * list */
         pthread_mutex_lock(&open_dag_mutex);
-        
+
         /* Specific streams are signified using a comma in the libtrace URI,
          * e.g. dag:/dev/dag0,1 refers to stream 1 on the dag0 device.
@@ -460 +483 @@
 
         dag_init_format_data(libtrace);
-        /* Grab the mutex while we're likely to be messing with the device 
+        /* Grab the mutex while we're likely to be messing with the device
          * list */
         pthread_mutex_lock(&open_dag_mutex);
-        
-        
-        /* Specific streams are signified using a comma in the libtrace URI,
+
+
+        /* DAG cards support multiple streams. In a single threaded capture,
+         * these are specified using a comma in the libtrace URI,
          * e.g. dag:/dev/dag0,2 refers to stream 2 on the dag0 device.
          *
-         * If no stream is specified, we will read from stream 0 */
+         * If no stream is specified, we will read from stream 0 with
+         * one thread
+         */
         if ((scan = strchr(libtrace->uridata,',')) == NULL) {
                 FORMAT_DATA->device_name = strdup(libtrace->uridata);
@@ -477 +503 @@
         }
 
-        FORMAT_DATA->dagstream = stream;
+        FORMAT_DATA_FIRST->dagstream = stream;
 
         /* See if our DAG device is already open */
@@ -496 +522 @@
         }
 
-        FORMAT_DATA->device = dag_device;
-
-        /* See Config_Status_API_Programming_Guide.pdf from the Endace Dag Documentation */
-        /* Check kBooleanAttributeActive is true -- no point capturing on an interface that's disabled
- 
-        *  The symptom of the port being disabled is that libtrace will appear to hang.
-        */
+        FORMAT_DATA_FIRST->device = dag_device;
+
+        /* See Config_Status_API_Programming_Guide.pdf from the Endace
+           Dag Documentation */
+        /* Check kBooleanAttributeActive is true -- no point capturing
+         * on an interface that's disabled
+         *
+         * The symptom of the port being disabled is that libtrace
+         * will appear to hang. */
         /* Check kBooleanAttributeFault is false */
         /* Check kBooleanAttributeLocalFault is false */
@@ -508 +536 @@
         /* Check kBooleanAttributePeerLink ? */
 
-        /* Set kBooleanAttributePromisc/kBooleanPromiscuousMode based on libtrace promisc attribute?*/
+        /* Set kBooleanAttributePromisc/kBooleanPromiscuousMode based
+           on libtrace promisc attribute?*/
         /* Set kUint32AttributeSnapLength to the snaplength */
 
         pthread_mutex_unlock(&open_dag_mutex);
-        return 0;
+        return 0;
 }
 
 /* Configures a DAG input trace */
 static int dag_config_input(libtrace_t *libtrace, trace_option_t option,
-                                void *data) {
-        char conf_str[4096];
+                            void *data)
+{
+        char conf_str[4096];
         switch(option) {
-                case TRACE_OPTION_META_FREQ:
-                        /* This option is used to specify the frequency of DUCK
-                         * updates */
-                        DUCK.duck_freq = *(int *)data;
-                        return 0;
-                case TRACE_OPTION_SNAPLEN:
-                        /* Tell the card our new snap length */
-                        snprintf(conf_str, 4096, "varlen slen=%i", *(int *)data);
-                        if (dag_configure(FORMAT_DATA->device->fd,
-                                                conf_str) != 0) {
-                                trace_set_err(libtrace, errno, "Failed to configure snaplen on DAG card: %s", libtrace->uridata);
-                                return -1;
-                        }
-                        return 0;
-                case TRACE_OPTION_PROMISC:
-                        /* DAG already operates in a promisc fashion */
-                        return -1;
-                case TRACE_OPTION_FILTER:
-                        /* We don't yet support pushing filters into DAG 
-                         * cards */
-                        return -1;
-                case TRACE_OPTION_EVENT_REALTIME:
-                        /* Live capture is always going to be realtime */
+        case TRACE_OPTION_META_FREQ:
+                /* This option is used to specify the frequency of DUCK
+                 * updates */
+                DUCK.duck_freq = *(int *)data;
+                return 0;
+        case TRACE_OPTION_SNAPLEN:
+                /* Tell the card our new snap length */
+                snprintf(conf_str, 4096, "varlen slen=%i", *(int *)data);
+                if (dag_configure(FORMAT_DATA_FIRST->device->fd,
+                                  conf_str) != 0) {
+                        trace_set_err(libtrace, errno, "Failed to configure "
+                                      "snaplen on DAG card: %s",
+                                      libtrace->uridata);
                         return -1;
-        }
+                }
+                return 0;
+        case TRACE_OPTION_PROMISC:
+                /* DAG already operates in a promisc fashion */
+                return -1;
+        case TRACE_OPTION_FILTER:
+                /* We don't yet support pushing filters into DAG
+                 * cards */
+                return -1;
+        case TRACE_OPTION_EVENT_REALTIME:
+                /* Live capture is always going to be realtime */
+                return -1;
+        }
         return -1;
 }
 
 /* Starts a DAG output trace */
-static int dag_start_output(libtrace_out_t *libtrace) {
+static int dag_start_output(libtrace_out_t *libtrace)
+{
         struct timeval zero, nopoll;
 
@@ -557 +590 @@
 
         /* Attach and start the DAG stream */
-
         if (dag_attach_stream(FORMAT_DATA_OUT->device->fd,
                         FORMAT_DATA_OUT->dagstream, 0, 4 * 1024 * 1024) < 0) {
@@ -572 +604 @@
 
         /* We don't want the dag card to do any sleeping */
-
         dag_set_stream_poll(FORMAT_DATA_OUT->device->fd,
                         FORMAT_DATA_OUT->dagstream, 0, &zero,
@@ -581 +612 @@
 
 /* Starts a DAG input trace */
-static int dag_start_input(libtrace_t *libtrace) {
-        struct timeval zero, nopoll;
-        uint8_t *top, *bottom, *starttop;
+static int dag_start_input(libtrace_t *libtrace)
+{
+        struct timeval zero, nopoll;
+        uint8_t *top, *bottom, *starttop;
         top = bottom = NULL;
 
         zero.tv_sec = 0;
-        zero.tv_usec = 10000;
-        nopoll = zero;
+        zero.tv_usec = 10000;
+        nopoll = zero;
 
         /* Attach and start the DAG stream */
-        if (dag_attach_stream(FORMAT_DATA->device->fd,
-                                FORMAT_DATA->dagstream, 0, 0) < 0) {
-                trace_set_err(libtrace, errno, "Cannot attach DAG stream");
-                return -1;
-        }
-
-        if (dag_start_stream(FORMAT_DATA->device->fd,
-                                FORMAT_DATA->dagstream) < 0) {
-                trace_set_err(libtrace, errno, "Cannot start DAG stream");
-                return -1;
-        }
+        if (dag_attach_stream(FORMAT_DATA_FIRST->device->fd,
+                              FORMAT_DATA_FIRST->dagstream, 0, 0) < 0) {
+                trace_set_err(libtrace, errno, "Cannot attach DAG stream");
+                return -1;
+        }
+
+        if (dag_start_stream(FORMAT_DATA_FIRST->device->fd,
+                             FORMAT_DATA_FIRST->dagstream) < 0) {
+                trace_set_err(libtrace, errno, "Cannot start DAG stream");
+                return -1;
+        }
         FORMAT_DATA->stream_attached = 1;
-        
+
         /* We don't want the dag card to do any sleeping */
-        dag_set_stream_poll(FORMAT_DATA->device->fd,
-                                FORMAT_DATA->dagstream, 0, &zero,
-                                &nopoll);
-
-        starttop = dag_advance_stream(FORMAT_DATA->device->fd,
-                                        FORMAT_DATA->dagstream,
-                                        &bottom);
+        dag_set_stream_poll(FORMAT_DATA_FIRST->device->fd,
+                            FORMAT_DATA_FIRST->dagstream, 0, &zero,
+                            &nopoll);
+
+        starttop = dag_advance_stream(FORMAT_DATA_FIRST->device->fd,
+                                      FORMAT_DATA_FIRST->dagstream,
+                                      &bottom);
 
         /* Should probably flush the memory hole now */
@@ -617 +649 @@
         while (starttop - bottom > 0) {
                 bottom += (starttop - bottom);
-                top = dag_advance_stream(FORMAT_DATA->device->fd,
-                                        FORMAT_DATA->dagstream,
-                                        &bottom);
-        }
-        FORMAT_DATA->top = top;
-        FORMAT_DATA->bottom = bottom;
-        FORMAT_DATA->processed = 0;
-        FORMAT_DATA->drops = 0;
+                top = dag_advance_stream(FORMAT_DATA_FIRST->device->fd,
+                                         FORMAT_DATA_FIRST->dagstream,
+                                         &bottom);
+        }
+        FORMAT_DATA_FIRST->top = top;
+        FORMAT_DATA_FIRST->bottom = bottom;
+        FORMAT_DATA_FIRST->processed = 0;
+        FORMAT_DATA_FIRST->drops = 0;
 
         return 0;
 }
 
+static int dag_pstart_input(libtrace_t *libtrace)
+{
+        char *scan, *tok;
+        uint16_t stream_count = 0, max_streams;
+        int iserror = 0;
+        struct dag_per_stream_t stream_data;
+
+        /* Check we aren't trying to create more threads than the DAG card can
+         * handle */
+        max_streams = dag_rx_get_stream_count(FORMAT_DATA_FIRST->device->fd);
+        if (libtrace->perpkt_thread_count > max_streams) {
+                trace_set_err(libtrace, TRACE_ERR_BAD_FORMAT,
+                              "trying to create too many threads (max is %u)",
+                              max_streams);
+                iserror = 1;
+                goto cleanup;
+        }
+
+        /* Get the stream names from the uri */
+        if ((scan = strchr(libtrace->uridata, ',')) == NULL) {
+                trace_set_err(libtrace, TRACE_ERR_BAD_FORMAT,
+                              "format uri doesn't specify the DAG streams");
+                iserror = 1;
+                goto cleanup;
+        }
+
+        scan++;
+
+        tok = strtok(scan, ",");
+        while (tok != NULL) {
+                /* Ensure we haven't specified too many streams */
+                if (stream_count >= libtrace->perpkt_thread_count) {
+                        trace_set_err(libtrace, TRACE_ERR_BAD_FORMAT,
+                                      "format uri specifies too many streams. "
+                                      "Max is %u", max_streams);
+                        iserror = 1;
+                        goto cleanup;
+                }
+
+                /* Save the stream details */
+                if (stream_count == 0) {
+                        /* Special case where we update the existing stream
+                         * data structure */
+                        FORMAT_DATA_FIRST->dagstream = (uint16_t)atoi(tok);
+                } else {
+                        memset(&stream_data, 0, sizeof(stream_data));
+                        stream_data.device = FORMAT_DATA_FIRST->device;
+                        stream_data.dagstream = (uint16_t)atoi(tok);
+                        libtrace_list_push_back(FORMAT_DATA->per_stream,
+                                                &stream_data);
+                }
+
+                stream_count++;
+                tok = strtok(NULL, ",");
+        }
+
+        FORMAT_DATA->stream_attached = 1;
+
+ cleanup:
+        if (iserror) {
+                return -1;
+        } else {
+                return 0;
+        }
+}
+
 /* Pauses a DAG output trace */
-static int dag_pause_output(libtrace_out_t *libtrace) {
-
+static int dag_pause_output(libtrace_out_t *libtrace)
+{
         /* Stop and detach the stream */
         if (dag_stop_stream(FORMAT_DATA_OUT->device->fd,
-                        FORMAT_DATA_OUT->dagstream) < 0) {
+                            FORMAT_DATA_OUT->dagstream) < 0) {
                 trace_set_err_out(libtrace, errno, "Could not stop DAG stream");
                 return -1;
         }
         if (dag_detach_stream(FORMAT_DATA_OUT->device->fd,
-                        FORMAT_DATA_OUT->dagstream) < 0) {
-                trace_set_err_out(libtrace, errno, "Could not detach DAG stream");
+                              FORMAT_DATA_OUT->dagstream) < 0) {
+                trace_set_err_out(libtrace, errno,
+                                  "Could not detach DAG stream");
                 return -1;
         }
@@ -648 +747 @@
 
 /* Pauses a DAG input trace */
-static int dag_pause_input(libtrace_t *libtrace) {
-
-        /* Stop and detach the stream */
-        if (dag_stop_stream(FORMAT_DATA->device->fd,
-                                FORMAT_DATA->dagstream) < 0) {
-                trace_set_err(libtrace, errno, "Could not stop DAG stream");
-                return -1;
-        }
-        if (dag_detach_stream(FORMAT_DATA->device->fd,
-                                FORMAT_DATA->dagstream) < 0) {
-                trace_set_err(libtrace, errno, "Could not detach DAG stream");
-                return -1;
-        }
+static int dag_pause_input(libtrace_t *libtrace)
+{
+        libtrace_list_node_t *tmp = FORMAT_DATA_HEAD;
+
+        /* Stop and detach each stream */
+        while (tmp != NULL) {
+                if (dag_stop_stream(STREAM_DATA(tmp)->device->fd,
+                                    STREAM_DATA(tmp)->dagstream) < 0) {
+                        trace_set_err(libtrace, errno,
+                                      "Could not stop DAG stream");
+                        printf("Count not stop DAG stream\n");
+                        return -1;
+                }
+                if (dag_detach_stream(STREAM_DATA(tmp)->device->fd,
+                                      STREAM_DATA(tmp)->dagstream) < 0) {
+                        trace_set_err(libtrace, errno,
+                                      "Could not detach DAG stream");
+                        printf("Count not detach DAG stream\n");
+                        return -1;
+                }
+
+                tmp = tmp->next;
+        }
+
         FORMAT_DATA->stream_attached = 0;
         return 0;
 }
 
+
+
 /* Closes a DAG input trace */
-static int dag_fin_input(libtrace_t *libtrace) {
+static int dag_fin_input(libtrace_t *libtrace)
+{
+        libtrace_list_node_t *tmp = FORMAT_DATA_HEAD;
+
         /* Need the lock, since we're going to be handling the device list */
         pthread_mutex_lock(&open_dag_mutex);
-        
+
         /* Detach the stream if we are not paused */
         if (FORMAT_DATA->stream_attached)
                 dag_pause_input(libtrace);
-        FORMAT_DATA->device->ref_count --;
-
-        /* Close the DAG device if there are no more references to it */
-        if (FORMAT_DATA->device->ref_count == 0)
-                dag_close_device(FORMAT_DATA->device);
+
+        /* Close any dag devices that have no more references */
+        while (tmp != NULL) {
+                STREAM_DATA(tmp)->device->ref_count--;
+                if (STREAM_DATA(tmp)->device->ref_count == 0)
+                        dag_close_device(STREAM_DATA(tmp)->device);
+
+                tmp = tmp->next;
+        }
+
         if (DUCK.dummy_duck)
                 trace_destroy_dead(DUCK.dummy_duck);
+
+        /* Clear the list */
+        libtrace_list_deinit(FORMAT_DATA->per_stream);
+
         if (FORMAT_DATA->device_name)
                 free(FORMAT_DATA->device_name);
         free(libtrace->format_data);
         pthread_mutex_unlock(&open_dag_mutex);
-        return 0; /* success */
+        return 0; /* success */
 }
 
 /* Closes a DAG output trace */
-static int dag_fin_output(libtrace_out_t *libtrace) {
-        
+static int dag_fin_output(libtrace_out_t *libtrace)
+{
+
         /* Commit any outstanding traffic in the txbuffer */
         if (FORMAT_DATA_OUT->waiting) {
-                dag_tx_stream_commit_bytes(FORMAT_DATA_OUT->device->fd, FORMAT_DATA_OUT->dagstream,
-                                FORMAT_DATA_OUT->waiting );
-        }
-
-        /* Wait until the buffer is nearly clear before exiting the program, 
+                dag_tx_stream_commit_bytes(FORMAT_DATA_OUT->device->fd,
+                                           FORMAT_DATA_OUT->dagstream,
+                                           FORMAT_DATA_OUT->waiting );
+        }
+
+        /* Wait until the buffer is nearly clear before exiting the program,
          * as we will lose packets otherwise */
-        dag_tx_get_stream_space(FORMAT_DATA_OUT->device->fd,
-                        FORMAT_DATA_OUT->dagstream,
-                        dag_get_stream_buffer_size(FORMAT_DATA_OUT->device->fd,
-                                        FORMAT_DATA_OUT->dagstream) - 8
-                        );
+        dag_tx_get_stream_space
+                (FORMAT_DATA_OUT->device->fd,
+                 FORMAT_DATA_OUT->dagstream,
+                 dag_get_stream_buffer_size(FORMAT_DATA_OUT->device->fd,
+                                            FORMAT_DATA_OUT->dagstream) - 8);
 
         /* Need the lock, since we're going to be handling the device list */
@@ -752 +878 @@
 
         /* Allocate memory for the DUCK data */
-        if (packet->buf_control == TRACE_CTRL_EXTERNAL ||
-                        !packet->buffer) {
-                packet->buffer = malloc(LIBTRACE_PACKET_BUFSIZE);
-                packet->buf_control = TRACE_CTRL_PACKET;
-                if (!packet->buffer) {
-                        trace_set_err(libtrace, errno,
-                                        "Cannot allocate packet buffer");
-                        return -1;
-                }
-        }
+        if (packet->buf_control == TRACE_CTRL_EXTERNAL ||
+            !packet->buffer) {
+                packet->buffer = malloc(LIBTRACE_PACKET_BUFSIZE);
+                packet->buf_control = TRACE_CTRL_PACKET;
+                if (!packet->buffer) {
+                        trace_set_err(libtrace, errno,
+                                      "Cannot allocate packet buffer");
+                        return -1;
+                }
+        }
 
         /* DUCK doesn't have a format header */
@@ -789 +915 @@
 
 /* Determines the amount of data available to read from the DAG card */
-static int dag_available(libtrace_t *libtrace) {
-        uint32_t diff = FORMAT_DATA->top - FORMAT_DATA->bottom;
+static int dag_available(libtrace_t *libtrace,
+                         struct dag_per_stream_t *stream_data)
+{
+        uint32_t diff = stream_data->top - stream_data->bottom;
 
         /* If we've processed more than 4MB of data since we last called
          * dag_advance_stream, then we should call it again to allow the
          * space occupied by that 4MB to be released */
-        if (diff >= dag_record_size && FORMAT_DATA->processed < 4 * 1024 * 1024)
+        if (diff >= dag_record_size && stream_data->processed < 4 * 1024 * 1024)
                 return diff;
-        
+
         /* Update the top and bottom pointers */
-        FORMAT_DATA->top = dag_advance_stream(FORMAT_DATA->device->fd,
-                        FORMAT_DATA->dagstream,
-                        &(FORMAT_DATA->bottom));
-        
-        if (FORMAT_DATA->top == NULL) {
+        stream_data->top = dag_advance_stream(stream_data->device->fd,
+                                              stream_data->dagstream,
+                                              &(stream_data->bottom));
+
+        if (stream_data->top == NULL) {
                 trace_set_err(libtrace, errno, "dag_advance_stream failed!");
                 return -1;
         }
-        FORMAT_DATA->processed = 0;
-        diff = FORMAT_DATA->top - FORMAT_DATA->bottom;
+        stream_data->processed = 0;
+        diff = stream_data->top - stream_data->bottom;
         return diff;
 }
 
 /* Returns a pointer to the start of the next complete ERF record */
-static dag_record_t *dag_get_record(libtrace_t *libtrace) {
-        dag_record_t *erfptr = NULL;
-        uint16_t size;
-        erfptr = (dag_record_t *)FORMAT_DATA->bottom;
+static dag_record_t *dag_get_record(struct dag_per_stream_t *stream_data)
+{
+        dag_record_t *erfptr = NULL;
+        uint16_t size;
+
+        erfptr = (dag_record_t *)stream_data->bottom;
         if (!erfptr)
-                return NULL;
-        size = ntohs(erfptr->rlen);
-        assert( size >= dag_record_size );
+                return NULL;
+
+        size = ntohs(erfptr->rlen);
+        assert( size >= dag_record_size );
+
         /* Make certain we have the full packet available */
-        if (size > (FORMAT_DATA->top - FORMAT_DATA->bottom))
+        if (size > (stream_data->top - stream_data->bottom))
                 return NULL;
-        FORMAT_DATA->bottom += size;
-        FORMAT_DATA->processed += size;
+
+        stream_data->bottom += size;
+        stream_data->processed += size;
         return erfptr;
 }
@@ -831 +964 @@
 /* Converts a buffer containing a recently read DAG packet record into a
  * libtrace packet */
-static int dag_prepare_packet(libtrace_t *libtrace, libtrace_packet_t *packet,
-                void *buffer, libtrace_rt_types_t rt_type, uint32_t flags) {
-
+static int dag_prepare_packet_real(libtrace_t *libtrace,
+                                   struct dag_per_stream_t *stream_data,
+                                   libtrace_packet_t *packet,
+                                   void *buffer, libtrace_rt_types_t rt_type,
+                                   uint32_t flags)
+{
         dag_record_t *erfptr;
-        
+
         /* If the packet previously owned a buffer that is not the buffer
-         * that contains the new packet data, we're going to need to free the
-         * old one to avoid memory leaks */
+         * that contains the new packet data, we're going to need to free the
+         * old one to avoid memory leaks */
         if (packet->buffer != buffer &&
-                        packet->buf_control == TRACE_CTRL_PACKET) {
+            packet->buf_control == TRACE_CTRL_PACKET) {
                 free(packet->buffer);
         }
@@ -853 +989 @@
         erfptr = (dag_record_t *)buffer;
         packet->buffer = erfptr;
-        packet->header = erfptr;
-        packet->type = rt_type;
+        packet->header = erfptr;
+        packet->type = rt_type;
 
         if (erfptr->flags.rxerror == 1) {
-                /* rxerror means the payload is corrupt - drop the payload
-                 * by tweaking rlen */
-                packet->payload = NULL;
-                erfptr->rlen = htons(erf_get_framing_length(packet));
-        } else {
-                packet->payload = (char*)packet->buffer
-                        + erf_get_framing_length(packet);
-        }
+                /* rxerror means the payload is corrupt - drop the payload
+                 * by tweaking rlen */
+                packet->payload = NULL;
+                erfptr->rlen = htons(erf_get_framing_length(packet));
+        } else {
+                packet->payload = (char*)packet->buffer
+                        + erf_get_framing_length(packet);
+        }
 
         if (libtrace->format_data == NULL) {
@@ -871 +1007 @@
 
         /* Update the dropped packets counter */
-
-        /* No loss counter for DSM coloured records - have to use
-         * some other API */
+        /* No loss counter for DSM coloured records - have to use some
+         * other API */
         if (erfptr->type == TYPE_DSM_COLOR_ETH) {
                 /* TODO */
         } else {
                 /* Use the ERF loss counter */
-                if (FORMAT_DATA->seeninterface[erfptr->flags.iface] == 0) {
-                        FORMAT_DATA->seeninterface[erfptr->flags.iface] = 1;
+                if (stream_data->seeninterface[erfptr->flags.iface]
+                    == 0) {
+                        stream_data->seeninterface[erfptr->flags.iface]
+                                = 1;
                 } else {
-                        FORMAT_DATA->drops += ntohs(erfptr->lctr);
-                }
-        }
+                        stream_data->drops += ntohs(erfptr->lctr);
+                }
+        }
+
+        packet->error = 1;
 
         return 0;
+}
+
+static int dag_prepare_packet(libtrace_t *libtrace, libtrace_packet_t *packet,
+                              void *buffer, libtrace_rt_types_t rt_type,
+                              uint32_t flags)
+{
+        return dag_prepare_packet_real(libtrace, FORMAT_DATA_FIRST, packet,
+                                       buffer, rt_type, flags);
 }
 
@@ -898 +1045 @@
 /* Pushes an ERF record onto the transmit stream */
 static int dag_dump_packet(libtrace_out_t *libtrace,
-                dag_record_t *erfptr, unsigned int pad, void *buffer) {
+                           dag_record_t *erfptr, unsigned int pad,
+                           void *buffer)
+{
         int size;
 
         /*
-         * If we've got 0 bytes waiting in the txqueue, assume that we haven't
-         * requested any space yet, and request some, storing the pointer at
-         * FORMAT_DATA_OUT->txbuffer.
+         * If we've got 0 bytes waiting in the txqueue, assume that we
+         * haven't requested any space yet, and request some, storing
+         * the pointer at FORMAT_DATA_OUT->txbuffer.
          *
          * The amount to request is slightly magical at the moment - it's
@@ -911 +1060 @@
          */
         if (FORMAT_DATA_OUT->waiting == 0) {
-                FORMAT_DATA_OUT->txbuffer = dag_tx_get_stream_space(FORMAT_DATA_OUT->device->fd,
-                                FORMAT_DATA_OUT->dagstream, 16908288);
+                FORMAT_DATA_OUT->txbuffer =
+                        dag_tx_get_stream_space(FORMAT_DATA_OUT->device->fd,
+                                                FORMAT_DATA_OUT->dagstream,
+                                                16908288);
         }
 
@@ -919 +1070 @@
          * are in contiguous memory
          */
-        memcpy(FORMAT_DATA_OUT->txbuffer + FORMAT_DATA_OUT->waiting,erfptr,(dag_record_size + pad));
+        memcpy(FORMAT_DATA_OUT->txbuffer + FORMAT_DATA_OUT->waiting, erfptr,
+               (dag_record_size + pad));
         FORMAT_DATA_OUT->waiting += (dag_record_size + pad);
-
-
 
         /*
@@ -929 +1079 @@
          */
         size = ntohs(erfptr->rlen)-(dag_record_size + pad);
-        memcpy(FORMAT_DATA_OUT->txbuffer + FORMAT_DATA_OUT->waiting,buffer,size);
+        memcpy(FORMAT_DATA_OUT->txbuffer + FORMAT_DATA_OUT->waiting, buffer,
+               size);
         FORMAT_DATA_OUT->waiting += size;
 
@@ -938 +1089 @@
          * case there is still data in the buffer at program exit.
          */
-
         if (FORMAT_DATA_OUT->waiting >= 16*1024*1024) {
-                FORMAT_DATA_OUT->txbuffer = dag_tx_stream_commit_bytes(FORMAT_DATA_OUT->device->fd, FORMAT_DATA_OUT->dagstream,
-                        FORMAT_DATA_OUT->waiting );
+                FORMAT_DATA_OUT->txbuffer =
+                        dag_tx_stream_commit_bytes(FORMAT_DATA_OUT->device->fd,
+                                                   FORMAT_DATA_OUT->dagstream,
+                                                   FORMAT_DATA_OUT->waiting);
                 FORMAT_DATA_OUT->waiting = 0;
         }
 
         return size + pad + dag_record_size;
-
 }
 
@@ -952 +1103 @@
  * if one is found, false otherwise */
 static bool find_compatible_linktype(libtrace_out_t *libtrace,
-                                libtrace_packet_t *packet, char *type)
-{
-         // Keep trying to simplify the packet until we can find
-         //something we can do with it
+                                     libtrace_packet_t *packet, char *type)
+{
+        /* Keep trying to simplify the packet until we can find
+         * something we can do with it */
 
         do {
-                *type=libtrace_to_erf_type(trace_get_link_type(packet));
-
-                // Success
+                *type = libtrace_to_erf_type(trace_get_link_type(packet));
+
+                /* Success */
                 if (*type != (char)-1)
                         return true;
@@ -966 +1117 @@
                 if (!demote_packet(packet)) {
                         trace_set_err_out(libtrace,
-                                        TRACE_ERR_NO_CONVERSION,
-                                        "No erf type for packet (%i)",
-                                        trace_get_link_type(packet));
+                                          TRACE_ERR_NO_CONVERSION,
+                                          "No erf type for packet (%i)",
+                                          trace_get_link_type(packet));
                         return false;
                 }
@@ -978 +1129 @@
 
 /* Writes a packet to the provided DAG output trace */
-static int dag_write_packet(libtrace_out_t *libtrace, libtrace_packet_t *packet) {
-        /*
-         * This is heavily borrowed from erf_write_packet(). Yes, CnP coding 
-         * sucks, sorry about that.
+static int dag_write_packet(libtrace_out_t *libtrace, libtrace_packet_t *packet)
+{
+        /* This is heavily borrowed from erf_write_packet(). Yes, CnP
+         * coding sucks, sorry about that.
          */
         unsigned int pad = 0;
@@ -990 +1141 @@
 
         if(!packet->header) {
-                /* No header, probably an RT packet. Lifted from 
+                /* No header, probably an RT packet. Lifted from
                  * erf_write_packet(). */
                 return -1;
@@ -1009 +1160 @@
 
         if (packet->type == TRACE_RT_DATA_ERF) {
-                numbytes = dag_dump_packet(libtrace,
-                                header,
-                                pad,
-                                payload
-                                );
-
+                numbytes = dag_dump_packet(libtrace, header, pad, payload);
         } else {
                 /* Build up a new packet header from the existing header */
 
-                /* Simplify the packet first - if we can't do this, break 
+                /* Simplify the packet first - if we can't do this, break
                  * early */
                 if (!find_compatible_linktype(libtrace,packet,&erf_type))
@@ -1037 +1183 @@
 
                 /* Packet length (rlen includes format overhead) */
-                assert(trace_get_capture_length(packet)>0
-                                && trace_get_capture_length(packet)<=65536);
-                assert(erf_get_framing_length(packet)>0
-                                && trace_get_framing_length(packet)<=65536);
-                assert(trace_get_capture_length(packet)+erf_get_framing_length(packet)>0
-                      &&trace_get_capture_length(packet)+erf_get_framing_length(packet)<=65536);
+                assert(trace_get_capture_length(packet) > 0
+                       && trace_get_capture_length(packet) <= 65536);
+                assert(erf_get_framing_length(packet) > 0
+                       && trace_get_framing_length(packet) <= 65536);
+                assert(trace_get_capture_length(packet) +
+                       erf_get_framing_length(packet) > 0
+                       && trace_get_capture_length(packet) +
+                       erf_get_framing_length(packet) <= 65536);
 
                 erfhdr.rlen = htons(trace_get_capture_length(packet)
-                        + erf_get_framing_length(packet));
+                                    + erf_get_framing_length(packet));
 
 
@@ -1054 +1202 @@
 
                 /* Write it out */
-                numbytes = dag_dump_packet(libtrace,
-                                &erfhdr,
-                                pad,
-                                payload);
-
+                numbytes = dag_dump_packet(libtrace, &erfhdr, pad, payload);
         }
 
@@ -1068 +1212 @@
  * If DUCK reporting is enabled, the packet returned may be a DUCK update
  */
-static int dag_read_packet(libtrace_t *libtrace, libtrace_packet_t *packet) {
-        int size = 0;
-        struct timeval tv;
-        dag_record_t *erfptr = NULL;
+static int dag_read_packet_real(libtrace_t *libtrace,
+                                struct dag_per_stream_t *stream_data,
+                                libtrace_thread_t *t, /* Optional */
+                                libtrace_packet_t *packet)
+{
+        dag_record_t *erfptr = NULL;
         int numbytes = 0;
         uint32_t flags = 0;
-        struct timeval maxwait;
-        struct timeval pollwait;
+        struct timeval maxwait, pollwait;
 
         pollwait.tv_sec = 0;
@@ -1088 +1233 @@
                 return size;
 
-
         /* Don't let anyone try to free our DAG memory hole! */
         flags |= TRACE_PREP_DO_NOT_OWN_BUFFER;
@@ -1094 +1238 @@
         /* If the packet buffer is currently owned by libtrace, free it so
          * that we can set the packet to point into the DAG memory hole */
-        if (packet->buf_control == TRACE_CTRL_PACKET) {
-                free(packet->buffer);
-                packet->buffer = 0;
-        }
-        
-        if (dag_set_stream_poll(FORMAT_DATA->device->fd, 
-                        FORMAT_DATA->dagstream, sizeof(dag_record_t), &maxwait, 
-                        &pollwait) == -1)
-        {
+        if (packet->buf_control == TRACE_CTRL_PACKET) {
+                free(packet->buffer);
+                packet->buffer = 0;
+        }
+
+        if (dag_set_stream_poll(stream_data->device->fd, stream_data->dagstream,
+                                sizeof(dag_record_t), &maxwait,
+                                &pollwait) == -1) {
                 trace_set_err(libtrace, errno, "dag_set_stream_poll");
                 return -1;
         }
 
-
         /* Grab a full ERF record */
         do {
-                numbytes = dag_available(libtrace);
+                numbytes = dag_available(libtrace, stream_data);
                 if (numbytes < 0)
                         return numbytes;
                 if (numbytes < dag_record_size) {
+                        /* Check the message queue if we have one to check */
+                        if (t != NULL &&
+                            libtrace_message_queue_count(&t->messages) > 0)
+                                return -2;
+
                         if (libtrace_halt)
                                 return 0;
@@ -1119 +1266 @@
                         continue;
                 }
-                erfptr = dag_get_record(libtrace);
+                erfptr = dag_get_record(stream_data);
         } while (erfptr == NULL);
 
         /* Prepare the libtrace packet */
-        if (dag_prepare_packet(libtrace, packet, erfptr, TRACE_RT_DATA_ERF,
-                                flags))
+        if (dag_prepare_packet_real(libtrace, stream_data, packet, erfptr,
+                                    TRACE_RT_DATA_ERF, flags))
                 return -1;
 
-        /* Update the DUCK timer */
-        tv = trace_get_timeval(packet);
-        DUCK.last_pkt = tv.tv_sec;
-
         return packet->payload ? htons(erfptr->rlen) :
-                                erf_get_framing_length(packet);
+                erf_get_framing_length(packet);
+}
+
+static int dag_read_packet(libtrace_t *libtrace, libtrace_packet_t *packet)
+{
+        return dag_read_packet_real(libtrace, FORMAT_DATA_FIRST, NULL, packet);
+}
+
+static int dag_pread_packets(libtrace_t *libtrace, libtrace_thread_t *t,
+                             libtrace_packet_t **packets, size_t nb_packets)
+{
+        int ret;
+        size_t read_packets = 0;
+        int numbytes = 0;
+
+        struct dag_per_stream_t *stream_data =
+                (struct dag_per_stream_t *)t->format_data;
+
+        /* Read as many packets as we can, but read atleast one packet */
+        do {
+                ret = dag_read_packet_real(libtrace, stream_data, t,
+                                           packets[read_packets]);
+                if (ret < 0)
+                        return ret;
+
+                read_packets++;
+
+                /* Make sure we don't read too many packets..! */
+                if (read_packets >= nb_packets)
+                        break;
+
+                numbytes = dag_available(libtrace, stream_data);
+        } while (numbytes >= dag_record_size);
+
+        return read_packets;
 }
 
@@ -1140 +1317 @@
  */
 static libtrace_eventobj_t trace_event_dag(libtrace_t *libtrace,
-                                        libtrace_packet_t *packet) {
-        libtrace_eventobj_t event = {0,0,0.0,0};
+                                           libtrace_packet_t *packet)
+{
+        libtrace_eventobj_t event = {0,0,0.0,0};
         dag_record_t *erfptr = NULL;
         int numbytes;
@@ -1160 +1338 @@
         }
         
-        if (dag_set_stream_poll(FORMAT_DATA->device->fd, 
-                        FORMAT_DATA->dagstream, 0, &minwait, 
-                        &minwait) == -1)
-        {
+        if (dag_set_stream_poll(FORMAT_DATA_FIRST->device->fd,
+                                FORMAT_DATA_FIRST->dagstream, 0, &minwait,
+                                &minwait) == -1) {
                 trace_set_err(libtrace, errno, "dag_set_stream_poll");
                 event.type = TRACE_EVENT_TERMINATE;
@@ -1172 +1349 @@
                 erfptr = NULL;
                 numbytes = 0;
-        
+
                 /* Need to call dag_available so that the top pointer will get
                  * updated, otherwise we'll never see any data! */
-                numbytes = dag_available(libtrace);
-
-                /* May as well not bother calling dag_get_record if 
+                numbytes = dag_available(libtrace, FORMAT_DATA_FIRST);
+
+                /* May as well not bother calling dag_get_record if
                  * dag_available suggests that there's no data */
                 if (numbytes != 0)
-                        erfptr = dag_get_record(libtrace);
+                        erfptr = dag_get_record(FORMAT_DATA_FIRST);
                 if (erfptr == NULL) {
                         /* No packet available - sleep for a very short time */
                         if (libtrace_halt) {
                                 event.type = TRACE_EVENT_TERMINATE;
-                        } else {                        
+                        } else {
                                 event.type = TRACE_EVENT_SLEEP;
                                 event.seconds = 0.0001;
@@ -1191 +1368 @@
                         break;
                 }
-                if (dag_prepare_packet(libtrace, packet, erfptr, 
-                                        TRACE_RT_DATA_ERF, flags)) {
+                if (dag_prepare_packet_real(libtrace, FORMAT_DATA_FIRST, packet,
+                                            erfptr, TRACE_RT_DATA_ERF, flags)) {
                         event.type = TRACE_EVENT_TERMINATE;
                         break;
@@ -1198 +1375 @@
 
 
-                event.size = trace_get_capture_length(packet) + 
-                                trace_get_framing_length(packet);
-                
+                event.size = trace_get_capture_length(packet) +
+                        trace_get_framing_length(packet);
+
                 /* XXX trace_read_packet() normally applies the following
                  * config options for us, but this function is called via
@@ -1206 +1383 @@
 
                 if (libtrace->filter) {
-                        int filtret = trace_apply_filter(libtrace->filter, 
-                                        packet);
+                        int filtret = trace_apply_filter(libtrace->filter,
+                                                         packet);
                         if (filtret == -1) {
                                 trace_set_err(libtrace, TRACE_ERR_BAD_FILTER,
-                                                "Bad BPF Filter");
+                                              "Bad BPF Filter");
                                 event.type = TRACE_EVENT_TERMINATE;
                                 break;
@@ -1221 +1398 @@
                                  * a sleep event in this case, like we used to
                                  * do! */
-                                libtrace->filtered_packets ++;
+                                libtrace->filtered_packets ++;
                                 trace_clear_cache(packet);
                                 continue;
                         }
-                                
+
                         event.type = TRACE_EVENT_PACKET;
                 } else {
@@ -1238 +1415 @@
                         trace_set_capture_length(packet, libtrace->snaplen);
                 }
-                libtrace->accepted_packets ++;
+                libtrace->accepted_packets ++;
                 break;
-        } while (1);
+        } while(1);
 
         return event;
@@ -1246 +1423 @@
 
 /* Gets the number of dropped packets */
-static uint64_t dag_get_dropped_packets(libtrace_t *trace) {
-        if (trace->format_data == NULL)
-                return (uint64_t)-1;
-        return DATA(trace)->drops;
+static uint64_t dag_get_dropped_packets(libtrace_t *libtrace)
+{
+        uint64_t sum = 0;
+        libtrace_list_node_t *tmp = FORMAT_DATA_HEAD;
+
+        /* Sum the drop counter for all the packets */
+        while (tmp != NULL) {
+                sum += STREAM_DATA(tmp)->drops;
+                tmp = tmp->next;
+        }
+
+        return sum;
 }
 
 /* Prints some semi-useful help text about the DAG format module */
 static void dag_help(void) {
-        printf("dag format module: $Revision: 1755 $\n");
-        printf("Supported input URIs:\n");
-        printf("\tdag:/dev/dagn\n");
-        printf("\n");
-        printf("\te.g.: dag:/dev/dag0\n");
-        printf("\n");
-        printf("Supported output URIs:\n");
-        printf("\tnone\n");
-        printf("\n");
+        printf("dag format module: $Revision: 1755 $\n");
+        printf("Supported input URIs:\n");
+        printf("\tdag:/dev/dagn\n");
+        printf("\n");
+        printf("\te.g.: dag:/dev/dag0\n");
+        printf("\n");
+        printf("Supported output URIs:\n");
+        printf("\tnone\n");
+        printf("\n");
+}
+
+static int dag_pconfig_input(UNUSED libtrace_t *libtrace,
+                             trace_parallel_option_t option, UNUSED void *value)
+{
+        /* We don't support any of these! Normally you configure the DAG card
+         * externally. */
+        switch(option) {
+        case TRACE_OPTION_SET_HASHER:
+        case TRACE_OPTION_SET_PERPKT_THREAD_COUNT:
+        case TRACE_OPTION_TRACETIME:
+        case TRACE_OPTION_TICK_INTERVAL:
+        case TRACE_OPTION_GET_CONFIG:
+        case TRACE_OPTION_SET_CONFIG:
+                return -1;
+        }
+        /* We don't provide a default option to ensure that future options will
+         * generate a compiler warning. */
+
+        return -1;
+}
+
+/* TODO: Should possibly make a more generic dag_start_input, as there's a
+ * fair bit of code duplication between that and this */
+static int dag_pregister_thread(libtrace_t *libtrace, libtrace_thread_t *t,
+                                bool reader)
+{
+        struct timeval zero, nopoll;
+        uint8_t *top, *bottom, *starttop;
+        struct dag_per_stream_t *stream_data;
+        top = bottom = NULL;
+
+        /* Minimum delay is 10mS */
+        zero.tv_sec = 0;
+        zero.tv_usec = 10000;
+        nopoll = zero;
+
+        if (reader) {
+                if (t->type == THREAD_PERPKT) {
+                        stream_data =
+                                (struct dag_per_stream_t *)
+                                libtrace_list_get_index(FORMAT_DATA->per_stream,
+                                                        t->perpkt_num)->data;
+
+                        /* Pass the per thread data to the thread */
+                        t->format_data = stream_data;
+
+                        /* Attach and start the DAG stream */
+                        printf("t%u: starting and attaching stream #%u\n",
+                               t->perpkt_num, stream_data->dagstream);
+                        if (dag_attach_stream(stream_data->device->fd,
+                                              stream_data->dagstream, 0,
+                                              0) < 0) {
+                                printf("can't attach DAG stream #%u\n",
+                                       stream_data->dagstream);
+                                trace_set_err(libtrace, errno,
+                                              "can't attach DAG stream #%u",
+                                              stream_data->dagstream);
+                                return -1;
+                        }
+                        if (dag_start_stream(stream_data->device->fd,
+                                             stream_data->dagstream) < 0) {
+                                trace_set_err(libtrace, errno,
+                                              "can't start DAG stream #%u",
+                                              stream_data->dagstream);
+                                printf("can't start DAG stream #%u\n",
+                                       stream_data->dagstream);
+                                return -1;
+                        }
+
+                        /* Ensure that dag_advance_stream will return without
+                         * blocking */
+                        if(dag_set_stream_poll(stream_data->device->fd,
+                                               stream_data->dagstream, 0, &zero,
+                                               &nopoll) < 0) {
+                                trace_set_err(libtrace, errno,
+                                              "dag_set_stream_poll failed!");
+                                return -1;
+                        }
+
+                        /* Clear all the data from the memory hole */
+                        starttop = dag_advance_stream(stream_data->
+                                                      device->fd,
+                                                      stream_data->dagstream,
+                                                      &bottom);
+
+                        top = starttop;
+                        while (starttop - bottom > 0) {
+                                bottom += (starttop - bottom);
+                                top = dag_advance_stream(stream_data->
+                                                         device->fd,
+                                                         stream_data->dagstream,
+                                                         &bottom);
+                        }
+                        stream_data->top = top;
+                        stream_data->bottom = bottom;
+                        stream_data->pkt_count = 0;
+                        stream_data->drops = 0;
+                } else {
+                        /* TODO: Figure out why t->type != THREAD_PERPKT in
+                         * order to figure out what this line does */
+                        t->format_data = FORMAT_DATA_FIRST;
+                }
+        }
+
+        fprintf(stderr, "t%u: registered thread\n", t->perpkt_num);
+
+        return 0;
 }
 
 static struct libtrace_format_t dag = {
-        "dag",
-        "$Id$",
-        TRACE_FORMAT_ERF,
+        "dag",
+        "$Id$",
+        TRACE_FORMAT_ERF,
         dag_probe_filename,             /* probe filename */
         NULL,                           /* probe magic */
-        dag_init_input,                 /* init_input */
-        dag_config_input,               /* config_input */
-        dag_start_input,                /* start_input */
-        dag_pause_input,                /* pause_input */
+        dag_init_input,                 /* init_input */
+        dag_config_input,               /* config_input */
+        dag_start_input,                /* start_input */
+        dag_pause_input,                /* pause_input */
         dag_init_output,                /* init_output */
-        NULL,                           /* config_output */
+        NULL,                           /* config_output */
         dag_start_output,               /* start_output */
-        dag_fin_input,                  /* fin_input */
+        dag_fin_input,                  /* fin_input */
         dag_fin_output,                 /* fin_output */
-        dag_read_packet,                /* read_packet */
-        dag_prepare_packet,             /* prepare_packet */
+        dag_read_packet,                /* read_packet */
+        dag_prepare_packet,             /* prepare_packet */
         NULL,                           /* fin_packet */
         dag_write_packet,               /* write_packet */
-        erf_get_link_type,              /* get_link_type */
-        erf_get_direction,              /* get_direction */
-        erf_set_direction,              /* set_direction */
-        erf_get_erf_timestamp,          /* get_erf_timestamp */
-        NULL,                           /* get_timeval */
-        NULL,                           /* get_seconds */
+        erf_get_link_type,              /* get_link_type */
+        erf_get_direction,              /* get_direction */
+        erf_set_direction,              /* set_direction */
+        erf_get_erf_timestamp,          /* get_erf_timestamp */
+        NULL,                           /* get_timeval */
+        NULL,                           /* get_seconds */
         NULL,                           /* get_timespec */
-        NULL,                           /* seek_erf */
-        NULL,                           /* seek_timeval */
-        NULL,                           /* seek_seconds */
-        erf_get_capture_length,         /* get_capture_length */
-        erf_get_wire_length,            /* get_wire_length */
-        erf_get_framing_length,         /* get_framing_length */
-        erf_set_capture_length,         /* set_capture_length */
+        NULL,                           /* seek_erf */
+        NULL,                           /* seek_timeval */
+        NULL,                           /* seek_seconds */
+        erf_get_capture_length,         /* get_capture_length */
+        erf_get_wire_length,            /* get_wire_length */
+        erf_get_framing_length,         /* get_framing_length */
+        erf_set_capture_length,         /* set_capture_length */
         NULL,                           /* get_received_packets */
         NULL,                           /* get_filtered_packets */
         dag_get_dropped_packets,        /* get_dropped_packets */
         NULL,                           /* get_captured_packets */
-        NULL,                           /* get_fd */
-        trace_event_dag,                /* trace_event */
-        dag_help,                       /* help */
-        NULL                            /* next pointer */
+        NULL,                           /* get_fd */
+        trace_event_dag,                /* trace_event */
+        dag_help,                       /* help */
+        NULL,                            /* next pointer */
+        {true, 0}, /* live packet capture, thread limit TBD */
+        dag_pstart_input,
+        dag_pread_packets,
+        dag_pause_input,
+        NULL,
+        dag_pconfig_input,
+        dag_pregister_thread,
+        NULL
 };
 
-void dag_constructor(void) {
+void dag_constructor(void)
+{
         register_format(&dag);
 }

lib/format_dpdk.c

@@ -3 +3 @@
  * This file is part of libtrace
  *
- * Copyright (c) 2007,2008,2009,2010 The University of Waikato, Hamilton, 
+ * Copyright (c) 2007,2008,2009,2010 The University of Waikato, Hamilton,
  * New Zealand.
  *
- * Author: Richard Sanger 
- *          
+ * Author: Richard Sanger
+ *
  * All rights reserved.
  *
- * This code has been developed by the University of Waikato WAND 
+ * This code has been developed by the University of Waikato WAND
  * research group. For further information please see http://www.wand.net.nz/
  *
@@ -36 +36 @@
  * Intel Data Plane Development Kit is a LIVE capture format.
  *
- * This format also supports writing which will write packets out to the 
- * network as a form of packet replay. This should not be confused with the 
- * RT protocol which is intended to transfer captured packet records between 
+ * This format also supports writing which will write packets out to the
+ * network as a form of packet replay. This should not be confused with the
+ * RT protocol which is intended to transfer captured packet records between
  * RT-speaking programs.
  */
+
+#define _GNU_SOURCE
 
 #include "config.h"
@@ -47 +49 @@
 #include "format_helper.h"
 #include "libtrace_arphrd.h"
+#include "hash_toeplitz.h"
 
 #ifdef HAVE_INTTYPES_H
@@ -59 +62 @@
 #include <endian.h>
 #include <string.h>
+
+#if HAVE_LIBNUMA
+#include <numa.h>
+#endif
 
 /* We can deal with any minor differences by checking the RTE VERSION
@@ -151 +158 @@
 #include <rte_mempool.h>
 #include <rte_mbuf.h>
-
-/* The default size of memory buffers to use - This is the max size of standard 
+#include <rte_launch.h>
+#include <rte_lcore.h>
+#include <rte_per_lcore.h>
+#include <rte_cycles.h>
+#include <pthread.h>
+
+/* The default size of memory buffers to use - This is the max size of standard
  * ethernet packet less the size of the MAC CHECKSUM */
 #define RX_MBUF_SIZE 1514
 
-/* The minimum number of memory buffers per queue tx or rx. Search for 
+/* The minimum number of memory buffers per queue tx or rx. Search for
  * _MIN_RING_DESC in DPDK. The largest minimum is 64 for 10GBit cards.
  */
@@ -174 +186 @@
 #define NB_TX_MBUF 1024
 
-/* The size of the PCI blacklist needs to be big enough to contain 
+/* The size of the PCI blacklist needs to be big enough to contain
  * every PCI device address (listed by lspci every bus:device.function tuple).
  */
@@ -181 +193 @@
 /* The maximum number of characters the mempool name can be */
 #define MEMPOOL_NAME_LEN 20
+
+/* For single threaded libtrace we read packets as a batch/burst
+ * this is the maximum size of said burst */
+#define BURST_SIZE 50
 
 #define MBUF(x) ((struct rte_mbuf *) x)
@@ -186 +202 @@
 #define MBUF_PKTDATA(x) ((char *) x + sizeof(struct rte_mbuf) + RTE_PKTMBUF_HEADROOM)
 #define FORMAT(x) ((struct dpdk_format_data_t*)(x->format_data))
+#define PERPKT_FORMAT(x) ((struct dpdk_per_lcore_t*)(x->format_data))
+
 #define TV_TO_NS(tv) ((uint64_t) tv.tv_sec*1000000000ull + \
-                        (uint64_t) tv.tv_usec*1000ull)
+                        (uint64_t) tv.tv_usec*1000ull)
 #define TS_TO_NS(ts) ((uint64_t) ts.tv_sec*1000000000ull + \
-                        (uint64_t) ts.tv_nsec)
+                        (uint64_t) ts.tv_nsec)
 
 #if RTE_PKTMBUF_HEADROOM != 128
 #warning "RTE_PKT_MBUF_HEADROOM is not set to the default value of 128 - " \
-         "any libtrace instance processing these packet must be have the" \
-         "same RTE_PKTMBUF_HEADROOM set"
+         "any libtrace instance processing these packet must be have the" \
+         "same RTE_PKTMBUF_HEADROOM set"
 #endif
 
 /* ~~~~~~~~~~~~~~~~~~~~~~ Advance settings ~~~~~~~~~~~~~~~~~~~~~~~~~~~~
- * THESE MAY REQUIRE MODIFICATIONS TO INTEL DPDK 
- * 
+ * THESE MAY REQUIRE MODIFICATIONS TO INTEL DPDK
+ *
  * Make sure you understand what these are doing before enabling them.
  * They might make traces incompatable with other builds etc.
- * 
+ *
  * These are also included to show how to do somethings which aren't
  * obvious in the DPDK documentation.
  */
 
-/* Print verbose messages to stdout */
+/* Print verbose messages to stderr */
 #define DEBUG 0
 
-/* Use clock_gettime() for nanosecond resolution rather than gettimeofday() 
- * only turn on if you know clock_gettime is a vsyscall on your system 
+/* Use clock_gettime() for nanosecond resolution rather than gettimeofday()
+ * only turn on if you know clock_gettime is a vsyscall on your system
  * overwise could be a large overhead. Again gettimeofday() should be
  * vsyscall also if it's not you should seriously consider updating your
@@ -218 +236 @@
 #ifdef HAVE_LIBRT
 /* You can turn this on (set to 1) to prefer clock_gettime */
-#define USE_CLOCK_GETTIME 0
+#define USE_CLOCK_GETTIME 1
 #else
 /* DONT CHANGE THIS !!! */
-#define USE_CLOCK_GETTIME 0
+#define USE_CLOCK_GETTIME 1
 #endif
 
@@ -229 +247 @@
  * hence writing out a port such as int: ring: and dpdk: assumes there
  * is no checksum and will attempt to write the checksum as part of the
- * packet 
+ * packet
  */
 #define GET_MAC_CRC_CHECKSUM 0
 
 /* This requires a modification of the pmd drivers (inside Intel DPDK)
+ * TODO this requires updating (packet sizes are wrong TS most likely also)
  */
 #define HAS_HW_TIMESTAMPS_82580 0
@@ -257 +276 @@
 };
 
+struct dpdk_per_lcore_t
+{
+        uint16_t queue_id;
+        uint8_t port;
+        uint64_t ts_last_sys; /* System timestamp of our most recent packet in nanoseconds */
+#if HAS_HW_TIMESTAMPS_82580
+        /* Timestamping only relevent to RX */
+        uint64_t ts_first_sys; /* Sytem timestamp of the first packet in nanoseconds */
+        uint32_t wrap_count; /* Number of times the NIC clock has wrapped around completely */
+#endif
+};
+
 /* Used by both input and output however some fields are not used
  * for output */
@@ -263 +294 @@
     uint8_t port; /* Always 0 we only whitelist a single port - Shared TX & RX */
     uint8_t nb_ports; /* Total number of usable ports on system should be 1 */
-    uint8_t paused; /* See paused_state */ 
+    uint8_t paused; /* See paused_state */
     uint16_t queue_id; /* Always 0 we use a single queue - Shared TX & RX */
+    uint16_t link_speed; /* Link speed 10,100,1000,10000 etc. */
     int snaplen; /* The snap length for the capture - RX only */
     /* We always have to setup both rx and tx queues even if we don't want them */
     int nb_rx_buf; /* The number of packet buffers in the rx ring */
     int nb_tx_buf; /* The number of packet buffers in the tx ring */
+    int nic_numa_node; /* The NUMA node that the NIC is attached to */
     struct rte_mempool * pktmbuf_pool; /* Our packet memory pool */
 #if DPDK_USE_BLACKLIST
@@ -275 +308 @@
 #endif
     char mempool_name[MEMPOOL_NAME_LEN]; /* The name of the mempool that we are using */
-#if HAS_HW_TIMESTAMPS_82580
-    /* Timestamping only relevent to RX */
-    uint64_t ts_first_sys; /* Sytem timestamp of the first packet in nanoseconds */
-    uint64_t ts_last_sys; /* System timestamp of our most recent packet in nanoseconds */
-    uint32_t wrap_count; /* Number of times the NIC clock has wrapped around completely */
-#endif
+    uint8_t rss_key[40]; // This is the RSS KEY
+    /* To improve performance we always batch reading packets, in a burst */
+    struct rte_mbuf* burst_pkts[BURST_SIZE];
+    int burst_size; /* The total number read in the burst */
+    int burst_offset; /* The offset we are into the burst */
+        // DPDK normally seems to have a limit of 8 queues for a given card
+        struct dpdk_per_lcore_t per_lcore[RTE_MAX_LCORE];
 };
 
@@ -288 +322 @@
 };
 
-/** 
+/**
  * A structure placed in front of the packet where we can store
  * additional information about the given packet.
@@ -294 +328 @@
  * |       rte_mbuf (pkt)     | sizeof(rte_mbuf)
  * +--------------------------+
- * |           padding        | RTE_PKTMBUF_HEADROOM-1-sizeof(dpdk_addt_hdr)
- * +--------------------------+
  * |       dpdk_addt_hdr      | sizeof(dpdk_addt_hdr)
  * +--------------------------+
- * |   sizeof(dpdk_addt_hdr)  | 1 byte
- * +--------------------------+ 
+ * |           padding        | RTE_PKTMBUF_HEADROOM-sizeof(dpdk_addt_hdr)
+ * +--------------------------+
  * *   hw_timestamp_82580     * 16 bytes Optional
  * +--------------------------+
@@ -317 +349 @@
  * We want to blacklist all devices except those on the whitelist
  * (I say list, but yes it is only the one).
- * 
+ *
  * The default behaviour of rte_pci_probe() will map every possible device
  * to its DPDK driver. The DPDK driver will take the ethernet device
  * out of the kernel (i.e. no longer /dev/ethx) and cannot be used.
- * 
- * So blacklist all devices except the one that we wish to use so that 
+ *
+ * So blacklist all devices except the one that we wish to use so that
  * the others can still be used as standard ethernet ports.
  *
@@ -336 +368 @@
 
         TAILQ_FOREACH(dev, &device_list, next) {
-        if (whitelist != NULL && whitelist->domain == dev->addr.domain 
-            && whitelist->bus == dev->addr.bus
-            && whitelist->devid == dev->addr.devid
-            && whitelist->function == dev->addr.function)
-            continue;
+        if (whitelist != NULL && whitelist->domain == dev->addr.domain
+            && whitelist->bus == dev->addr.bus
+            && whitelist->devid == dev->addr.devid
+            && whitelist->function == dev->addr.function)
+            continue;
                 if (format_data->nb_blacklist >= sizeof (format_data->blacklist)
-                                / sizeof (format_data->blacklist[0])) {
-                        printf("Warning: too many devices to blacklist consider"
-                                        " increasing BLACK_LIST_SIZE");
+                                / sizeof (format_data->blacklist[0])) {
+                        fprintf(stderr, "Warning: too many devices to blacklist consider"
+                                        " increasing BLACK_LIST_SIZE");
                         break;
                 }
@@ -360 +392 @@
         char pci_str[20] = {0};
         snprintf(pci_str, sizeof(pci_str), PCI_PRI_FMT,
-                 whitelist->domain,
-                 whitelist->bus,
-                 whitelist->devid,
-                 whitelist->function);
+                 whitelist->domain,
+                 whitelist->bus,
+                 whitelist->devid,
+                 whitelist->function);
         if (rte_eal_devargs_add(RTE_DEVTYPE_WHITELISTED_PCI, pci_str) < 0) {
                 return -1;
@@ -375 +407 @@
  * Fills in addr, note core is optional and is unchanged if
  * a value for it is not provided.
- * 
+ *
  * i.e. ./libtrace dpdk:0:1:0.0 -> 0:1:0.0
- * or ./libtrace dpdk:0:1:0.1-2 -> 0:1:0.1 (Using CPU core #2) 
+ * or ./libtrace dpdk:0:1:0.1-2 -> 0:1:0.1 (Using CPU core #2)
  */
 static int parse_pciaddr(char * str, struct rte_pci_addr * addr, long * core) {
@@ -391 +423 @@
 }
 
+/**
+ * Convert a pci address to the numa node it is
+ * connected to.
+ *
+ * This checks /sys/bus/pci/devices/XXXX:XX:XX.X/numa_node
+ * so we can call it before DPDK
+ *
+ * @return -1 if unknown otherwise a number 0 or higher of the numa node
+ */
+static int pci_to_numa(struct rte_pci_addr * dev_addr) {
+        char path[50] = {0};
+        FILE *file;
+
+        /* Read from the system */
+        snprintf(path, sizeof(path), "/sys/bus/pci/devices/"PCI_PRI_FMT"/numa_node",
+                 dev_addr->domain,
+                 dev_addr->bus,
+                 dev_addr->devid,
+                 dev_addr->function);
+
+        if((file = fopen(path, "r")) != NULL) {
+                int numa_node = -1;
+                fscanf(file, "%d", &numa_node);
+                fclose(file);
+                return numa_node;
+        }
+        return -1;
+}
+
 #if DEBUG
 /* For debugging */
@@ -399 +460 @@
 
     if (nb_cpu <= 0) {
-        perror("sysconf(_SC_NPROCESSORS_ONLN) failed. Falling back to the first core.");
-        nb_cpu = 1; /* fallback to just 1 core */
+        perror("sysconf(_SC_NPROCESSORS_ONLN) failed. Falling back to the first core.");
+        nb_cpu = 1; /* fallback to just 1 core */
     }
     if (nb_cpu > RTE_MAX_LCORE)
-        nb_cpu = RTE_MAX_LCORE;
+        nb_cpu = RTE_MAX_LCORE;
 
     global_config = rte_eal_get_configuration();
 
     if (global_config != NULL) {
-        int i;
-        fprintf(stderr, "Intel DPDK setup\n"
-               "---Version      : %s\n"
-               "---Master LCore : %"PRIu32"\n"
-               "---LCore Count  : %"PRIu32"\n",
-               rte_version(),
-               global_config->master_lcore, global_config->lcore_count);
-
-        for (i = 0 ; i < nb_cpu; i++) {
-            fprintf(stderr, "   ---Core %d : %s\n", i,
-                   global_config->lcore_role[i] == ROLE_RTE ? "on" : "off");
-        }
-
-        const char * proc_type;
-        switch (global_config->process_type) {
-            case RTE_PROC_AUTO:
-                proc_type = "auto";
-                break;
-            case RTE_PROC_PRIMARY:
-                proc_type = "primary";
-                break;
-            case RTE_PROC_SECONDARY:
-                proc_type = "secondary";
-                break;
-            case RTE_PROC_INVALID:
-                proc_type = "invalid";
-                break;
-            default:
-                proc_type = "something worse than invalid!!";
-        }
-        fprintf(stderr, "---Process Type : %s\n", proc_type);
-    }
-
-}
-#endif
+        int i;
+        fprintf(stderr, "Intel DPDK setup\n"
+               "---Version      : %s\n"
+               "---Master LCore : %"PRIu32"\n"
+               "---LCore Count  : %"PRIu32"\n",
+               rte_version(),
+               global_config->master_lcore, global_config->lcore_count);
+
+        for (i = 0 ; i < nb_cpu; i++) {
+            fprintf(stderr, "   ---Core %d : %s\n", i,
+                   global_config->lcore_role[i] == ROLE_RTE ? "on" : "off");
+        }
+
+        const char * proc_type;
+        switch (global_config->process_type) {
+            case RTE_PROC_AUTO:
+                proc_type = "auto";
+                break;
+            case RTE_PROC_PRIMARY:
+                proc_type = "primary";
+                break;
+            case RTE_PROC_SECONDARY:
+                proc_type = "secondary";
+                break;
+            case RTE_PROC_INVALID:
+                proc_type = "invalid";
+                break;
+            default:
+                proc_type = "something worse than invalid!!";
+        }
+        fprintf(stderr, "---Process Type : %s\n", proc_type);
+    }
+
+}
+#endif
+
+/**
+ * Expects to be called from the master lcore and moves it to the given dpdk id
+ * @param core (zero indexed) If core is on the physical system affinity is bound otherwise
+ *               affinity is set to all cores. Must be less than RTE_MAX_LCORE
+ *               and not already in use.
+ * @return 0 is successful otherwise -1 on error.
+ */
+static inline int dpdk_move_master_lcore(size_t core) {
+    struct rte_config *cfg = rte_eal_get_configuration();
+    cpu_set_t cpuset;
+    int i;
+
+    assert (core < RTE_MAX_LCORE);
+    assert (rte_get_master_lcore() == rte_lcore_id());
+
+    if (core == rte_lcore_id())
+        return 0;
+
+    // Make sure we are not overwriting someone else
+    assert(!rte_lcore_is_enabled(core));
+
+    // Move the core
+    cfg->lcore_role[rte_lcore_id()] = ROLE_OFF;
+    cfg->lcore_role[core] = ROLE_RTE;
+    lcore_config[core].thread_id = lcore_config[rte_lcore_id()].thread_id;
+    rte_eal_get_configuration()->master_lcore = core;
+    RTE_PER_LCORE(_lcore_id) = core;
+
+    // Now change the affinity
+    CPU_ZERO(&cpuset);
+
+    if (lcore_config[core].detected) {
+        CPU_SET(core, &cpuset);
+    } else {
+        for (i = 0; i < RTE_MAX_LCORE; ++i) {
+            if (lcore_config[i].detected)
+                CPU_SET(i, &cpuset);
+        }
+    }
+
+    i = pthread_setaffinity_np(pthread_self(), sizeof(cpuset), &cpuset);
+    if (i != 0) {
+        // TODO proper libtrace style error here!!
+        fprintf(stderr, "pthread_setaffinity_np failed\n");
+        return -1;
+    }
+    return 0;
+}
+
 
 /**
@@ -473 +584 @@
 
 static inline int dpdk_init_environment(char * uridata, struct dpdk_format_data_t * format_data,
-                                        char * err, int errlen) {
+                                        char * err, int errlen) {
     int ret; /* Returned error codes */
     struct rte_pci_addr use_addr; /* The only address that we don't blacklist */
@@ -480 +591 @@
     long nb_cpu; /* The number of CPUs in the system */
     long my_cpu; /* The CPU number we want to bind to */
+    int i;
+    struct rte_config *cfg = rte_eal_get_configuration();
         struct saved_getopts save_opts;
-    
+
 #if DEBUG
     rte_set_log_level(RTE_LOG_DEBUG);
@@ -489 +602 @@
     /*
      * Using unique file prefixes mean separate memory is used, unlinking
-     * the two processes. However be careful we still cannot access a 
+     * the two processes. However be careful we still cannot access a
+     * port that already in use.
+     *
+     * Using unique file prefixes mean separate memory is used, unlinking
+     * the two processes. However be careful we still cannot access a
      * port that already in use.
      */
@@ -510 +627 @@
     /* This initialises the Environment Abstraction Layer (EAL)
      * If we had slave workers these are put into WAITING state
-     * 
+     *
      * Basically binds this thread to a fixed core, which we choose as
      * the last core on the machine (assuming fewer interrupts mapped here).
@@ -521 +638 @@
      */
 
-    /* Get the number of cpu cores in the system and use the last core */
+    /* Get the number of cpu cores in the system and use the last core
+     * on the correct numa node */
     nb_cpu = sysconf(_SC_NPROCESSORS_ONLN);
     if (nb_cpu <= 0) {
-        perror("sysconf(_SC_NPROCESSORS_ONLN) failed. Falling back to the first core.");
-        nb_cpu = 1; /* fallback to the first core */
+        perror("sysconf(_SC_NPROCESSORS_ONLN) failed. Falling back to the first core.");
+        nb_cpu = 1; /* fallback to the first core */
     }
     if (nb_cpu > RTE_MAX_LCORE)
-        nb_cpu = RTE_MAX_LCORE;
-
-    my_cpu = nb_cpu;
-    /* This allows the user to specify the core - we would try to do this 
+        nb_cpu = RTE_MAX_LCORE;
+
+    my_cpu = -1;
+    /* This allows the user to specify the core - we would try to do this
      * automatically but it's hard to tell that this is secondary
-     * before running rte_eal_init(...). Currently we are limited to 1 
+     * before running rte_eal_init(...). Currently we are limited to 1
      * instance per core due to the way memory is allocated. */
     if (parse_pciaddr(uridata, &use_addr, &my_cpu) != 0) {
-        snprintf(err, errlen, "Failed to parse URI");
-        return -1;
-    }
+        snprintf(err, errlen, "Failed to parse URI");
+        return -1;
+    }
+
+#if HAVE_LIBNUMA
+        format_data->nic_numa_node = pci_to_numa(&use_addr);
+        if (my_cpu < 0) {
+                /* If we can assign to a core on the same numa node */
+                fprintf(stderr, "Using pci card on numa_node%d\n", format_data->nic_numa_node);
+                if(format_data->nic_numa_node >= 0) {
+                        int max_node_cpu = -1;
+                        struct bitmask *mask = numa_allocate_cpumask();
+                        assert(mask);
+                        numa_node_to_cpus(format_data->nic_numa_node, mask);
+                        for (i = 0 ; i < nb_cpu; ++i) {
+                                if (numa_bitmask_isbitset(mask,i))
+                                        max_node_cpu = i+1;
+                        }
+                        my_cpu = max_node_cpu;
+                }
+        }
+#endif
+        if (my_cpu < 0) {
+                my_cpu = nb_cpu;
+        }
+
 
     snprintf(format_data->mempool_name, MEMPOOL_NAME_LEN,
-                "libtrace_pool_%"PRIu32, (uint32_t) nb_cpu);
+                "libtrace_pool_%"PRIu32, (uint32_t) nb_cpu);
 
     if (!(my_cpu > 0 && my_cpu <= nb_cpu)) {
-        snprintf(err, errlen, 
-          "Intel DPDK - User defined a bad CPU number %"PRIu32" must be"
-          " between 1 and %"PRIu32, (uint32_t) my_cpu, (uint32_t) nb_cpu);
-        return -1;
-    }
-
-    /* Make our mask */
-    snprintf(cpu_number, sizeof(cpu_number), "%x", 0x1 << (my_cpu - 1));
+        snprintf(err, errlen,
+          "Intel DPDK - User defined a bad CPU number %"PRIu32" must be"
+          " between 1 and %"PRIu32, (uint32_t) my_cpu, (uint32_t) nb_cpu);
+        return -1;
+    }
+
+    /* Make our mask with all cores turned on this is so that DPDK to gets CPU
+       info older versions */
+    snprintf(cpu_number, sizeof(cpu_number), "%x", ~(UINT32_MAX<<MIN(31, nb_cpu)));
+    //snprintf(cpu_number, sizeof(cpu_number), "%x", 0x1 << (my_cpu - 1));
 
 #if !DPDK_USE_BLACKLIST
     /* Black list all ports besides the one that we want to use */
-    if ((ret = whitelist_device(format_data, &use_addr)) < 0) {
-        snprintf(err, errlen, "Intel DPDK - Whitelisting PCI device failed,"
-                 " are you sure the address is correct?: %s", strerror(-ret));
-        return -1;
-    }
+        if ((ret = whitelist_device(format_data, &use_addr)) < 0) {
+                snprintf(err, errlen, "Intel DPDK - Whitelisting PCI device failed,"
+                         " are you sure the address is correct?: %s", strerror(-ret));
+                return -1;
+        }
 #endif
 
         /* Give the memory map a unique name */
         snprintf(mem_map, sizeof(mem_map), "libtrace-%d", (int) getpid());
-    /* rte_eal_init it makes a call to getopt so we need to reset the 
+    /* rte_eal_init it makes a call to getopt so we need to reset the
      * global optind variable of getopt otherwise this fails */
         save_getopts(&save_opts);
     optind = 1;
     if ((ret = rte_eal_init(argc, argv)) < 0) {
-        snprintf(err, errlen, 
-          "Intel DPDK - Initialisation of EAL failed: %s", strerror(-ret));
-        return -1;
+        snprintf(err, errlen,
+          "Intel DPDK - Initialisation of EAL failed: %s", strerror(-ret));
+        return -1;
     }
         restore_getopts(&save_opts);
+    // These are still running but will never do anything with DPDK v1.7 we
+    // should remove this XXX in the future
+    for(i = 0; i < RTE_MAX_LCORE; ++i) {
+            if (rte_lcore_is_enabled(i) && i != (int) rte_get_master_lcore()) {
+            cfg->lcore_role[i] = ROLE_OFF;
+            cfg->lcore_count--;
+        }
+    }
+    // Only the master should be running
+    assert(cfg->lcore_count == 1);
+
+    dpdk_move_master_lcore(my_cpu-1);
 
 #if DEBUG
@@ -584 +739 @@
      */
     if ((ret = rte_pmd_init_all()) < 0) {
-        snprintf(err, errlen, 
-          "Intel DPDK - rte_pmd_init_all failed: %s", strerror(-ret));
-        return -1;
+        snprintf(err, errlen,
+          "Intel DPDK - rte_pmd_init_all failed: %s", strerror(-ret));
+        return -1;
     }
 #endif
@@ -594 +749 @@
         if ((ret = blacklist_devices(format_data, &use_addr)) < 0) {
                 snprintf(err, errlen, "Intel DPDK - Whitelisting PCI device failed,"
-                         " are you sure the address is correct?: %s", strerror(-ret));
+                         " are you sure the address is correct?: %s", strerror(-ret));
                 return -1;
         }
@@ -602 +757 @@
     /* This loads DPDK drivers against all ports that are not blacklisted */
         if ((ret = rte_eal_pci_probe()) < 0) {
-        snprintf(err, errlen, 
-            "Intel DPDK - rte_eal_pci_probe failed: %s", strerror(-ret));
-        return -1;
+        snprintf(err, errlen,
+            "Intel DPDK - rte_eal_pci_probe failed: %s", strerror(-ret));
+        return -1;
     }
 #endif
@@ -611 +766 @@
 
     if (format_data->nb_ports != 1) {
-        snprintf(err, errlen, 
-            "Intel DPDK - rte_eth_dev_count returned %d but it should be 1",
-            format_data->nb_ports);
-        return -1;
-    }
+        snprintf(err, errlen,
+            "Intel DPDK - rte_eth_dev_count returned %d but it should be 1",
+            format_data->nb_ports);
+        return -1;
+    }
+
+    struct rte_eth_dev_info dev_info;
+    rte_eth_dev_info_get(0, &dev_info);
+    fprintf(stderr, "Device port=0\n\tmin_rx_bufsize=%d\n\tmax_rx_pktlen=%d\n\tmax rx queues=%d\n\tmax tx queues=%d",
+                (int) dev_info.min_rx_bufsize, (int) dev_info.max_rx_pktlen, (int) dev_info.max_rx_queues, (int) dev_info.max_tx_queues);
 
     return 0;
@@ -623 +783 @@
     char err[500];
     err[0] = 0;
-    
+
     libtrace->format_data = (struct dpdk_format_data_t *)
-                            malloc(sizeof(struct dpdk_format_data_t));
+                            malloc(sizeof(struct dpdk_format_data_t));
     FORMAT(libtrace)->port = 0; /* Always assume 1 port loaded */
     FORMAT(libtrace)->queue_id = 0; /* Single queue per port */
@@ -632 +792 @@
     FORMAT(libtrace)->nb_rx_buf = NB_RX_MBUF;
     FORMAT(libtrace)->nb_tx_buf = MIN_NB_BUF;
+    FORMAT(libtrace)->nic_numa_node = -1;
     FORMAT(libtrace)->promisc = -1;
     FORMAT(libtrace)->pktmbuf_pool = NULL;
@@ -639 +800 @@
     FORMAT(libtrace)->paused = DPDK_NEVER_STARTED;
     FORMAT(libtrace)->mempool_name[0] = 0;
-#if HAS_HW_TIMESTAMPS_82580
-    FORMAT(libtrace)->ts_first_sys = 0;
-    FORMAT(libtrace)->ts_last_sys = 0;
-    FORMAT(libtrace)->wrap_count = 0;
-#endif
+    memset(FORMAT(libtrace)->burst_pkts, 0, sizeof(FORMAT(libtrace)->burst_pkts[0]) * BURST_SIZE);
+    FORMAT(libtrace)->burst_size = 0;
+    FORMAT(libtrace)->burst_offset = 0;
 
     if (dpdk_init_environment(libtrace->uridata, FORMAT(libtrace), err, sizeof(err)) != 0) {
-        trace_set_err(libtrace, TRACE_ERR_INIT_FAILED, "%s", err);
-        free(libtrace->format_data);
-        libtrace->format_data = NULL;
-        return -1;
+        trace_set_err(libtrace, TRACE_ERR_INIT_FAILED, "%s", err);
+        free(libtrace->format_data);
+        libtrace->format_data = NULL;
+        return -1;
     }
     return 0;
@@ -658 +817 @@
     char err[500];
     err[0] = 0;
-    
+
     libtrace->format_data = (struct dpdk_format_data_t *)
-                            malloc(sizeof(struct dpdk_format_data_t));
+                            malloc(sizeof(struct dpdk_format_data_t));
     FORMAT(libtrace)->port = 0; /* Always assume 1 port loaded */
     FORMAT(libtrace)->queue_id = 0; /* Single queue per port */
@@ -667 +826 @@
     FORMAT(libtrace)->nb_rx_buf = MIN_NB_BUF;
     FORMAT(libtrace)->nb_tx_buf = NB_TX_MBUF;
+    FORMAT(libtrace)->nic_numa_node = -1;
     FORMAT(libtrace)->promisc = -1;
     FORMAT(libtrace)->pktmbuf_pool = NULL;
@@ -674 +834 @@
     FORMAT(libtrace)->paused = DPDK_NEVER_STARTED;
     FORMAT(libtrace)->mempool_name[0] = 0;
-#if HAS_HW_TIMESTAMPS_82580
-    FORMAT(libtrace)->ts_first_sys = 0;
-    FORMAT(libtrace)->ts_last_sys = 0;
-    FORMAT(libtrace)->wrap_count = 0;
-#endif
+    memset(FORMAT(libtrace)->burst_pkts, 0, sizeof(FORMAT(libtrace)->burst_pkts[0]) * BURST_SIZE);
+    FORMAT(libtrace)->burst_size = 0;
+    FORMAT(libtrace)->burst_offset = 0;
 
     if (dpdk_init_environment(libtrace->uridata, FORMAT(libtrace), err, sizeof(err)) != 0) {
-        trace_set_err_out(libtrace, TRACE_ERR_INIT_FAILED, "%s", err);
-        free(libtrace->format_data);
-        libtrace->format_data = NULL;
-        return -1;
+        trace_set_err_out(libtrace, TRACE_ERR_INIT_FAILED, "%s", err);
+        free(libtrace->format_data);
+        libtrace->format_data = NULL;
+        return -1;
     }
     return 0;
 };
 
+static int dpdk_pconfig_input (libtrace_t *libtrace,
+                                trace_parallel_option_t option,
+                                void *data) {
+        switch (option) {
+                case TRACE_OPTION_SET_HASHER:
+                        switch (*((enum hasher_types *) data))
+                        {
+                                case HASHER_BALANCE:
+                                case HASHER_UNIDIRECTIONAL:
+                                        toeplitz_create_unikey(FORMAT(libtrace)->rss_key);
+                                        return 0;
+                                case HASHER_BIDIRECTIONAL:
+                                        toeplitz_create_bikey(FORMAT(libtrace)->rss_key);
+                                        return 0;
+                                case HASHER_HARDWARE:
+                                case HASHER_CUSTOM:
+                                        // We don't support these
+                                        return -1;
+                        }
+        break;
+        }
+        return -1;
+}
 /**
  * Note here snaplen excludes the MAC checksum. Packets over
  * the requested snaplen will be dropped. (Excluding MAC checksum)
- * 
+ *
  * I.e the maximum size of a standard ethernet packet is 1518 (Including MAC checksum)
  * So to allow packets upto 1518 this would be set to 1514 and if GET_MAC_CRC_CHECKSUM
  * is set the maximum size of the returned packet would be 1518 otherwise
  * 1514 would be the largest size possibly returned.
- * 
+ *
  */
 static int dpdk_config_input (libtrace_t *libtrace,
-                                        trace_option_t option,
-                                        void *data) {
+                                        trace_option_t option,
+                                        void *data) {
     switch (option) {
-        case TRACE_OPTION_SNAPLEN:
-            /* Only support changing snaplen before a call to start is 
-             * made */
-            if (FORMAT(libtrace)->paused == DPDK_NEVER_STARTED)
-                FORMAT(libtrace)->snaplen=*(int*)data;
-            else
-                return -1;
-            return 0;
+        case TRACE_OPTION_SNAPLEN:
+            /* Only support changing snaplen before a call to start is
+             * made */
+            if (FORMAT(libtrace)->paused == DPDK_NEVER_STARTED)
+                FORMAT(libtrace)->snaplen=*(int*)data;
+            else
+                return -1;
+            return 0;
                 case TRACE_OPTION_PROMISC:
                         FORMAT(libtrace)->promisc=*(int*)data;
-            return 0;
-        case TRACE_OPTION_FILTER:
-            /* TODO filtering */
-            break;
-        case TRACE_OPTION_META_FREQ:
-            break;
-        case TRACE_OPTION_EVENT_REALTIME:
-            break;
-        /* Avoid default: so that future options will cause a warning
-         * here to remind us to implement it, or flag it as
-         * unimplementable
-         */
+            return 0;
+        case TRACE_OPTION_FILTER:
+            /* TODO filtering */
+            break;
+        case TRACE_OPTION_META_FREQ:
+            break;
+        case TRACE_OPTION_EVENT_REALTIME:
+            break;
+        /* Avoid default: so that future options will cause a warning
+         * here to remind us to implement it, or flag it as
+         * unimplementable
+         */
     }
 
@@ -734 +915 @@
 /* Can set jumbo frames/ or limit the size of a frame by setting both
  * max_rx_pkt_len and jumbo_frame. This can be limited to less than
- * 
+ *
  */
 static struct rte_eth_conf port_conf = {
         .rxmode = {
+                .mq_mode = ETH_RSS,
                 .split_hdr_size = 0,
                 .header_split   = 0, /**< Header Split disabled */
@@ -743 +925 @@
                 .hw_vlan_filter = 0, /**< VLAN filtering disabled */
                 .jumbo_frame    = 0, /**< Jumbo Frame Support disabled */
-        .max_rx_pkt_len = 0, /**< Max frame Size if Jumbo enabled */
+                .max_rx_pkt_len = 0, /**< Max frame Size if Jumbo enabled */
 #if GET_MAC_CRC_CHECKSUM
 /* So it appears that if hw_strip_crc is turned off the driver will still
@@ -756 +938 @@
  * always cut off the checksum in the future
  */
-        .hw_strip_crc   = 1, /**< CRC stripped by hardware */
+                .hw_strip_crc   = 1, /**< CRC stripped by hardware */
 #endif
         },
@@ -762 +944 @@
                 .mq_mode = ETH_DCB_NONE,
         },
+        .rx_adv_conf = {
+                .rss_conf = {
+                        // .rss_key = &rss_key, // We set this per format
+                        .rss_hf = ETH_RSS_IPV4_UDP | ETH_RSS_IPV6 | ETH_RSS_IPV4 | ETH_RSS_IPV4_TCP | ETH_RSS_IPV6_TCP | ETH_RSS_IPV6_UDP,
+                },
+        },
+        .intr_conf = {
+                .lsc = 1
+        }
 };
 
@@ -776 +967 @@
 static const struct rte_eth_txconf tx_conf = {
         .tx_thresh = {
-        /**
-         * TX_PTHRESH prefetch
-         * Set on the NIC, if the number of unprocessed descriptors to queued on
-         * the card fall below this try grab at least hthresh more unprocessed
-         * descriptors.
-         */
+        /**
+         * TX_PTHRESH prefetch
+         * Set on the NIC, if the number of unprocessed descriptors to queued on
+         * the card fall below this try grab at least hthresh more unprocessed
+         * descriptors.
+         */
                 .pthresh = 36,
 
-        /* TX_HTHRESH host
-         * Set on the NIC, the batch size to prefetch unprocessed tx descriptors.
-         */
+        /* TX_HTHRESH host
+         * Set on the NIC, the batch size to prefetch unprocessed tx descriptors.
+         */
                 .hthresh = 0,
-        
-        /* TX_WTHRESH writeback
-         * Set on the NIC, the number of sent descriptors before writing back
-         * status to confirm the transmission. This is done more efficiently as
-         * a bulk DMA-transfer rather than writing one at a time.
-         * Similar to tx_free_thresh however this is applied to the NIC, where
-         * as tx_free_thresh is when DPDK will check these. This is extended
-         * upon by tx_rs_thresh (10Gbit cards) which doesn't write all
-         * descriptors rather only every n'th item, reducing DMA memory bandwidth.
-         */
+
+        /* TX_WTHRESH writeback
+         * Set on the NIC, the number of sent descriptors before writing back
+         * status to confirm the transmission. This is done more efficiently as
+         * a bulk DMA-transfer rather than writing one at a time.
+         * Similar to tx_free_thresh however this is applied to the NIC, where
+         * as tx_free_thresh is when DPDK will check these. This is extended
+         * upon by tx_rs_thresh (10Gbit cards) which doesn't write all
+         * descriptors rather only every n'th item, reducing DMA memory bandwidth.
+         */
                 .wthresh = 4,
         },
@@ -808 +999 @@
 
     /* This is the Report Status threshold, used by 10Gbit cards,
-     * This signals the card to only write back status (such as 
+     * This signals the card to only write back status (such as
      * transmission successful) after this minimum number of transmit
      * descriptors are seen. The default is 32 (if set to 0) however if set
@@ -817 +1008 @@
 };
 
+/**
+ * A callback for a link state change (LSC).
+ *
+ * Packets may be received before this notification. In fact the DPDK IGXBE
+ * driver likes to put a delay upto 5sec before sending this.
+ *
+ * We use this to ensure the link speed is correct for our timestamp
+ * calculations. Because packets might be received before the link up we still
+ * update this when the packet is received.
+ *
+ * @param port The DPDK port
+ * @param event The TYPE of event (expected to be RTE_ETH_EVENT_INTR_LSC)
+ * @param cb_arg The dpdk_format_data_t structure associated with the format
+ */
+static void dpdk_lsc_callback(uint8_t port, enum rte_eth_event_type event,
+                              void *cb_arg) {
+        struct dpdk_format_data_t * format_data = cb_arg;
+        struct rte_eth_link link_info;
+        assert(event == RTE_ETH_EVENT_INTR_LSC);
+        assert(port == format_data->port);
+
+        rte_eth_link_get_nowait(port, &link_info);
+
+        if (link_info.link_status)
+                format_data->link_speed = link_info.link_speed;
+        else
+                format_data->link_speed = 0;
+
+#if DEBUG
+        fprintf(stderr, "LSC - link status is %s %s speed=%d\n",
+                link_info.link_status ? "up" : "down",
+                (link_info.link_duplex == ETH_LINK_FULL_DUPLEX) ?
+                                          "full-duplex" : "half-duplex",
+                (int) link_info.link_speed);
+#endif
+
+        /* Turns out DPDK drivers might not come back up if the link speed
+         * changes. So we reset the autoneg procedure. This is very unsafe
+         * we have have threads reading packets and we stop the port. */
+#if 0
+        if (!link_info.link_status) {
+                int ret;
+                rte_eth_dev_stop(port);
+                ret = rte_eth_dev_start(port);
+                if (ret < 0) {
+                        fprintf(stderr, "Resetting the DPDK port failed : %s\n",
+                                strerror(-ret));
+                }
+        }
+#endif
+}
+
 /* Attach memory to the port and start the port or restart the port.
  */
@@ -822 +1065 @@
     int ret; /* Check return values for errors */
     struct rte_eth_link link_info; /* Wait for link */
-    
+    unsigned cpu_numa_node = rte_lcore_to_socket_id(rte_lcore_id());
+
     /* Already started */
     if (format_data->paused == DPDK_RUNNING)
-        return 0;
-
-    /* First time started we need to alloc our memory, doing this here 
+        return 0;
+
+    /* First time started we need to alloc our memory, doing this here
      * rather than in environment setup because we don't have snaplen then */
     if (format_data->paused == DPDK_NEVER_STARTED) {
-        if (format_data->snaplen == 0) {
-            format_data->snaplen = RX_MBUF_SIZE;
-            port_conf.rxmode.jumbo_frame = 0;
-            port_conf.rxmode.max_rx_pkt_len = 0;
-        } else {
-            /* Use jumbo frames */
-            port_conf.rxmode.jumbo_frame = 1;
-            port_conf.rxmode.max_rx_pkt_len = format_data->snaplen;
-        }
-
-        /* This is additional overhead so make sure we allow space for this */
+        if (format_data->snaplen == 0) {
+            format_data->snaplen = RX_MBUF_SIZE;
+            port_conf.rxmode.jumbo_frame = 0;
+            port_conf.rxmode.max_rx_pkt_len = 0;
+        } else {
+            /* Use jumbo frames */
+            port_conf.rxmode.jumbo_frame = 1;
+            port_conf.rxmode.max_rx_pkt_len = format_data->snaplen;
+        }
+
+        /* This is additional overhead so make sure we allow space for this */
 #if GET_MAC_CRC_CHECKSUM
-        format_data->snaplen += ETHER_CRC_LEN;
+        format_data->snaplen += ETHER_CRC_LEN;
 #endif
 #if HAS_HW_TIMESTAMPS_82580
-        format_data->snaplen += sizeof(struct hw_timestamp_82580);
-#endif
-
-        /* Create the mbuf pool, which is the place our packets are allocated
-         * from - TODO figure out if there is is a free function (I cannot see one) 
-         * NOTE: RX queue requires nb_packets + 1 otherwise it fails to 
-         * allocate however that extra 1 packet is not used. 
-         * (I assume <= vs < error some where in DPDK code)
-         * TX requires nb_tx_buffers + 1 in the case the queue is full 
-         * so that will fill the new buffer and wait until slots in the
-         * ring become available.
-         */
+        format_data->snaplen += sizeof(struct hw_timestamp_82580);
+#endif
+
+        /* Create the mbuf pool, which is the place our packets are allocated
+         * from - TODO figure out if there is is a free function (I cannot see one)
+         * NOTE: RX queue requires nb_packets + 1 otherwise it fails to
+         * allocate however that extra 1 packet is not used.
+         * (I assume <= vs < error some where in DPDK code)
+         * TX requires nb_tx_buffers + 1 in the case the queue is full
+         * so that will fill the new buffer and wait until slots in the
+         * ring become available.
+         */
 #if DEBUG
     fprintf(stderr, "Creating mempool named %s\n", format_data->mempool_name);
 #endif
-        format_data->pktmbuf_pool =
-            rte_mempool_create(format_data->mempool_name,
-                       format_data->nb_rx_buf + format_data->nb_tx_buf + 1,
-                       format_data->snaplen + sizeof(struct rte_mbuf) 
-                                        + RTE_PKTMBUF_HEADROOM,
-                       8, sizeof(struct rte_pktmbuf_pool_private),
-                       rte_pktmbuf_pool_init, NULL, rte_pktmbuf_init, NULL,
-                       rte_socket_id(), MEMPOOL_F_SP_PUT | MEMPOOL_F_SC_GET);
-
-        if (format_data->pktmbuf_pool == NULL) {
-            snprintf(err, errlen, "Intel DPDK - Initialisation of mbuf "
-                        "pool failed: %s", strerror(rte_errno));
-            return -1;
-        }
-    }
-    
+    format_data->pktmbuf_pool =
+            rte_mempool_create(format_data->mempool_name,
+                       (format_data->nb_rx_buf + format_data->nb_tx_buf + 1),
+                       format_data->snaplen + sizeof(struct rte_mbuf)
+                                        + RTE_PKTMBUF_HEADROOM,
+                       128, sizeof(struct rte_pktmbuf_pool_private),
+                       rte_pktmbuf_pool_init, NULL, rte_pktmbuf_init, NULL,
+                       cpu_numa_node, 0/*MEMPOOL_F_SP_PUT | MEMPOOL_F_SC_GET*/);
+
+    if (format_data->pktmbuf_pool == NULL) {
+            snprintf(err, errlen, "Intel DPDK - Initialisation of mbuf"
+                        "pool failed: %s NODE %u", strerror(rte_errno), cpu_numa_node);
+            return -1;
+        }
+    }
+
     /* ----------- Now do the setup for the port mapping ------------ */
-    /* Order of calls must be 
+    /* Order of calls must be
      * rte_eth_dev_configure()
      * rte_eth_tx_queue_setup()
@@ -884 +1128 @@
      * other rte_eth calls
      */
-    
+
+
+    port_conf.rx_adv_conf.rss_conf.rss_key = format_data->rss_key;
+
     /* This must be called first before another *eth* function
      * 1 rx, 1 tx queue, port_conf sets checksum stripping etc */
     ret = rte_eth_dev_configure(format_data->port, 1, 1, &port_conf);
     if (ret < 0) {
-        snprintf(err, errlen, "Intel DPDK - Cannot configure device port"
-                            " %"PRIu8" : %s", format_data->port,
-                            strerror(-ret));
-        return -1;
+        snprintf(err, errlen, "Intel DPDK - Cannot configure device port"
+                            " %"PRIu8" : %s", format_data->port,
+                            strerror(-ret));
+        return -1;
     }
     /* Initialise the TX queue a minimum value if using this port for
@@ -898 +1145 @@
      */
     ret = rte_eth_tx_queue_setup(format_data->port, format_data->queue_id,
-                        format_data->nb_tx_buf, rte_socket_id(), &tx_conf);
+                        format_data->nb_tx_buf, SOCKET_ID_ANY, &tx_conf);
     if (ret < 0) {
-        snprintf(err, errlen, "Intel DPDK - Cannot configure TX queue on port"
-                            " %"PRIu8" : %s", format_data->port,
-                            strerror(-ret));
-        return -1;
+        snprintf(err, errlen, "Intel DPDK - Cannot configure TX queue on port"
+                            " %"PRIu8" : %s", format_data->port,
+                            strerror(-ret));
+        return -1;
     }
     /* Initialise the RX queue with some packets from memory */
     ret = rte_eth_rx_queue_setup(format_data->port, format_data->queue_id,
-                            format_data->nb_rx_buf, rte_socket_id(),
-                            &rx_conf, format_data->pktmbuf_pool);
+                                 format_data->nb_rx_buf, cpu_numa_node,
+                                 &rx_conf, format_data->pktmbuf_pool);
     if (ret < 0) {
-        snprintf(err, errlen, "Intel DPDK - Cannot configure RX queue on port"
-                    " %"PRIu8" : %s", format_data->port,
-                    strerror(-ret));
-        return -1;
-    }
-    
+        snprintf(err, errlen, "Intel DPDK - Cannot configure RX queue on port"
+                    " %"PRIu8" : %s", format_data->port,
+                    strerror(-ret));
+        return -1;
+    }
+
     /* Start device */
     ret = rte_eth_dev_start(format_data->port);
     if (ret < 0) {
-        snprintf(err, errlen, "Intel DPDK - rte_eth_dev_start failed : %s",
-                    strerror(-ret));
-        return -1;
+        snprintf(err, errlen, "Intel DPDK - rte_eth_dev_start failed : %s",
+                    strerror(-ret));
+        return -1;
     }
 
     /* Default promiscuous to on */
     if (format_data->promisc == -1)
-        format_data->promisc = 1;
-    
+        format_data->promisc = 1;
+
     if (format_data->promisc == 1)
-        rte_eth_promiscuous_enable(format_data->port);
+        rte_eth_promiscuous_enable(format_data->port);
     else
-        rte_eth_promiscuous_disable(format_data->port);
-    
-    /* Wait for the link to come up */
-    rte_eth_link_get(format_data->port, &link_info);
+        rte_eth_promiscuous_disable(format_data->port);
+
+        /* Register a callback for link state changes */
+        ret = rte_eth_dev_callback_register(format_data->port,
+                                            RTE_ETH_EVENT_INTR_LSC,
+                                            dpdk_lsc_callback,
+                                            format_data);
+        /* If this fails it is not a show stopper */
+#if DEBUG
+        fprintf(stderr, "rte_eth_dev_callback_register failed %d : %s\n",
+                ret, strerror(-ret));
+#endif
+
+    /* Get the current link status */
+    rte_eth_link_get_nowait(format_data->port, &link_info);
+    format_data->link_speed = link_info.link_speed;
 #if DEBUG
     fprintf(stderr, "Link status is %d %d %d\n", (int) link_info.link_status,
-            (int) link_info.link_duplex, (int) link_info.link_speed);
-#endif
-
+            (int) link_info.link_duplex, (int) link_info.link_speed);
+#endif
     /* We have now successfully started/unpaused */
     format_data->paused = DPDK_RUNNING;
-    
+
+    return 0;
+}
+
+/* Attach memory to the port and start (or restart) the port/s.
+ */
+static int dpdk_start_port_queues (struct dpdk_format_data_t *format_data, char *err, int errlen, uint16_t rx_queues) {
+    int ret, i; /* Check return values for errors */
+    struct rte_eth_link link_info; /* Wait for link */
+
+    /* Already started */
+    if (format_data->paused == DPDK_RUNNING)
+        return 0;
+
+    /* First time started we need to alloc our memory, doing this here
+     * rather than in environment setup because we don't have snaplen then */
+    if (format_data->paused == DPDK_NEVER_STARTED) {
+        if (format_data->snaplen == 0) {
+            format_data->snaplen = RX_MBUF_SIZE;
+            port_conf.rxmode.jumbo_frame = 0;
+            port_conf.rxmode.max_rx_pkt_len = 0;
+        } else {
+            /* Use jumbo frames */
+            port_conf.rxmode.jumbo_frame = 1;
+            port_conf.rxmode.max_rx_pkt_len = format_data->snaplen;
+        }
+
+        /* This is additional overhead so make sure we allow space for this */
+#if GET_MAC_CRC_CHECKSUM
+        format_data->snaplen += ETHER_CRC_LEN;
+#endif
+#if HAS_HW_TIMESTAMPS_82580
+        format_data->snaplen += sizeof(struct hw_timestamp_82580);
+#endif
+
+        /* Create the mbuf pool, which is the place our packets are allocated
+         * from - TODO figure out if there is a free function (I cannot see one)
+         * NOTE: RX queue requires nb_packets + 1 otherwise it fails to
+         * allocate however that extra 1 packet is not used.
+         * (I assume <= vs < error some where in DPDK code)
+         * TX requires nb_tx_buffers + 1 in the case the queue is full
+         * so that will fill the new buffer and wait until slots in the
+         * ring become available.
+         */
+#if DEBUG
+    fprintf(stderr, "Creating mempool named %s\n", format_data->mempool_name);
+#endif
+    format_data->pktmbuf_pool =
+            rte_mempool_create(format_data->mempool_name,
+                       (format_data->nb_rx_buf * rx_queues + format_data->nb_tx_buf + 1)*2,
+                       format_data->snaplen + sizeof(struct rte_mbuf)
+                                        + RTE_PKTMBUF_HEADROOM,
+                       128, sizeof(struct rte_pktmbuf_pool_private),
+                       rte_pktmbuf_pool_init, NULL, rte_pktmbuf_init, NULL,
+                       format_data->nic_numa_node, 0);
+
+        if (format_data->pktmbuf_pool == NULL) {
+            snprintf(err, errlen, "Intel DPDK - Initialisation of mbuf "
+                        "pool failed: %s", strerror(rte_errno));
+            return -1;
+        }
+    }
+
+    /* ----------- Now do the setup for the port mapping ------------ */
+    /* Order of calls must be
+     * rte_eth_dev_configure()
+     * rte_eth_tx_queue_setup()
+     * rte_eth_rx_queue_setup()
+     * rte_eth_dev_start()
+     * other rte_eth calls
+     */
+
+    /* This must be called first before another *eth* function
+     * 1 rx, 1 tx queue, port_conf sets checksum stripping etc */
+    ret = rte_eth_dev_configure(format_data->port, rx_queues, 1, &port_conf);
+    if (ret < 0) {
+        snprintf(err, errlen, "Intel DPDK - Cannot configure device port"
+                            " %"PRIu8" : %s", format_data->port,
+                            strerror(-ret));
+        return -1;
+    }
+#if DEBUG
+    fprintf(stderr, "Doing dev configure\n");
+#endif
+    /* Initialise the TX queue a minimum value if using this port for
+     * receiving. Otherwise a larger size if writing packets.
+     */
+    ret = rte_eth_tx_queue_setup(format_data->port, format_data->queue_id,
+                        format_data->nb_tx_buf, SOCKET_ID_ANY, &tx_conf);
+    if (ret < 0) {
+        snprintf(err, errlen, "Intel DPDK - Cannot configure TX queue on port"
+                            " %"PRIu8" : %s", format_data->port,
+                            strerror(-ret));
+        return -1;
+    }
+
+    for (i=0; i < rx_queues; i++) {
+#if DEBUG
+    fprintf(stderr, "Doing queue configure\n");
+#endif
+
+                /* Initialise the RX queue with some packets from memory */
+                ret = rte_eth_rx_queue_setup(format_data->port, i,
+                                             format_data->nb_rx_buf, format_data->nic_numa_node,
+                                             &rx_conf, format_data->pktmbuf_pool);
+        /* Init per_thread data structures */
+        format_data->per_lcore[i].port = format_data->port;
+        format_data->per_lcore[i].queue_id = i;
+
+                if (ret < 0) {
+                        snprintf(err, errlen, "Intel DPDK - Cannot configure RX queue on port"
+                                                " %"PRIu8" : %s", format_data->port,
+                                                strerror(-ret));
+                        return -1;
+                }
+        }
+
+#if DEBUG
+    fprintf(stderr, "Doing start device\n");
+#endif
+    /* Start device */
+    ret = rte_eth_dev_start(format_data->port);
+#if DEBUG
+    fprintf(stderr, "Done start device\n");
+#endif
+    if (ret < 0) {
+        snprintf(err, errlen, "Intel DPDK - rte_eth_dev_start failed : %s",
+                    strerror(-ret));
+        return -1;
+    }
+
+
+    /* Default promiscuous to on */
+    if (format_data->promisc == -1)
+        format_data->promisc = 1;
+
+    if (format_data->promisc == 1)
+        rte_eth_promiscuous_enable(format_data->port);
+    else
+        rte_eth_promiscuous_disable(format_data->port);
+
+
+    /* We have now successfully started/unpased */
+    format_data->paused = DPDK_RUNNING;
+
+    // Can use remote launch for all
+    /*RTE_LCORE_FOREACH_SLAVE(i) {
+                rte_eal_remote_launch(perpkt_threads_entry, (void *)libtrace, i);
+        }*/
+
+    /* Register a callback for link state changes */
+    ret = rte_eth_dev_callback_register(format_data->port,
+                                        RTE_ETH_EVENT_INTR_LSC,
+                                        dpdk_lsc_callback,
+                                        format_data);
+    /* If this fails it is not a show stopper */
+#if DEBUG
+    fprintf(stderr, "rte_eth_dev_callback_register failed %d : %s\n",
+            ret, strerror(-ret));
+#endif
+
+    /* Get the current link status */
+    rte_eth_link_get_nowait(format_data->port, &link_info);
+    format_data->link_speed = link_info.link_speed;
+#if DEBUG
+    fprintf(stderr, "Link status is %d %d %d\n", (int) link_info.link_status,
+            (int) link_info.link_duplex, (int) link_info.link_speed);
+        struct rte_eth_rss_reta reta_conf = {0};
+        reta_conf.mask_lo = ~reta_conf.mask_lo;
+        reta_conf.mask_hi = ~reta_conf.mask_hi;
+        int qew = rte_eth_dev_rss_reta_query(format_data->port, &reta_conf);
+        fprintf(stderr, "err=%d", qew);
+        for (i = 0; i < ETH_RSS_RETA_NUM_ENTRIES; i++) {
+                fprintf(stderr, "[%d] = %d\n", i, (int)reta_conf.reta[i]);
+        }
+
+#endif
+
     return 0;
 }
@@ -951 +1386 @@
 
     if (dpdk_start_port(FORMAT(libtrace), err, sizeof(err)) != 0) {
-        trace_set_err(libtrace, TRACE_ERR_INIT_FAILED, "%s", err);
-        free(libtrace->format_data);
-        libtrace->format_data = NULL;
-        return -1;
+        trace_set_err(libtrace, TRACE_ERR_INIT_FAILED, "%s", err);
+        free(libtrace->format_data);
+        libtrace->format_data = NULL;
+        return -1;
     }
     return 0;
+}
+
+static inline size_t dpdk_get_max_rx_queues (uint8_t port_id) {
+    struct rte_eth_dev_info dev_info;
+    rte_eth_dev_info_get(port_id, &dev_info);
+    return dev_info.max_rx_queues;
+}
+
+static inline size_t dpdk_processor_count () {
+    long nb_cpu = sysconf(_SC_NPROCESSORS_ONLN);
+    if (nb_cpu <= 0)
+        return 1;
+    else
+        return (size_t) nb_cpu;
+}
+
+static int dpdk_pstart_input (libtrace_t *libtrace) {
+    char err[500];
+    int i=0, phys_cores=0;
+    int tot = libtrace->perpkt_thread_count;
+    err[0] = 0;
+
+    if (rte_lcore_id() != rte_get_master_lcore())
+        fprintf(stderr, "Warning dpdk_pstart_input should be called from the master DPDK thread!\n");
+
+    // If the master is not on the last thread we move it there
+    if (rte_get_master_lcore() != RTE_MAX_LCORE - 1) {
+        // Consider error handling here
+        dpdk_move_master_lcore(RTE_MAX_LCORE - 1);
+    }
+
+    // Don't exceed the number of cores in the system/detected by dpdk
+    // We don't have to force this but performance wont be good if we don't
+    for (i = 0; i < RTE_MAX_LCORE; ++i) {
+        if (lcore_config[i].detected) {
+            if (rte_lcore_is_enabled(i))
+                fprintf(stderr, "Found core %d already in use!\n", i);
+            else
+                phys_cores++;
+        }
+    }
+
+        tot = MIN(libtrace->perpkt_thread_count, dpdk_get_max_rx_queues(FORMAT(libtrace)->port));
+        tot = MIN(tot, phys_cores);
+
+        fprintf(stderr, "Running pstart DPDK tot=%d req=%d phys=%d\n", tot, libtrace->perpkt_thread_count, phys_cores);
+
+    if (dpdk_start_port_queues(FORMAT(libtrace), err, sizeof(err), tot) != 0) {
+        trace_set_err(libtrace, TRACE_ERR_INIT_FAILED, "%s", err);
+        free(libtrace->format_data);
+        libtrace->format_data = NULL;
+        return -1;
+    }
+
+    // Make sure we only start the number that we should
+    libtrace->perpkt_thread_count = tot;
+    return 0;
+}
+
+
+/**
+ * Register a thread with the DPDK system,
+ * When we start DPDK in parallel libtrace we move the 'main thread' to the
+ * MAXIMUM CPU core slot (32) and remove any affinity restrictions DPDK
+ * gives it.
+ *
+ * We then allow a mapper thread to be started on every real core as DPDK would,
+ * we also bind these to the corresponding CPU cores.
+ *
+ * @param libtrace A pointer to the trace
+ * @param reading True if the thread will be used to read packets, i.e. will
+ *                call pread_packet(), false if thread used to process packet
+ *                in any other manner including statistics functions.
+ */
+static int dpdk_pregister_thread(libtrace_t *libtrace, libtrace_thread_t *t, bool reading)
+{
+    struct rte_config *cfg = rte_eal_get_configuration();
+    int i;
+    int new_id = -1;
+
+    // If 'reading packets' fill in cores from 0 up and bind affinity
+    // otherwise start from the MAX core (which is also the master) and work backwards
+    // in this case physical cores on the system will not exist so we don't bind
+    // these to any particular physical core
+    pthread_mutex_lock(&libtrace->libtrace_lock);
+    if (reading) {
+#if HAVE_LIBNUMA
+        for (i = 0; i < RTE_MAX_LCORE; ++i) {
+                if (!rte_lcore_is_enabled(i) && numa_node_of_cpu(i) == FORMAT(libtrace)->nic_numa_node) {
+                                new_id = i;
+                        if (!lcore_config[i].detected)
+                                new_id = -1;
+                        break;
+                }
+        }
+#endif
+        /* Retry without the the numa restriction */
+        if (new_id == -1) {
+                for (i = 0; i < RTE_MAX_LCORE; ++i) {
+                                if (!rte_lcore_is_enabled(i)) {
+                                        new_id = i;
+                                if (!lcore_config[i].detected)
+                                        fprintf(stderr, "Warning the number of 'reading' threads exceed cores on machine!!\n");
+                                break;
+                        }
+                }
+        }
+    } else {
+        for (i = RTE_MAX_LCORE-1; i >= 0; --i) {
+            if (!rte_lcore_is_enabled(i)) {
+                new_id = i;
+                break;
+            }
+        }
+    }
+
+    if (new_id == -1) {
+        assert(cfg->lcore_count == RTE_MAX_LCORE);
+        // TODO proper libtrace style error here!!
+        fprintf(stderr, "Too many threads for DPDK!!\n");
+        pthread_mutex_unlock(&libtrace->libtrace_lock);
+        return -1;
+    }
+
+    // Enable the core in global DPDK structs
+    cfg->lcore_role[new_id] = ROLE_RTE;
+    cfg->lcore_count++;
+    // Set TLS to reflect our new number
+    assert(rte_lcore_id() == 0); // I think new threads are going get a default thread number of 0
+    fprintf(stderr, "original id%d", rte_lcore_id());
+    RTE_PER_LCORE(_lcore_id) = new_id;
+        char name[99];
+        pthread_getname_np(pthread_self(),
+                              name, sizeof(name));
+
+    fprintf(stderr, "%s new id%d\n", name, rte_lcore_id());
+
+    if (reading) {
+        // Set affinity bind to corresponding core
+        cpu_set_t cpuset;
+        CPU_ZERO(&cpuset);
+        CPU_SET(rte_lcore_id(), &cpuset);
+        i = pthread_setaffinity_np(pthread_self(), sizeof(cpuset), &cpuset);
+        if (i != 0) {
+            fprintf(stderr, "Warning pthread_setaffinity_np failed\n");
+            pthread_mutex_unlock(&libtrace->libtrace_lock);
+            return -1;
+        }
+    }
+
+    // Map our TLS to the thread data
+    if (reading) {
+        if(t->type == THREAD_PERPKT) {
+            t->format_data = &FORMAT(libtrace)->per_lcore[t->perpkt_num];
+        } else {
+            t->format_data = &FORMAT(libtrace)->per_lcore[0];
+        }
+    }
+    pthread_mutex_unlock(&libtrace->libtrace_lock);
+    return 0;
+}
+
+
+/**
+ * Unregister a thread with the DPDK system.
+ *
+ * Only previously registered threads should be calling this just before
+ * they are destroyed.
+ */
+static void dpdk_punregister_thread(libtrace_t *libtrace UNUSED, libtrace_thread_t *t UNUSED)
+{
+    struct rte_config *cfg = rte_eal_get_configuration();
+
+    assert(rte_lcore_id() < RTE_MAX_LCORE);
+    pthread_mutex_lock(&libtrace->libtrace_lock);
+    // Skip if master!!
+    if (rte_lcore_id() == rte_get_master_lcore()) {
+        fprintf(stderr, "INFO: we are skipping unregistering the master lcore\n");
+        pthread_mutex_unlock(&libtrace->libtrace_lock);
+        return;
+    }
+
+    // Disable this core in global DPDK structs
+    cfg->lcore_role[rte_lcore_id()] = ROLE_OFF;
+    cfg->lcore_count--;
+    RTE_PER_LCORE(_lcore_id) = -1; // Might make the world burn if used again
+    assert(cfg->lcore_count >= 1); // We cannot unregister the master LCORE!!
+    pthread_mutex_unlock(&libtrace->libtrace_lock);
+    return;
 }
 
@@ -963 +1587 @@
     char err[500];
     err[0] = 0;
-    
+
     if (dpdk_start_port(FORMAT(libtrace), err, sizeof(err)) != 0) {
-        trace_set_err_out(libtrace, TRACE_ERR_INIT_FAILED, "%s", err);
-        free(libtrace->format_data);
-        libtrace->format_data = NULL;
-        return -1;
+        trace_set_err_out(libtrace, TRACE_ERR_INIT_FAILED, "%s", err);
+        free(libtrace->format_data);
+        libtrace->format_data = NULL;
+        return -1;
     }
     return 0;
 }
 
-static int dpdk_pause_input(libtrace_t * libtrace){
+static int dpdk_pause_input(libtrace_t * libtrace) {
     /* This stops the device, but can be restarted using rte_eth_dev_start() */
     if (FORMAT(libtrace)->paused == DPDK_RUNNING) {
-#if DEBUG      
-        fprintf(stderr, "Pausing port\n");
-#endif
-        rte_eth_dev_stop(FORMAT(libtrace)->port);
-        FORMAT(libtrace)->paused = DPDK_PAUSED;
-        /* If we pause it the driver will be reset and likely our counter */
+#if DEBUG
+        fprintf(stderr, "Pausing DPDK port\n");
+#endif
+        rte_eth_dev_stop(FORMAT(libtrace)->port);
+        FORMAT(libtrace)->paused = DPDK_PAUSED;
+        /* Empty the queue of packets */
+        for (; FORMAT(libtrace)->burst_offset < FORMAT(libtrace)->burst_size; ++FORMAT(libtrace)->burst_offset) {
+                rte_pktmbuf_free(FORMAT(libtrace)->burst_pkts[FORMAT(libtrace)->burst_offset]);
+        }
+        FORMAT(libtrace)->burst_offset = 0;
+        FORMAT(libtrace)->burst_size = 0;
+        /* If we pause it the driver will be reset and likely our counter */
+
+        FORMAT(libtrace)->per_lcore[0].ts_last_sys = 0;
 #if HAS_HW_TIMESTAMPS_82580
-        FORMAT(libtrace)->ts_first_sys = 0;
-        FORMAT(libtrace)->ts_last_sys = 0;
+        FORMAT(libtrace)->per_lcore[0].ts_first_sys = 0;
 #endif
     }
@@ -990 +1621 @@
 }
 
-static int dpdk_write_packet(libtrace_out_t *trace, 
+static int dpdk_write_packet(libtrace_out_t *trace,
                 libtrace_packet_t *packet){
     struct rte_mbuf* m_buff[1];
-    
+
     int wirelen = trace_get_wire_length(packet);
     int caplen = trace_get_capture_length(packet);
-    
+
     /* Check for a checksum and remove it */
     if (trace_get_link_type(packet) == TRACE_TYPE_ETH &&
-                                            wirelen == caplen)
-        caplen -= ETHER_CRC_LEN;
+                                            wirelen == caplen)
+        caplen -= ETHER_CRC_LEN;
 
     m_buff[0] = rte_pktmbuf_alloc(FORMAT(trace)->pktmbuf_pool);
     if (m_buff[0] == NULL) {
-        trace_set_err_out(trace, errno, "Cannot get an empty packet buffer");
-        return -1;
+        trace_set_err_out(trace, errno, "Cannot get an empty packet buffer");
+        return -1;
     } else {
-        int ret;
-        memcpy(rte_pktmbuf_append(m_buff[0], caplen), packet->payload, caplen);
-        do {
-            ret = rte_eth_tx_burst(FORMAT(trace)->queue_id, FORMAT(trace)->port, m_buff, 1);
-        } while (ret != 1);
+        int ret;
+        memcpy(rte_pktmbuf_append(m_buff[0], caplen), packet->payload, caplen);
+        do {
+            ret = rte_eth_tx_burst(FORMAT(trace)->queue_id, FORMAT(trace)->port, m_buff, 1);
+        } while (ret != 1);
     }
 
@@ -1020 +1651 @@
     /* Free our memory structures */
     if (libtrace->format_data != NULL) {
-        /* Close the device completely, device cannot be restarted */
-        if (FORMAT(libtrace)->port != 0xFF)
-            rte_eth_dev_close(FORMAT(libtrace)->port);
-        /* filter here if we used it */
+        /* Close the device completely, device cannot be restarted */
+        if (FORMAT(libtrace)->port != 0xFF)
+                rte_eth_dev_callback_unregister(FORMAT(libtrace)->port,
+                                                RTE_ETH_EVENT_INTR_LSC,
+                                                dpdk_lsc_callback,
+                                                FORMAT(libtrace));
+                rte_eth_dev_close(FORMAT(libtrace)->port);
+                /* filter here if we used it */
                 free(libtrace->format_data);
         }
@@ -1037 +1672 @@
     /* Free our memory structures */
     if (libtrace->format_data != NULL) {
-        /* Close the device completely, device cannot be restarted */
-        if (FORMAT(libtrace)->port != 0xFF)
-            rte_eth_dev_close(FORMAT(libtrace)->port);
-        /* filter here if we used it */
+        /* Close the device completely, device cannot be restarted */
+        if (FORMAT(libtrace)->port != 0xFF)
+            rte_eth_dev_close(FORMAT(libtrace)->port);
+        /* filter here if we used it */
                 free(libtrace->format_data);
         }
@@ -1050 +1685 @@
 }
 
-/** 
- * Get the start of additional header that we added to a packet.
+/**
+ * Get the start of the additional header that we added to a packet.
  */
 static inline struct dpdk_addt_hdr * get_addt_hdr (const libtrace_packet_t *packet) {
-    uint8_t *hdrsize;
     assert(packet);
     assert(packet->buffer);
-    hdrsize = (uint8_t *) MBUF_PKTDATA(packet->buffer);
-    /* The byte before the original packet data denotes the size in bytes
-     * of our additional header that we added sits before the 'size byte' */
-    hdrsize--;
-    return (struct dpdk_addt_hdr *) (hdrsize - *hdrsize);
+    /* Our header sits straight after the mbuf header */
+    return (struct dpdk_addt_hdr *) ((struct rte_mbuf*) packet->buffer + 1);
 }
 
@@ -1072 +1703 @@
     struct dpdk_addt_hdr * hdr = get_addt_hdr(packet);
     if (size > hdr->cap_len) {
-        /* Cannot make a packet bigger */
+        /* Cannot make a packet bigger */
                 return trace_get_capture_length(packet);
         }
@@ -1086 +1717 @@
     int org_cap_size; /* The original capture size */
     if (hdr->flags & INCLUDES_HW_TIMESTAMP) {
-        org_cap_size = (int) rte_pktmbuf_pkt_len(MBUF(packet->buffer)) -
-                            (MBUF_PKTDATA(packet->buffer) - (char *) hdr) -
-                            sizeof(struct hw_timestamp_82580);
+        org_cap_size = (int) rte_pktmbuf_pkt_len(MBUF(packet->buffer)) -
+                            sizeof(struct hw_timestamp_82580);
     } else {
-        org_cap_size = (int) rte_pktmbuf_pkt_len(MBUF(packet->buffer)) -
-                            (MBUF_PKTDATA(packet->buffer) - (char *) hdr);
+        org_cap_size = (int) rte_pktmbuf_pkt_len(MBUF(packet->buffer));
     }
     if (hdr->flags & INCLUDES_CHECKSUM) {
-        return org_cap_size;
+        return org_cap_size;
     } else {
-        /* DPDK packets are always TRACE_TYPE_ETH packets */
-        return org_cap_size + ETHER_CRC_LEN;
+        /* DPDK packets are always TRACE_TYPE_ETH packets */
+        return org_cap_size + ETHER_CRC_LEN;
     }
 }
@@ -1103 +1732 @@
     struct dpdk_addt_hdr * hdr = get_addt_hdr(packet);
     if (hdr->flags & INCLUDES_HW_TIMESTAMP)
-        return sizeof(struct rte_mbuf) + RTE_PKTMBUF_HEADROOM +
-                sizeof(struct hw_timestamp_82580);
+        return sizeof(struct rte_mbuf) + RTE_PKTMBUF_HEADROOM +
+                sizeof(struct hw_timestamp_82580);
     else
-        return sizeof(struct rte_mbuf) + RTE_PKTMBUF_HEADROOM;
+        return sizeof(struct rte_mbuf) + RTE_PKTMBUF_HEADROOM;
 }
 
@@ -1114 +1743 @@
     assert(packet);
     if (packet->buffer != buffer &&
-        packet->buf_control == TRACE_CTRL_PACKET) {
-        free(packet->buffer);
+        packet->buf_control == TRACE_CTRL_PACKET) {
+        free(packet->buffer);
     }
 
     if ((flags & TRACE_PREP_OWN_BUFFER) == TRACE_PREP_OWN_BUFFER) {
-        packet->buf_control = TRACE_CTRL_PACKET;
+        packet->buf_control = TRACE_CTRL_PACKET;
     } else
-        packet->buf_control = TRACE_CTRL_EXTERNAL;
+        packet->buf_control = TRACE_CTRL_EXTERNAL;
 
     packet->buffer = buffer;
@@ -1132 +1761 @@
 }
 
+
+/**
+ * Given a packet size and a link speed, computes the
+ * time to transmit in nanoseconds.
+ *
+ * @param format_data The dpdk format data from which we get the link speed
+ *        and if unset updates it in a thread safe manner
+ * @param pkt_size The size of the packet in bytes
+ * @return The wire time in nanoseconds
+ */
+static inline uint32_t calculate_wire_time(struct dpdk_format_data_t* format_data, uint32_t pkt_size) {
+        uint32_t wire_time;
+        /* 20 extra bytes of interframe gap and preamble */
+# if GET_MAC_CRC_CHECKSUM
+        wire_time = ((pkt_size + 20) * 8000);
+# else
+        wire_time = ((pkt_size + 20 + ETHER_CRC_LEN) * 8000);
+# endif
+
+        /* Division is really slow and introduces a pipeline stall
+         * The compiler will optimise this into magical multiplication and shifting
+         * See http://ridiculousfish.com/blog/posts/labor-of-division-episode-i.html
+         */
+retry_calc_wiretime:
+        switch (format_data->link_speed) {
+        case ETH_LINK_SPEED_40G:
+                wire_time /=  ETH_LINK_SPEED_40G;
+                break;
+        case ETH_LINK_SPEED_20G:
+                wire_time /= ETH_LINK_SPEED_20G;
+                break;
+        case ETH_LINK_SPEED_10G:
+                wire_time /= ETH_LINK_SPEED_10G;
+                break;
+        case ETH_LINK_SPEED_1000:
+                wire_time /= ETH_LINK_SPEED_1000;
+                break;
+        case 0:
+                {
+                /* Maybe the link was down originally, but now it should be up */
+                struct rte_eth_link link = {0};
+                rte_eth_link_get_nowait(format_data->port, &link);
+                if (link.link_status && link.link_speed) {
+                        format_data->link_speed = link.link_speed;
+#ifdef DEBUG
+                        fprintf(stderr, "Link has come up updated speed=%d\n", (int) link.link_speed);
+#endif
+                        goto retry_calc_wiretime;
+                }
+                /* We don't know the link speed, make sure numbers are counting up */
+                wire_time = 1;
+                break;
+                }
+        default:
+                wire_time /= format_data->link_speed;
+        }
+        return wire_time;
+}
+
+
+
 /*
- * Does any extra preperation to a captured packet.
- * This includes adding our extra header to it with the timestamp
- */
-static inline int dpdk_ready_pkt(libtrace_t *libtrace, libtrace_packet_t *packet,
-                                                        struct rte_mbuf* pkt){
-    uint8_t * hdr_size;
-    struct dpdk_addt_hdr *hdr;
+ * Does any extra preperation to all captured packets
+ * This includes adding our extra header to it with the timestamp,
+ * and any snapping
+ *
+ * @param format_data The DPDK format data
+ * @param plc The DPDK per lcore format data
+ * @param pkts An array of size nb_pkts of DPDK packets
+ * @param nb_pkts The number of packets in pkts and optionally packets
+ * @param packets Optional - If not null nb_pkts of libtrace packets which will be prepared
+ */
+static inline void dpdk_ready_pkts(struct dpdk_format_data_t *format_data, struct dpdk_per_lcore_t *plc,
+                                   struct rte_mbuf **pkts, size_t nb_pkts, libtrace_packet_t **packets) {
+        struct dpdk_addt_hdr *hdr;
+        size_t i;
+        uint64_t cur_sys_time_ns;
 #if HAS_HW_TIMESTAMPS_82580
-    struct hw_timestamp_82580 *hw_ts;
-    struct timeval cur_sys_time;
-    uint64_t cur_sys_time_ns;
-    uint64_t estimated_wraps;
-    
-    /* Using gettimeofday because it's most likely to be a vsyscall
-     * We don't want to slow down anything with systemcalls we dont need
-     * accauracy */
-    gettimeofday(&cur_sys_time, NULL);
+        struct hw_timestamp_82580 *hw_ts;
+        uint64_t estimated_wraps;
 #else
-# if USE_CLOCK_GETTIME
-    struct timespec cur_sys_time;
-    
-    /* This looks terrible and I feel bad doing it. But it's OK 
-     * on new kernels, because this is a vsyscall */
-    clock_gettime(CLOCK_REALTIME, &cur_sys_time);
-# else
-    struct timeval cur_sys_time;
-    /* Should be a vsyscall */
-    gettimeofday(&cur_sys_time, NULL);
-# endif
-#endif
-
-    /* Record the size of our header */
-    hdr_size = (uint8_t *) rte_pktmbuf_prepend(pkt, sizeof(uint8_t));
-    *hdr_size = sizeof(struct dpdk_addt_hdr);
-    /* Now put our header in front of that size */
-    hdr = (struct dpdk_addt_hdr *) rte_pktmbuf_prepend(pkt, sizeof(struct dpdk_addt_hdr));
-    memset(hdr, 0, sizeof(struct dpdk_addt_hdr));
-    
+
+#endif
+
+#if USE_CLOCK_GETTIME
+        struct timespec cur_sys_time = {0};
+        /* This looks terrible and I feel bad doing it. But it's OK
+         * on new kernels, because this is a fast vsyscall */
+        clock_gettime(CLOCK_REALTIME, &cur_sys_time);
+        cur_sys_time_ns = TS_TO_NS(cur_sys_time);
+#else
+        struct timeval cur_sys_time = {0};
+        /* Also a fast vsyscall */
+        gettimeofday(&cur_sys_time, NULL);
+        cur_sys_time_ns = TV_TO_NS(cur_sys_time);
+#endif
+
+        /* The system clock is not perfect so when running
+         * at linerate we could timestamp a packet in the past.
+         * To avoid this we munge the timestamp to appear 1ns
+         * after the previous packet. We should eventually catch up
+         * to system time since a 64byte packet on a 10G link takes 67ns.
+         *
+         * Note with parallel readers timestamping packets
+         * with duplicate stamps or out of order is unavoidable without
+         * hardware timestamping from the NIC.
+         */
+#if !HAS_HW_TIMESTAMPS_82580
+        if (plc->ts_last_sys >= cur_sys_time_ns) {
+                cur_sys_time_ns = plc->ts_last_sys + 1;
+        }
+#endif
+
+        assert(RTE_PKTMBUF_HEADROOM >= sizeof(struct dpdk_addt_hdr)); // TODO static compile time assert sometime??
+        for (i = 0 ; i < nb_pkts ; ++i) {
+
+                /* We put our header straight after the dpdk header */
+                hdr = (struct dpdk_addt_hdr *) (pkts[i] + 1);
+                memset(hdr, 0, sizeof(struct dpdk_addt_hdr));
+
 #if GET_MAC_CRC_CHECKSUM
+<<<<<<< HEAD
+                /* Add back in the CRC sum */
+                pkts[i]->pkt.pkt_len += ETHER_CRC_LEN;
+                pkts[i]->pkt.data_len += ETHER_CRC_LEN;
+                hdr->flags |= INCLUDES_CHECKSUM;
+=======
     /* Add back in the CRC sum */
     rte_pktmbuf_pkt_len(pkt) += ETHER_CRC_LEN;
     rte_pktmbuf_data_len(pkt) += ETHER_CRC_LEN;
     hdr->flags |= INCLUDES_CHECKSUM;
-#endif
+>>>>>>> master
+#endif
+
+                hdr->cap_len = rte_pktmbuf_pkt_len(pkts[i]);
 
 #if HAS_HW_TIMESTAMPS_82580
-    /* Taken from igb_ptp.c part of Intel Linux drivers (Good example code)
-     *
-     *        +----------+---+   +--------------+
-     *  82580 |    24    | 8 |   |      32      |
-     *        +----------+---+   +--------------+
-     *          reserved  \______ 40 bits _____/
-     * 
-     * The 40 bit 82580 SYSTIM overflows every
-     *   2^40 * 10^-9 /  60  = 18.3 minutes.
-     * 
-     * NOTE picture is in Big Endian order, in memory it's acutally in Little
-     * Endian (for the full 64 bits) i.e. picture is mirrored
-     */
-    
-    /* The timestamp is sitting before our packet and is included in pkt_len */
-    hdr->flags |= INCLUDES_HW_TIMESTAMP;
-    hw_ts = (struct hw_timestamp_82580 *) MBUF_PKTDATA(pkt);
-    
-    /* Despite what the documentation says this is in Little 
-     * Endian byteorder. Mask the reserved section out.
-     */
-    hdr->timestamp = le64toh(hw_ts->timestamp) & 
-                ~(((~0ull)>>TS_NBITS_82580)<<TS_NBITS_82580);
-                
-    cur_sys_time_ns = TV_TO_NS(cur_sys_time);
-    if (unlikely(FORMAT(libtrace)->ts_first_sys == 0)) {
-        FORMAT(libtrace)->ts_first_sys = cur_sys_time_ns - hdr->timestamp;
-        FORMAT(libtrace)->ts_last_sys = FORMAT(libtrace)->ts_first_sys;
-    }
-    
-    /* This will have serious problems if packets aren't read quickly
-     * that is within a couple of seconds because our clock cycles every
-     * 18 seconds */
-    estimated_wraps = (cur_sys_time_ns - FORMAT(libtrace)->ts_last_sys)
-                            / (1ull<<TS_NBITS_82580);
-    
-    /* Estimated_wraps gives the number of times the counter should have
-     * wrapped (however depending on value last time it could have wrapped
-     * twice more (if hw clock is close to its max value) or once less (allowing
-     * for a bit of variance between hw and sys clock). But if the clock
-     * shouldn't have wrapped once then don't allow it to go backwards in time */
-    if (unlikely(estimated_wraps >= 2)) {
-        /* 2 or more wrap arounds add all but the very last wrap */
-        FORMAT(libtrace)->wrap_count += estimated_wraps - 1;
-    }
-    
-    /* Set the timestamp to the lowest possible value we're considering */
-    hdr->timestamp += FORMAT(libtrace)->ts_first_sys +
-                        FORMAT(libtrace)->wrap_count * (1ull<<TS_NBITS_82580);
-    
-    /* In most runs only the first if() will need evaluating - i.e our
-     * estimate is correct. */
-    if (unlikely(!WITHIN_VARIANCE(cur_sys_time_ns,
-                                hdr->timestamp, MAXSKEW_82580))) {
-        /* Failed to match estimated_wraps-1 (or estimated_wraps in ==0 case) */
-        FORMAT(libtrace)->wrap_count++;
-        hdr->timestamp += (1ull<<TS_NBITS_82580);
-        if (!WITHIN_VARIANCE(cur_sys_time_ns,
-                                hdr->timestamp, MAXSKEW_82580)) {
-            /* Failed to match estimated_wraps */
-            FORMAT(libtrace)->wrap_count++;
-            hdr->timestamp += (1ull<<TS_NBITS_82580);
-            if (!WITHIN_VARIANCE(cur_sys_time_ns,
-                                hdr->timestamp, MAXSKEW_82580)) {
-                if (estimated_wraps == 0) {
-                    /* 0 case Failed to match estimated_wraps+2 */
-                    printf("WARNING - Hardware Timestamp failed to"
-                                            " match using systemtime!\n");
-                    hdr->timestamp = cur_sys_time_ns;
-                } else {
-                    /* Failed to match estimated_wraps+1 */
-                    FORMAT(libtrace)->wrap_count++;
-                    hdr->timestamp += (1ull<<TS_NBITS_82580);
-                    if (!WITHIN_VARIANCE(cur_sys_time_ns,
-                                hdr->timestamp, MAXSKEW_82580)) {
-                        /* Failed to match estimated_wraps+2 */
-                        printf("WARNING - Hardware Timestamp failed to"
-                                            " match using systemtime!!\n");
-                    }
-                }
-            }
-        }
-    }
-
-    /* Log our previous for the next loop */
-    FORMAT(libtrace)->ts_last_sys = TV_TO_NS(cur_sys_time);
-
+                /* The timestamp is sitting before our packet and is included in pkt_len */
+                hdr->flags |= INCLUDES_HW_TIMESTAMP;
+                hdr->cap_len -= sizeof(struct hw_timestamp_82580);
+                hw_ts = (struct hw_timestamp_82580 *) MBUF_PKTDATA(pkts[i]);
+
+                /* Taken from igb_ptp.c part of Intel Linux drivers (Good example code)
+                 *
+                 *        +----------+---+   +--------------+
+                 *  82580 |    24    | 8 |   |      32      |
+                 *        +----------+---+   +--------------+
+                 *          reserved  \______ 40 bits _____/
+                 *
+                 * The 40 bit 82580 SYSTIM overflows every
+                 *   2^40 * 10^-9 /  60  = 18.3 minutes.
+                 *
+                 * NOTE picture is in Big Endian order, in memory it's acutally in Little
+                 * Endian (for the full 64 bits) i.e. picture is mirrored
+                 */
+
+                /* Despite what the documentation says this is in Little
+                 * Endian byteorder. Mask the reserved section out.
+                 */
+                hdr->timestamp = le64toh(hw_ts->timestamp) &
+                        ~(((~0ull)>>TS_NBITS_82580)<<TS_NBITS_82580);
+
+                if (unlikely(plc->ts_first_sys == 0)) {
+                        plc->ts_first_sys = cur_sys_time_ns - hdr->timestamp;
+                        plc->ts_last_sys = plc->ts_first_sys;
+                }
+
+                /* This will have serious problems if packets aren't read quickly
+                 * that is within a couple of seconds because our clock cycles every
+                 * 18 seconds */
+                estimated_wraps = (cur_sys_time_ns - plc->ts_last_sys)
+                                  / (1ull<<TS_NBITS_82580);
+
+                /* Estimated_wraps gives the number of times the counter should have
+                 * wrapped (however depending on value last time it could have wrapped
+                 * twice more (if hw clock is close to its max value) or once less (allowing
+                 * for a bit of variance between hw and sys clock). But if the clock
+                 * shouldn't have wrapped once then don't allow it to go backwards in time */
+                if (unlikely(estimated_wraps >= 2)) {
+                        /* 2 or more wrap arounds add all but the very last wrap */
+                        plc->wrap_count += estimated_wraps - 1;
+                }
+
+                /* Set the timestamp to the lowest possible value we're considering */
+                hdr->timestamp += plc->ts_first_sys +
+                                  plc->wrap_count * (1ull<<TS_NBITS_82580);
+
+                /* In most runs only the first if() will need evaluating - i.e our
+                 * estimate is correct. */
+                if (unlikely(!WITHIN_VARIANCE(cur_sys_time_ns,
+                                              hdr->timestamp, MAXSKEW_82580))) {
+                        /* Failed to match estimated_wraps-1 (or estimated_wraps in ==0 case) */
+                        plc->wrap_count++;
+                        hdr->timestamp += (1ull<<TS_NBITS_82580);
+                        if (!WITHIN_VARIANCE(cur_sys_time_ns,
+                                             hdr->timestamp, MAXSKEW_82580)) {
+                                /* Failed to match estimated_wraps */
+                                plc->wrap_count++;
+                                hdr->timestamp += (1ull<<TS_NBITS_82580);
+                                if (!WITHIN_VARIANCE(cur_sys_time_ns,
+                                                     hdr->timestamp, MAXSKEW_82580)) {
+                                        if (estimated_wraps == 0) {
+                                                /* 0 case Failed to match estimated_wraps+2 */
+                                                printf("WARNING - Hardware Timestamp failed to"
+                                                       " match using systemtime!\n");
+                                                hdr->timestamp = cur_sys_time_ns;
+                                        } else {
+                                                /* Failed to match estimated_wraps+1 */
+                                                plc->wrap_count++;
+                                                hdr->timestamp += (1ull<<TS_NBITS_82580);
+                                                if (!WITHIN_VARIANCE(cur_sys_time_ns,
+                                                                     hdr->timestamp, MAXSKEW_82580)) {
+                                                        /* Failed to match estimated_wraps+2 */
+                                                        printf("WARNING - Hardware Timestamp failed to"
+                                                               " match using systemtime!!\n");
+                                                }
+                                        }
+                                }
+                        }
+                }
 #else
-# if USE_CLOCK_GETTIME
-    hdr->timestamp = TS_TO_NS(cur_sys_time);
-# else
-    hdr->timestamp = TV_TO_NS(cur_sys_time);
-# endif
-#endif
-
-    /* Intels samples prefetch into level 0 cache lets assume it is a good
-     * idea and do the same */
-    rte_prefetch0(rte_pktmbuf_mtod(pkt, void *));
-    packet->buffer = pkt;
-    dpdk_prepare_packet(libtrace, packet, packet->buffer, packet->type, 0);
-
-    /* Set our capture length for the first time */
-    hdr->cap_len = dpdk_get_wire_length(packet);
-    if (!(hdr->flags & INCLUDES_CHECKSUM)) {
-        hdr->cap_len -= ETHER_CRC_LEN;
-    }
-    
-
-    return dpdk_get_framing_length(packet) +
-                        dpdk_get_capture_length(packet);
-}
+
+                hdr->timestamp = cur_sys_time_ns;
+                /* Offset the next packet by the wire time of previous */
+                calculate_wire_time(format_data, hdr->cap_len);
+
+#endif
+                if(packets) {
+                        packets[i]->buffer = pkts[i];
+                        packets[i]->header = pkts[i];
+#if HAS_HW_TIMESTAMPS_82580
+                        packets[i]->payload = (char *) pkts[i] + sizeof(struct rte_mbuf) +
+                                              RTE_PKTMBUF_HEADROOM + sizeof(struct hw_timestamp_82580);
+#else
+                        packets[i]->payload = (char *) pkts[i] + sizeof(struct rte_mbuf) +
+                                              RTE_PKTMBUF_HEADROOM;
+#endif
+                        packets[i]->error = 1;
+                }
+        }
+
+        plc->ts_last_sys = cur_sys_time_ns;
+
+        return;
+}
+
+
+static void dpdk_fin_packet(libtrace_packet_t *packet)
+{
+        if ( packet->buf_control == TRACE_CTRL_EXTERNAL ) {
+                rte_pktmbuf_free(packet->buffer);
+                packet->buffer = NULL;
+        }
+}
+
 
 static int dpdk_read_packet (libtrace_t *libtrace, libtrace_packet_t *packet) {
-    int nb_rx; /* Number of rx packets we've recevied */
-    struct rte_mbuf* pkts_burst[1]; /* Array of 1 pointer(s) */
+    int nb_rx; /* Number of rx packets we've received */
 
     /* Free the last packet buffer */
     if (packet->buffer != NULL) {
-        /* Buffer is owned by DPDK */
-        if ( packet->buf_control == TRACE_CTRL_EXTERNAL ) {
-            rte_pktmbuf_free(packet->buffer);
-            packet->buffer = NULL;
-        } else
-        /* Buffer is owned by packet i.e. has been malloc'd */
-        if (packet->buf_control == TRACE_CTRL_PACKET) {
-            free(packet->buffer);
-            packet->buffer = NULL;
-        }
-    }
-    
+        /* Buffer is owned by DPDK */
+        if ( packet->buf_control == TRACE_CTRL_EXTERNAL ) {
+            rte_pktmbuf_free(packet->buffer);
+            packet->buffer = NULL;
+        } else
+        /* Buffer is owned by packet i.e. has been malloc'd */
+        if (packet->buf_control == TRACE_CTRL_PACKET) {
+            free(packet->buffer);
+            packet->buffer = NULL;
+        }
+    }
+
     packet->buf_control = TRACE_CTRL_EXTERNAL;
     packet->type = TRACE_RT_DATA_DPDK;
-    
+
+    /* Check if we already have some packets buffered */
+    if (FORMAT(libtrace)->burst_size != FORMAT(libtrace)->burst_offset) {
+            packet->buffer =  FORMAT(libtrace)->burst_pkts[FORMAT(libtrace)->burst_offset++];
+            dpdk_prepare_packet(libtrace, packet, packet->buffer, packet->type, 0);
+            return 1; // TODO should be bytes read, which essentially useless anyway
+    }
     /* Wait for a packet */
     while (1) {
-        /* Poll for a single packet */
-        nb_rx = rte_eth_rx_burst(FORMAT(libtrace)->port,
-                            FORMAT(libtrace)->queue_id, pkts_burst, 1);
-        if (nb_rx > 0) { /* Got a packet - otherwise we keep spining */
-            return dpdk_ready_pkt(libtrace, packet, pkts_burst[0]);
-        }
-        if (libtrace_halt) {
-            return 0;
-        }
-    }
-    
+        /* Poll for a single packet */
+        nb_rx = rte_eth_rx_burst(FORMAT(libtrace)->port,
+                                 FORMAT(libtrace)->queue_id, FORMAT(libtrace)->burst_pkts, BURST_SIZE);
+        if (nb_rx > 0) { /* Got a packet - otherwise we keep spining */
+                FORMAT(libtrace)->burst_size = nb_rx;
+                FORMAT(libtrace)->burst_offset = 1;
+                dpdk_ready_pkts(FORMAT(libtrace), &FORMAT(libtrace)->per_lcore[0], FORMAT(libtrace)->burst_pkts, nb_rx, NULL);
+                packet->buffer = FORMAT(libtrace)->burst_pkts[0];
+                dpdk_prepare_packet(libtrace, packet, packet->buffer, packet->type, 0);
+                return 1; // TODO should be bytes read, which essentially useless anyway
+        }
+        if (libtrace_halt) {
+                return 0;
+        }
+        /* Wait a while, polling on memory degrades performance
+         * This relieves the pressure on memory allowing the NIC to DMA */
+        rte_delay_us(10);
+    }
+
+    /* We'll never get here - but if we did it would be bad */
+    return -1;
+}
+
+static int dpdk_pread_packets (libtrace_t *libtrace, libtrace_thread_t *t, libtrace_packet_t **packets, size_t nb_packets) {
+    size_t nb_rx; /* Number of rx packets we've recevied */
+    struct rte_mbuf* pkts_burst[nb_packets]; /* Array of pointer(s) */
+    size_t i;
+
+    for (i = 0 ; i < nb_packets ; ++i) {
+            /* Free the last packet buffer */
+            if (packets[i]->buffer != NULL) {
+                /* Buffer is owned by DPDK */
+                if (packets[i]->buf_control == TRACE_CTRL_EXTERNAL) {
+                    rte_pktmbuf_free(packets[i]->buffer);
+                    packets[i]->buffer = NULL;
+                } else
+                /* Buffer is owned by packet i.e. has been malloc'd */
+                if (packets[i]->buf_control == TRACE_CTRL_PACKET) {
+                    free(packets[i]->buffer);
+                    packets[i]->buffer = NULL;
+                }
+            }
+            packets[i]->buf_control = TRACE_CTRL_EXTERNAL;
+            packets[i]->type = TRACE_RT_DATA_DPDK;
+    }
+
+    /* Wait for a packet */
+    while (1) {
+        /* Poll for a single packet */
+        nb_rx = rte_eth_rx_burst(PERPKT_FORMAT(t)->port,
+                            PERPKT_FORMAT(t)->queue_id, pkts_burst, nb_packets);
+        if (nb_rx > 0) {
+                /* Got some packets - otherwise we keep spining */
+                //fprintf(stderr, "Doing P READ PACKET port=%d q=%d\n", (int) FORMAT(libtrace)->port, (int) get_thread_table_num(libtrace));
+                dpdk_ready_pkts(FORMAT(libtrace), PERPKT_FORMAT(t), pkts_burst, nb_rx, packets);
+                return nb_rx;
+        }
+        // Check the message queue this could be (Well it shouldn't but anyway) be less than 0
+        if (libtrace_message_queue_count(&t->messages) > 0) {
+                printf("Extra message yay");
+                return -2;
+        }
+        if (libtrace_halt) {
+                return 0;
+        }
+        /* Wait a while, polling on memory degrades performance
+         * This relieves the pressure on memory allowing the NIC to DMA */
+        rte_delay_us(10);
+    }
+
     /* We'll never get here - but if we did it would be bad */
     return -1;
@@ -1332 +2122 @@
     struct timeval tv;
     struct dpdk_addt_hdr * hdr = get_addt_hdr(packet);
-    
+
     tv.tv_sec = hdr->timestamp / (uint64_t) 1000000000;
     tv.tv_usec = (hdr->timestamp % (uint64_t) 1000000000) / 1000;
@@ -1341 +2131 @@
     struct timespec ts;
     struct dpdk_addt_hdr * hdr = get_addt_hdr(packet);
-    
+
     ts.tv_sec = hdr->timestamp / (uint64_t) 1000000000;
     ts.tv_nsec = hdr->timestamp % (uint64_t) 1000000000;
@@ -1368 +2158 @@
 static uint64_t dpdk_get_dropped_packets (libtrace_t *trace) {
     struct rte_eth_stats stats = {0};
-    
+
     if (trace->format_data == NULL || FORMAT(trace)->port == 0xFF)
-        return UINT64_MAX;
+        return UINT64_MAX;
     /* Grab the current stats */
     rte_eth_stats_get(FORMAT(trace)->port, &stats);
-    
+
     /* Get the drop counter */
     return (uint64_t) stats.ierrors;
@@ -1380 +2170 @@
 static uint64_t dpdk_get_captured_packets (libtrace_t *trace) {
     struct rte_eth_stats stats = {0};
-    
+
     if (trace->format_data == NULL || FORMAT(trace)->port == 0xFF)
-        return UINT64_MAX;
+        return UINT64_MAX;
     /* Grab the current stats */
     rte_eth_stats_get(FORMAT(trace)->port, &stats);
-    
+
     /* Get the drop counter */
     return (uint64_t) stats.ipackets;
@@ -1393 +2183 @@
  * This is the number of packets filtered by the NIC
  * and maybe ahead of number read using libtrace.
- * 
+ *
  * XXX we are yet to implement any filtering, but if it was this should
  * get the result. So this will just return 0 for now.
@@ -1399 +2189 @@
 static uint64_t dpdk_get_filtered_packets (libtrace_t *trace) {
     struct rte_eth_stats stats = {0};
-    
+
     if (trace->format_data == NULL || FORMAT(trace)->port == 0xFF)
-        return UINT64_MAX;
+        return UINT64_MAX;
     /* Grab the current stats */
     rte_eth_stats_get(FORMAT(trace)->port, &stats);
-    
+
     /* Get the drop counter */
     return (uint64_t) stats.fdirmiss;
@@ -1414 +2204 @@
  */
 static libtrace_eventobj_t dpdk_trace_event(libtrace_t *trace,
-                                        libtrace_packet_t *packet) {
+                                        libtrace_packet_t *packet) {
     libtrace_eventobj_t event = {0,0,0.0,0};
     int nb_rx; /* Number of receive packets we've read */
     struct rte_mbuf* pkts_burst[1]; /* Array of 1 pointer(s) to rx buffers */
-    
+
     do {
-    
-        /* See if we already have a packet waiting */
-        nb_rx = rte_eth_rx_burst(FORMAT(trace)->port,
-                        FORMAT(trace)->queue_id, pkts_burst, 1);
-        
-        if (nb_rx > 0) {
-            /* Free the last packet buffer */
-            if (packet->buffer != NULL) {
-                /* Buffer is owned by DPDK */
-                if ( packet->buf_control == TRACE_CTRL_EXTERNAL ) {
-                    rte_pktmbuf_free(packet->buffer);
-                    packet->buffer = NULL;
-                } else
-                /* Buffer is owned by packet i.e. has been malloc'd */
-                if (packet->buf_control == TRACE_CTRL_PACKET) {
-                    free(packet->buffer);
-                    packet->buffer = NULL;
-                }
-            }
-            
-            packet->buf_control = TRACE_CTRL_EXTERNAL;
-            packet->type = TRACE_RT_DATA_DPDK;
-            event.type = TRACE_EVENT_PACKET;
-            event.size = dpdk_ready_pkt(trace, packet, pkts_burst[0]);
-            
-            /* XXX - Check this passes the filter trace_read_packet normally
-             * does this for us but this wont */
-            if (trace->filter) {
-                if (!trace_apply_filter(trace->filter, packet)) {
-                    /* Failed the filter so we loop for another packet */
-                    trace->filtered_packets ++;
-                    continue;
-                }
-            }
-            trace->accepted_packets ++;
-        } else {
-            /* We only want to sleep for a very short time - we are non-blocking */
-            event.type = TRACE_EVENT_SLEEP;
-            event.seconds = 0.0001;
-            event.size = 0;
-        }
-        
-        /* If we get here we have our event */
-        break;
+
+        /* See if we already have a packet waiting */
+        nb_rx = rte_eth_rx_burst(FORMAT(trace)->port,
+                        FORMAT(trace)->queue_id, pkts_burst, 1);
+
+        if (nb_rx > 0) {
+            /* Free the last packet buffer */
+            if (packet->buffer != NULL) {
+                /* Buffer is owned by DPDK */
+                if ( packet->buf_control == TRACE_CTRL_EXTERNAL ) {
+                    rte_pktmbuf_free(packet->buffer);
+                    packet->buffer = NULL;
+                } else
+                /* Buffer is owned by packet i.e. has been malloc'd */
+                if (packet->buf_control == TRACE_CTRL_PACKET) {
+                    free(packet->buffer);
+                    packet->buffer = NULL;
+                }
+            }
+
+            packet->buf_control = TRACE_CTRL_EXTERNAL;
+            packet->type = TRACE_RT_DATA_DPDK;
+            event.type = TRACE_EVENT_PACKET;
+            dpdk_ready_pkts(FORMAT(trace), &FORMAT(trace)->per_lcore[0], pkts_burst, 1, &packet);
+            event.size = 1; // TODO should be bytes read, which essentially useless anyway
+
+            /* XXX - Check this passes the filter trace_read_packet normally
+             * does this for us but this wont */
+            if (trace->filter) {
+                if (!trace_apply_filter(trace->filter, packet)) {
+                    /* Failed the filter so we loop for another packet */
+                    trace->filtered_packets ++;
+                    continue;
+                }
+            }
+            trace->accepted_packets ++;
+        } else {
+            /* We only want to sleep for a very short time - we are non-blocking */
+            event.type = TRACE_EVENT_SLEEP;
+            event.seconds = 0.0001;
+            event.size = 0;
+        }
+
+        /* If we get here we have our event */
+        break;
     } while (1);
 
@@ -1488 +2279 @@
 }
 
- static struct libtrace_format_t dpdk = {
+static struct libtrace_format_t dpdk = {
         "dpdk",
         "$Id: format_dpdk.c 1805 2013-03-08 02:01:35Z salcock $",
@@ -1505 +2296 @@
         dpdk_read_packet,           /* read_packet */
         dpdk_prepare_packet,    /* prepare_packet */
-        NULL,                               /* fin_packet */
+        dpdk_fin_packet,                                    /* fin_packet */
         dpdk_write_packet,          /* write_packet */
         dpdk_get_link_type,         /* get_link_type */
@@ -1528 +2319 @@
         dpdk_trace_event,               /* trace_event */
     dpdk_help,              /* help */
-        NULL
+    NULL,                   /* next pointer */
+    {true, 8},              /* Live, NICs typically have 8 threads */
+    dpdk_pstart_input, /* pstart_input */
+        dpdk_pread_packets, /* pread_packets */
+        dpdk_pause_input, /* ppause */
+        dpdk_fin_input, /* p_fin */
+        dpdk_pconfig_input, /* pconfig_input */
+    dpdk_pregister_thread, /* pregister_thread */
+    dpdk_punregister_thread /* unpregister_thread */
 };
 

lib/format_duck.c

@@ -366 +366 @@
         NULL,                           /* trace_event */
         duck_help,                      /* help */
-        NULL                            /* next pointer */
+        NULL,                            /* next pointer */
+        NON_PARALLEL(false)
 };
 

lib/format_erf.c

@@ -837 +837 @@
         erf_event,                      /* trace_event */
         erf_help,                       /* help */
-        NULL                            /* next pointer */
+        NULL,                           /* next pointer */
+        NON_PARALLEL(false)
 };
 
@@ -880 +881 @@
         erf_event,                      /* trace_event */
         erf_help,                       /* help */
-        NULL                            /* next pointer */
+        NULL,                           /* next pointer */
+        NON_PARALLEL(false)
 };
 

lib/format_legacy.c

@@ -552 +552 @@
         trace_event_trace,              /* trace_event */
         legacyatm_help,                 /* help */
-        NULL                            /* next pointer */
+        NULL,                           /* next pointer */
+        NON_PARALLEL(false)
 };
 
@@ -595 +596 @@
         trace_event_trace,              /* trace_event */
         legacyeth_help,                 /* help */
-        NULL                            /* next pointer */
+        NULL,                           /* next pointer */
+        NON_PARALLEL(false)
 };
 
@@ -639 +641 @@
         legacypos_help,                 /* help */
         NULL,                           /* next pointer */
+        NON_PARALLEL(false)
 };
 
@@ -682 +685 @@
         legacynzix_help,                /* help */
         NULL,                           /* next pointer */
+        NON_PARALLEL(false)
 };
         

lib/format_pcap.c

@@ -834 +834 @@
         trace_event_trace,              /* trace_event */
         pcap_help,                      /* help */
-        NULL                            /* next pointer */
+        NULL,                   /* next pointer */
+        NON_PARALLEL(false)
 };
 
@@ -877 +878 @@
         trace_event_device,             /* trace_event */
         pcapint_help,                   /* help */
-        NULL                            /* next pointer */
+        NULL,                   /* next pointer */
+        NON_PARALLEL(true)
 };
 

lib/format_pcapfile.c

@@ -786 +786 @@
         pcapfile_event,         /* trace_event */
         pcapfile_help,                  /* help */
-        NULL                            /* next pointer */
+        NULL,                   /* next pointer */
+        NON_PARALLEL(false)
 };
 

lib/format_rt.c

@@ -458 +458 @@
                                 /* This may fail on a non-Linux machine */
                                 if (trace_is_err(RT_INFO->dummy_ring)) {
-                                        trace_perror(RT_INFO->dummy_ring, "Creating dead int trace");
+                                        trace_perror(RT_INFO->dummy_ring, "Creating dead ring trace");
                                         return -1;
                                 }
@@ -863 +863 @@
         trace_event_rt,             /* trace_event */
         rt_help,                        /* help */
-        NULL                            /* next pointer */
+        NULL,                   /* next pointer */
+        NON_PARALLEL(true) /* This is normally live */
 };
 

lib/format_tsh.c

@@ -274 +274 @@
         trace_event_trace,              /* trace_event */
         tsh_help,                       /* help */
-        NULL                            /* next pointer */
+        NULL,                   /* next pointer */
+        NON_PARALLEL(false)
 };
 
@@ -322 +323 @@
         trace_event_trace,              /* trace_event */
         tsh_help,                       /* help */
-        NULL                            /* next pointer */
+        NULL,                   /* next pointer */
+        NON_PARALLEL(false)
 };
 

lib/libtrace.h.in

@@ -117 +117 @@
 /** 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)
+//#define ASSERT_RET(run, cond) run
+#endif
     
 #ifdef __cplusplus 
@@ -197 +210 @@
 #endif
 
+// Used to fight against false sharing
+#define CACHE_LINE_SIZE 64
+#define ALIGN_STRUCT(x) __attribute__((aligned(x)))
+
 #ifdef _MSC_VER
     #ifdef LT_BUILDING_DLL
@@ -225 +242 @@
 /** Opaque structure holding information about a bpf filter */
 typedef struct libtrace_filter_t libtrace_filter_t;
+
+typedef struct libtrace_thread_t libtrace_thread_t;
 
 /** If the packet has allocated its own memory the buffer_control should be
@@ -512 +531 @@
         uint8_t transport_proto;        /**< Cached transport protocol */
         uint32_t l4_remaining;          /**< Cached transport remaining */
+        uint64_t order; /**< Notes the order of this packet in relation to the input */
+        uint64_t hash; /**< A hash of the packet as supplied by the user */
+        int error; /**< The error status of pread_packet */
 } libtrace_packet_t;
 
@@ -3231 +3253 @@
 /*@}*/
 
+/**
+ * A collection of types for convenience used in place of a
+ * simple void* to allow a any type of data to be stored.
+ *
+ * This is expected to be 8 bytes in length.
+ */
+typedef union {
+        /* Pointers */
+        void *ptr;
+        libtrace_packet_t *pkt;
+
+        /* C99 Integer types */
+        /* NOTE: Standard doesn't require 64-bit
+     * but x32 and x64 gcc does */
+        int64_t sint64;
+        uint64_t uint64;
+
+        uint32_t uint32s[2];
+        int32_t sint32s[2];
+        uint32_t uint32;
+        int32_t sint32;
+
+        uint16_t uint16s[4];
+        int16_t sint16s[4];
+        uint16_t uint16;
+        int16_t sint16;
+
+        uint8_t uint8s[8];
+        int8_t sint8s[8];
+        uint8_t uint8;
+        int8_t sint8;
+
+        size_t size;
+
+        /* C basic types - we cannot be certian of the size */
+        int sint;
+        unsigned int uint;
+
+        signed char schars[8];
+        unsigned char uchars[8];
+        signed char schar;
+        unsigned char uchar;
+
+        /* Real numbers */
+        float rfloat;
+        double rdouble;
+} libtrace_generic_types_t;
+
+typedef struct libtrace_message_t {
+        int code;
+        libtrace_generic_types_t additional;
+        libtrace_thread_t *sender;
+} libtrace_message_t;
+
+/** Structure holding information about a result */
+typedef struct libtrace_result_t {
+        uint64_t key;
+        libtrace_generic_types_t value;
+        int type;
+} libtrace_result_t;
+#define RESULT_NORMAL 0
+#define RESULT_PACKET 1
+#define RESULT_TICK   2
+
+
+typedef void* (*fn_per_pkt)(libtrace_t* trace, libtrace_packet_t *p, libtrace_message_t *m, libtrace_thread_t *thread);
+typedef void (*fn_reporter)(libtrace_t* trace, libtrace_result_t *r, libtrace_message_t *m);
+typedef uint64_t (*fn_hasher)(const libtrace_packet_t* packet, void *data);
+
+DLLEXPORT int trace_pstart(libtrace_t *libtrace, void* global_blob, fn_per_pkt per_pkt, fn_reporter reporter);
+DLLEXPORT int trace_ppause(libtrace_t *libtrace);
+DLLEXPORT int trace_pstop(libtrace_t *libtrace);
+DLLEXPORT void trace_join(libtrace_t * trace);
+DLLEXPORT void print_contention_stats (libtrace_t *libtrace);
+
+DLLEXPORT void libtrace_result_set_key(libtrace_result_t * result, uint64_t key);
+DLLEXPORT uint64_t libtrace_result_get_key(libtrace_result_t * result);
+DLLEXPORT void libtrace_result_set_value(libtrace_result_t * result, libtrace_generic_types_t value);
+DLLEXPORT libtrace_generic_types_t libtrace_result_get_value(libtrace_result_t * result);
+DLLEXPORT void libtrace_result_set_key_value(libtrace_result_t * result, uint64_t key, libtrace_generic_types_t value);
+DLLEXPORT void trace_destroy_result(libtrace_result_t ** result);
+
+// Ways to access Global and TLS storage that we provide the user
+DLLEXPORT void * trace_get_global(libtrace_t *trace);
+DLLEXPORT void * trace_set_global(libtrace_t *trace, void * data);
+DLLEXPORT void * trace_set_tls(libtrace_thread_t *t, void * data);
+DLLEXPORT void * trace_get_tls(libtrace_thread_t *t);
+
+
+DLLEXPORT void trace_publish_result(libtrace_t *libtrace, libtrace_thread_t *t, uint64_t key, libtrace_generic_types_t value, int type);
+typedef struct libtrace_vector libtrace_vector_t;
+
+DLLEXPORT int trace_post_reporter(libtrace_t *libtrace);
+DLLEXPORT int libtrace_thread_get_message_count(libtrace_t * libtrace);
+DLLEXPORT int libtrace_thread_get_message(libtrace_t * libtrace, libtrace_message_t * message);
+DLLEXPORT int libtrace_thread_try_get_message(libtrace_t * libtrace, libtrace_message_t * message);
+DLLEXPORT int trace_send_message_to_reporter(libtrace_t * libtrace, libtrace_message_t * message);
+DLLEXPORT int trace_send_message_to_perpkts(libtrace_t * libtrace, libtrace_message_t * message);
+DLLEXPORT int trace_send_message_to_thread(libtrace_t * libtrace, libtrace_thread_t *t, libtrace_message_t * message);
+DLLEXPORT int trace_finished(libtrace_t * libtrace);
+DLLEXPORT uint64_t trace_packet_get_order(libtrace_packet_t * packet);
+DLLEXPORT uint64_t trace_packet_get_hash(libtrace_packet_t * packet);
+DLLEXPORT void trace_packet_set_order(libtrace_packet_t * packet, uint64_t order);
+DLLEXPORT void trace_packet_set_hash(libtrace_packet_t * packet, uint64_t hash);
+DLLEXPORT uint64_t tv_to_usec(struct timeval *tv);
+
+DLLEXPORT int retrive_first_packet(libtrace_t *libtrace, libtrace_packet_t **packet, struct timeval **tv);
+
+DLLEXPORT void libtrace_make_packet_safe(libtrace_packet_t *pkt);
+DLLEXPORT void libtrace_make_result_safe(libtrace_result_t *res);
+
+typedef enum {
+        /** 
+         * Sets the hasher function, if NULL(default) no hashing is used a
+         * cores will get packets on a first in first served basis
+         */
+        TRACE_OPTION_SET_HASHER,
+        
+        /**
+         * Libtrace set perpkt thread count
+         */
+        TRACE_OPTION_SET_PERPKT_THREAD_COUNT,
+        
+        /**
+         * Delays packets so they are played back in trace-time rather than as fast
+         * as possible.
+         */
+        TRACE_OPTION_TRACETIME,
+
+        /**
+         * Specifies the interval between tick packets in milliseconds, if 0
+         * or less this is ignored.
+         */
+        TRACE_OPTION_TICK_INTERVAL,
+        TRACE_OPTION_GET_CONFIG,
+        TRACE_OPTION_SET_CONFIG
+} trace_parallel_option_t;
+
+enum libtrace_messages {
+        MESSAGE_STARTING,
+        MESSAGE_RESUMING,
+        MESSAGE_STOPPING,
+        MESSAGE_PAUSING,
+        MESSAGE_DO_PAUSE,
+        MESSAGE_DO_STOP,
+        MESSAGE_FIRST_PACKET,
+        MESSAGE_PERPKT_ENDED,
+        MESSAGE_PERPKT_RESUMED,
+        MESSAGE_PERPKT_PAUSED,
+        MESSAGE_PERPKT_EOF,
+        MESSAGE_POST_REPORTER,
+        MESSAGE_POST_RANGE,
+        MESSAGE_TICK,
+        MESSAGE_USER
+};
+
+enum hasher_types {
+        /** 
+         * Balance load across CPUs best as possible, this is basically to say do 
+         * not care about hash. This might still might be implemented
+         * using a hash or round robin etc. under the hood depending on the format
+         */
+        HASHER_BALANCE,
+
+        /** Use a hash which is bi-directional for TCP flows, that is packets with
+         * the same hash are sent to the same thread. All non TCP packets will be
+         * sent to the same thread. UDP may or may not be sent to separate
+         * threads like TCP, this depends on the format support.
+         */
+        HASHER_BIDIRECTIONAL,
+        
+        /**
+         * Use a hash which is uni-directional across TCP flows, that means the
+         * opposite directions of the same 5 tuple might end up on separate cores.
+         * Otherwise is identical to HASHER_BIDIRECTIONAL
+         */
+        HASHER_UNIDIRECTIONAL,
+
+        /**
+         * Always use the user supplied hasher, this currently disables native
+         * support and is likely significantly slower.
+         */
+        HASHER_CUSTOM,
+
+        /**
+         * This is not a valid option, used internally only!!! TODO remove
+         * Set by the format if the hashing is going to be done in hardware
+         */
+        HASHER_HARDWARE
+};
+
+typedef struct libtrace_info_t {
+        /** 
+         * True if a live format (i.e. packets have to be tracetime).
+         * Otherwise false, indicating packets can be read as fast
+         * as possible from the format.
+         */
+        bool live; 
+
+        /**
+         * The maximum number of threads supported by a parallel trace. 1
+         * if parallel support is not native (in this case libtrace will simulate
+         * an unlimited number of threads), -1 means unlimited and 0 unknown.
+         */
+        int max_threads;
+
+        /* TODO hash fn supported list */
+
+        /* TODO consider time/clock details?? */
+} libtrace_info_t;
+
+DLLEXPORT int trace_parallel_config(libtrace_t *libtrace, trace_parallel_option_t option, void *value);
+DLLEXPORT libtrace_packet_t* trace_result_packet(libtrace_t * libtrace, libtrace_packet_t * packet);
+DLLEXPORT void trace_free_result_packet(libtrace_t * libtrace, libtrace_packet_t * packet);
+DLLEXPORT int trace_set_hasher(libtrace_t *trace, enum hasher_types type, fn_hasher hasher, void *data);
+DLLEXPORT libtrace_info_t *trace_get_information(libtrace_t * libtrace);
+
+/**
+ * Tuning the parallel sizes
+ */
+struct user_configuration {
+        // Packet memory cache settings (ocache_init) total
+        /**
+         * See diagrams, this sets the maximum size of freelist used to
+         * maintain packets and their memory buffers.
+         * NOTE setting this to less than recommend could cause deadlock a
+         * trace that manages its own packets.
+         * A unblockable error message will be printed.
+         */
+        size_t packet_cache_size;
+        /**
+         * Per thread local cache size for the packet freelist
+         */
+        size_t packet_thread_cache_size;
+        /**
+         * If true the total number of packets that can be created by a trace is limited
+         * to the packet_cache_size, otherwise once packet_cache_size is exceeded alloc
+         * and free will be used to create and free packets, this will be slower than
+         * using the freelist and could run a machine out of memory.
+         *
+         * However this does make it easier to ensure that deadlocks will not occur
+         * due to running out of packets
+         */
+        bool fixed_packet_count;
+        /**
+         * When reading from a single threaded input source to reduce
+         * lock contention a 'burst' of packets is read per pkt thread
+         * this determines the bursts size.
+         */
+        size_t burst_size;
+        // Each perpkt thread has a queue leading into the reporter
+        //size_t reporter_queue_size;
+
+        /**
+         * The tick interval - in milliseconds
+         * When a live trace is used messages are sent at the tick
+         * interval to ensure that all perpkt threads receive data
+         * this allows results to be printed in cases flows are
+         * not being directed to a certian thread, while still
+         * maintaining order.
+         */
+        size_t tick_interval;
+
+        /**
+         * Like the tick interval but used in the case of file format
+         * This specifies the number of packets before inserting a tick to
+         * every thread.
+         */
+        size_t tick_count;
+
+        /**
+         * The number of per packet threads requested, 0 means use default.
+         * Default typically be the number of processor threads detected less one or two.
+         */
+        size_t perpkt_threads;
+
+        /**
+         * See diagrams, this sets the maximum size of buffers used between
+         * the single hasher thread and the buffer.
+         * NOTE setting this to less than recommend could cause deadlock a
+         * trace that manages its own packets.
+         * A unblockable warning message will be printed to stderr in this case.
+         */
+        /** The number of packets that can queue per thread from hasher thread */
+        size_t hasher_queue_size;
+
+        /**
+         * If true use a polling hasher queue, that means that we will spin/or yeild
+         * when rather than blocking on a lock. This applies to both the hasher thread
+         * and perpkts reading the queues.
+         */
+        bool hasher_polling;
+
+        /**
+         * If true the reporter thread will continuously poll waiting for results
+         * if false they are only checked when a message is received, this message
+         * is controlled by reporter_thold.
+         */
+        bool reporter_polling;
+
+        /**
+         * Perpkt thread result queue size before triggering the reporter step to read results
+         */
+        size_t reporter_thold;
+
+        /**
+         * Prints a line to standard error for every state change
+         * for both the trace as a whole and for each thread.
+         */
+        bool debug_state;
+};
+#include <stdio.h>
+DLLEXPORT void parse_user_config(struct user_configuration* uc, char * str);
+DLLEXPORT void parse_user_config_file(struct user_configuration* uc, FILE *file);
+DLLEXPORT int libtrace_get_perpkt_count(libtrace_t* t);
+
+/**
+ * The methods we use to combine multiple outputs into a single output
+ * This is not considered a stable API however is public.
+ * Where possible use built in combiners
+ *
+ * NOTE this structure is duplicated per trace and as such can
+ * have functions rewritten, and in fact should if possible.
+ */
+typedef struct libtrace_combine libtrace_combine_t;
+struct libtrace_combine {
+
+        /**
+         * Called at the start of the trace to allow datastructures
+         * to be initilised and allow functions to be swapped if approriate.
+         *
+         * Also factors such as whether the trace is live or not can
+         * be used to determine the functions used.
+         * @return 0 if successful, -1 if an error occurs
+         */
+        int (*initialise)(libtrace_t *,libtrace_combine_t *);
+
+        /**
+         * Called when the trace ends, clean up any memory here
+         * from libtrace_t * init.
+         */
+        void (*destroy)(libtrace_t *, libtrace_combine_t *);
+
+        /**
+         * Publish a result against it's a threads queue.
+         * If null publish directly, expected to be used
+         * as a single threaded optimisation and can be
+         * set to NULL by init if this case is detected.
+         */
+        void (*publish)(libtrace_t *, int thread_id, libtrace_combine_t *, libtrace_result_t *);
+
+        /**
+         * Read as many results as possible from the trace.
+         * Directy calls the users code to handle results from here.
+         *
+         * THIS SHOULD BE NON-BLOCKING AND READ AS MANY AS POSSIBLE
+         * If publish is NULL, this probably should be NULL also otherwise
+         * it will not be called.
+         */
+        void (*read)(libtrace_t *, libtrace_combine_t *);
+
+        /**
+         * Called when the trace is finished to flush the final
+         * results to the reporter thread.
+         *
+         * There may be no results, in which case this should
+         * just return.
+         *
+         * Libtrace state:
+         * Called from reporter thread
+         * No perpkt threads will be running, i.e. publish will not be
+         * called again.
+         *
+         * If publish is NULL, this probably should be NULL also otherwise
+         * it will not be called.
+         */
+        void (*read_final)(libtrace_t *, libtrace_combine_t *);
+
+        /**
+         * Pause must make sure any results of the type packet are safe.
+         * That means trace_copy_packet() and destroy the original.
+         * This also should be NULL if publish is NULL.
+         */
+        void (*pause)(libtrace_t *, libtrace_combine_t *);
+
+        /**
+         * Data storage for all the combiner threads
+         */
+        void *queues;
+
+        /**
+         * Configuration options, what this does is upto the combiner
+         * chosen.
+         */
+        libtrace_generic_types_t configuration;
+};
+
+DLLEXPORT void trace_set_combiner(libtrace_t *trace, const libtrace_combine_t *combiner, libtrace_generic_types_t config);
+
+#define READ_EOF 0
+#define READ_ERROR -1
+#define READ_MESSAGE -2
+// Used for inband tick message
+#define READ_TICK -3
+
+#define ZERO_USER_CONFIG(config) memset(&config, 0, sizeof(struct user_configuration));
+
 #ifdef __cplusplus
 } /* extern "C" */

lib/libtrace_int.h

@@ -148 +148 @@
 #endif
 
+#include "data-struct/ring_buffer.h"
+#include "data-struct/object_cache.h"
+#include "data-struct/vector.h"
+#include "data-struct/message_queue.h"
+#include "data-struct/deque.h"
+#include "data-struct/linked_list.h"
+#include "data-struct/sliding_window.h"
 
 //#define RP_BUFSIZE 65536U
@@ -166 +173 @@
         bool waiting;
 };
+
+enum thread_types {
+        THREAD_EMPTY,
+        THREAD_HASHER,
+        THREAD_PERPKT,
+        THREAD_REPORTER,
+        THREAD_KEEPALIVE
+};
+
+enum thread_states {
+        THREAD_RUNNING,
+        THREAD_FINISHING,
+        THREAD_FINISHED,
+        THREAD_PAUSED,
+        THREAD_STATE_MAX
+};
+
+/**
+ * Information of this thread
+ */
+struct libtrace_thread_t {
+        uint64_t accepted_packets; // The number of packets accepted only used if pread
+        uint64_t filtered_packets;
+        // is retreving packets
+        // Set to true once the first packet has been stored
+        bool recorded_first;
+        // For thread safety reason we actually must store this here
+        int64_t tracetime_offset_usec;
+        void* user_data; // TLS for the user to use
+        void* format_data; // TLS for the format to use
+        libtrace_message_queue_t messages; // Message handling
+        libtrace_ringbuffer_t rbuffer; // Input
+        libtrace_t * trace;
+        void* ret;
+        enum thread_types type;
+        enum thread_states state;
+        pthread_t tid;
+        int perpkt_num; // A number from 0-X that represents this perpkt threads number
+                                // in the table, intended to quickly identify this thread
+                                // -1 represents NA (such as the case this is not a perpkt thread)
+};
+
+/**
+ * Storage to note time value against each.
+ * Used both internally to do trace time playback
+ * and can be used externally to assist applications which need
+ * a trace starting time such as tracertstats.
+ */
+struct first_packets {
+        pthread_spinlock_t lock;
+        size_t count; // If == perpkt_thread_count threads we have all
+        size_t first; // Valid if count != 0
+        struct __packet_storage_magic_type {
+                libtrace_packet_t * packet;
+                struct timeval tv;
+        } * packets;
+};
+
+#define TRACE_STATES \
+        X(STATE_NEW) \
+        X(STATE_RUNNING) \
+        X(STATE_PAUSING) \
+        X(STATE_PAUSED) \
+        X(STATE_FINSHED) \
+        X(STATE_DESTROYED) \
+        X(STATE_JOINED) \
+        X(STATE_ERROR)
+
+#define X(a) a,
+enum trace_state {
+        TRACE_STATES
+};
+#undef X
+
+#define X(a) case a: return #a;
+static inline char *get_trace_state_name(enum trace_state ts){
+        switch(ts) {
+                TRACE_STATES
+                default:
+                        return "UNKNOWN";
+        }
+}
+#undef X
 
 /** A libtrace input trace 
@@ -188 +278 @@
         uint64_t filtered_packets;      
         /** The filename from the uri for the trace */
-        char *uridata;                  
+        char *uridata;
         /** The libtrace IO reader for this trace (if applicable) */
-        io_t *io;                       
+        io_t *io;
         /** Error information for the trace */
-        libtrace_err_t err;             
+        libtrace_err_t err;
         /** Boolean flag indicating whether the trace has been started */
-        bool started;                   
+        bool started;
+        /** Synchronise writes/reads across this format object and attached threads etc */
+        pthread_mutex_t libtrace_lock;
+        /** State */
+        enum trace_state state;
+        /** Use to control pausing threads and finishing threads etc always used with libtrace_lock */
+        pthread_cond_t perpkt_cond;
+        /* Keep track of counts of threads in any given state */
+        int perpkt_thread_states[THREAD_STATE_MAX]; 
+
+        /** Set to indicate a perpkt's queue is full as such the writing perpkt cannot proceed */
+        bool perpkt_queue_full;
+        /** Global storage for this trace, shared among all the threads  */
+        void* global_blob;
+        /** The actual freelist */
+        libtrace_ocache_t packet_freelist;
+        /** User defined per_pkt function called when a pkt is ready */
+        fn_per_pkt per_pkt;
+        /** User defined reporter function entry point XXX not hooked up */
+        fn_reporter reporter;
+        /** The hasher function */
+        enum hasher_types hasher_type;
+        /** The hasher function - NULL implies they don't care or balance */
+        fn_hasher hasher; // If valid using a separate thread
+        void *hasher_data;
+        /** The pread_packet choosen path for the configuration */
+        int (*pread)(libtrace_t *, libtrace_thread_t *, libtrace_packet_t **, size_t);
+
+        libtrace_thread_t hasher_thread;
+        libtrace_thread_t reporter_thread;
+        libtrace_thread_t keepalive_thread;
+        int perpkt_thread_count;
+        libtrace_thread_t * perpkt_threads; // All our perpkt threads
+        // Used to keep track of the first packet seen on each thread
+        struct first_packets first_packets;
+        int tracetime;
+
+        /*
+         * Caches statistic counters in the case that our trace is
+         * paused or stopped before this counter is taken
+         */
+        uint64_t dropped_packets;
+        uint64_t received_packets;
+        struct user_configuration config;
+        libtrace_combine_t combiner;
 };
+
+void trace_fin_packet(libtrace_packet_t *packet);
+void libtrace_zero_thread(libtrace_thread_t * t);
+void store_first_packet(libtrace_t *libtrace, libtrace_packet_t *packet, libtrace_thread_t *t);
+libtrace_thread_t * get_thread_table(libtrace_t *libtrace);
+int get_thread_table_num(libtrace_t *libtrace);
+
 
 /** A libtrace output trace
@@ -202 +343 @@
 struct libtrace_out_t {
         /** The capture format for the output trace */
-        struct libtrace_format_t *format;
+        struct libtrace_format_t *format;
         /** Pointer to the "global" data for the capture format module */
         void *format_data;              
@@ -210 +351 @@
         libtrace_err_t err;
         /** Boolean flag indicating whether the trace has been started */
-        bool started;                   
+        bool started;
 };
 
@@ -303 +444 @@
 } PACKED libtrace_pflog_header_t;
 
-
-
 /** A libtrace capture format module */
 /* All functions should return -1, or NULL on failure */
@@ -734 +873 @@
         /** Prints some useful help information to standard output. */
         void (*help)(void);
-
+        
         /** Next pointer, should always be NULL - used by the format module
          * manager. */
         struct libtrace_format_t *next;
+
+        /** Holds information about the trace format */
+        struct libtrace_info_t info;
+
+        /**
+         * Starts or unpauses an input trace in parallel mode - note that
+         * this function is often the one that opens the file or device for
+         * reading.
+         *
+         * @param libtrace      The input trace to be started or unpaused
+         * @return 0 upon success.
+         *         Otherwise in event of an error -1 is returned.
+         * 
+         */
+        int (*pstart_input)(libtrace_t *trace);
+        
+        /**
+         * Read a batch of packets from the input stream related to thread.
+         * At most read nb_packets, however should return with less if packets
+         * are not waiting. However still must return at least 1, 0 still indicates
+         * EOF.
+         *
+         * @param libtrace      The input trace
+         * @param t     The thread
+         * @param packets       An array of packets
+         * @param nb_packets    The number of packets in the array (the maximum to read)
+         * @return The number of packets read, or 0 in the case of EOF or -1 in error or -2 to represent
+         * interrupted due to message waiting before packets had been read.
+         */
+        int (*pread_packets)(libtrace_t *trace, libtrace_thread_t *t, libtrace_packet_t **packets, size_t nb_packets);
+        
+        /** Pause a parallel trace
+         *
+         * @param libtrace      The input trace to be paused
+         */
+        int (*ppause_input)(libtrace_t *trace);
+        
+        /** Called after all threads have been paused, Finish (close) a parallel trace
+         *
+         * @param libtrace      The input trace to be stopped
+         */
+        int (*pfin_input)(libtrace_t *trace);
+        
+        /** Applies a configuration option to an input trace.
+         *
+         * @param libtrace      The input trace to apply the option to
+         * @param option        The option that is being configured
+         * @param value         A pointer to the value that the option is to be
+         *                      set to
+         * @return 0 if successful, -1 if the option is unsupported or an error
+         * occurs
+         */
+        int (*pconfig_input)(libtrace_t *libtrace,trace_parallel_option_t option,void *value);
+
+        /**
+         * Register a thread for use with the format or using the packets produced
+         * by it. This is NOT only used for threads reading packets in fact all
+         * threads use this.
+         *
+         * The libtrace lock is not held by this format but can be aquired
+         * by the format.
+         *
+         * Some use cases include setting up any thread local storage required for
+         * to read packets and free packets. For DPDK we require any thread that
+         * may release or read a packet to have have an internal number associated
+         * with it.
+         * 
+         * The thread type can be used to see if this thread is going to be used
+         * to read packets or otherwise.
+         *
+         * @return 0 if successful, -1 if the option is unsupported or an error
+         * occurs (such as a maximum of threads being reached)
+         */
+        int (*pregister_thread)(libtrace_t *libtrace, libtrace_thread_t *t, bool reader);
+
+        /**
+         * If needed any memory allocated with pregister_thread can be released
+         * in this function. The thread will be destroyed directly after this
+         * function is called.
+         */
+        void (*punregister_thread)(libtrace_t *libtrace, libtrace_thread_t *t);
 };
+
+/** Macro to zero out a single thread format */
+#define NON_PARALLEL(live) \
+{live, 1},              /* trace info */ \
+NULL,                   /* pstart_input */ \
+NULL,                   /* pread_packet */ \
+NULL,                   /* ppause_input */ \
+NULL,                   /* pfin_input */ \
+NULL,                   /* pconfig_input */ \
+NULL,                   /* pregister_thread */ \
+NULL                    /* punregister_thread */
 
 /** The list of registered capture formats */
@@ -943 +1174 @@
 /** Constructor for the Linux Native format module */
 void linuxnative_constructor(void);
+/** Constructor for the Linux Ring format module */
+void linuxring_constructor(void);
 /** Constructor for the PCAP format module */
 void pcap_constructor(void);

lib/trace.c

@@ -99 +99 @@
 #include "rt_protocol.h"
 
+#include <pthread.h>
+#include <signal.h>
+
 #define MAXOPTS 1024
 
@@ -106 +109 @@
 
 volatile int libtrace_halt = 0;
+/* Set once pstart is called used for backwards compatibility reasons */
+int libtrace_parallel = 0;
 
 /* strncpy is not assured to copy the final \0, so we
@@ -137 +142 @@
                 legacy_constructor();
                 atmhdr_constructor();
+                linuxring_constructor();
                 linuxnative_constructor();
 #ifdef HAVE_LIBPCAP
@@ -253 +259 @@
         libtrace->filtered_packets = 0;
         libtrace->accepted_packets = 0;
+        
+        /* Parallel inits */
+        ASSERT_RET(pthread_mutex_init(&libtrace->libtrace_lock, NULL), == 0);
+        ASSERT_RET(pthread_cond_init(&libtrace->perpkt_cond, NULL), == 0);
+        libtrace->state = STATE_NEW;
+        libtrace->perpkt_queue_full = false;
+        libtrace->global_blob = NULL;
+        libtrace->per_pkt = NULL;
+        libtrace->reporter = NULL;
+        libtrace->hasher = NULL;
+        libtrace_zero_ocache(&libtrace->packet_freelist);
+        libtrace_zero_thread(&libtrace->hasher_thread);
+        libtrace_zero_thread(&libtrace->reporter_thread);
+        libtrace_zero_thread(&libtrace->keepalive_thread);
+        libtrace->reporter_thread.type = THREAD_EMPTY;
+        libtrace->perpkt_thread_count = 0;
+        libtrace->perpkt_threads = NULL;
+        libtrace->tracetime = 0;
+        libtrace->first_packets.first = 0;
+        libtrace->first_packets.count = 0;
+        libtrace->first_packets.packets = NULL;
+        libtrace->dropped_packets = UINT64_MAX;
+        libtrace->received_packets = UINT64_MAX;
+        libtrace->pread = NULL;
+        ZERO_USER_CONFIG(libtrace->config);
 
         /* Parse the URI to determine what sort of trace we are dealing with */
@@ -348 +379 @@
         libtrace->io = NULL;
         libtrace->filtered_packets = 0;
+        
+        /* Parallel inits */
+        ASSERT_RET(pthread_mutex_init(&libtrace->libtrace_lock, NULL), == 0);
+        ASSERT_RET(pthread_cond_init(&libtrace->perpkt_cond, NULL), == 0);
+        libtrace->state = STATE_NEW; // TODO MAYBE DEAD
+        libtrace->perpkt_queue_full = false;
+        libtrace->global_blob = NULL;
+        libtrace->per_pkt = NULL;
+        libtrace->reporter = NULL;
+        libtrace->hasher = NULL;
+        libtrace_zero_ocache(&libtrace->packet_freelist);
+        libtrace_zero_thread(&libtrace->hasher_thread);
+        libtrace_zero_thread(&libtrace->reporter_thread);
+        libtrace_zero_thread(&libtrace->keepalive_thread);
+        libtrace->reporter_thread.type = THREAD_EMPTY;
+        libtrace->perpkt_thread_count = 0;
+        libtrace->perpkt_threads = NULL;
+        libtrace->tracetime = 0;
+        libtrace->pread = NULL;
+        ZERO_USER_CONFIG(libtrace->config);
         
         for(tmp=formats_list;tmp;tmp=tmp->next) {
@@ -583 +634 @@
  */
 DLLEXPORT void trace_destroy(libtrace_t *libtrace) {
-        assert(libtrace);
+        int i;
+        assert(libtrace);
+
+        ASSERT_RET(pthread_mutex_destroy(&libtrace->libtrace_lock), == 0);
+        ASSERT_RET(pthread_cond_destroy(&libtrace->perpkt_cond), == 0);
+
+        /* destroy any packets that are still around */
+        if (libtrace->state != STATE_NEW && libtrace->first_packets.packets) {