Changeset 6463f32

Timestamp:

03/01/17 11:48:37 (4 years ago)

Author:

Shane Alcock <salcock@…>

Branches:

cachetimestamps, develop, dpdk-ndag, etsilive, master, rc-4.0.3, rc-4.0.4, ringdecrementfix, ringperformance

Children:

14b7489

Parents:

d0a067f (diff), 204d65b (diff)
Note: this is a merge changeset, the changes displayed below correspond to the merge itself.
Use the (diff) links above to see all the changes relative to each parent.

Message:

Merge branch 'develop'

Files:

• .gitignore (modified) (1 diff)
• AUTHORS (modified) (2 diffs)
• README (modified) (1 diff)
• configure.in (modified) (4 diffs)
• debian/autogen.sh (added)
• debian/changelog (added)
• debian/compat (added)
• debian/control (added)
• debian/copyright (added)
• debian/libpacketdump4-dev.dirs (added)
• debian/libpacketdump4-dev.install (added)
• debian/libpacketdump4.dirs (added)
• debian/libpacketdump4.install (added)
• debian/libtrace4-dev.dirs (added)
• debian/libtrace4-dev.install (added)
• debian/libtrace4-tools.dirs (added)
• debian/libtrace4-tools.install (added)
• debian/libtrace4.dirs (added)
• debian/libtrace4.install (added)
• debian/rules (added)
• debian/source/format (added)
• lib/Makefile.am (modified) (1 diff)
• lib/format_bpf.c (modified) (1 diff)
• lib/format_dag25.c (modified) (3 diffs)
• lib/format_dpdk.c (modified) (26 diffs)
• lib/format_linux_common.c (modified) (1 diff)
• lib/format_linux_common.h (modified) (2 diffs)
• lib/format_linux_int.c (modified) (3 diffs)
• lib/format_linux_ring.c (modified) (5 diffs)
• lib/format_pcap.c (modified) (1 diff)
• lib/hash_toeplitz.c (modified) (3 diffs)
• lib/hash_toeplitz.h (modified) (1 diff)
• lib/libtrace.h.in (modified) (17 diffs)
• lib/libtrace_int.h (modified) (3 diffs)
• lib/protocols_l2.c (modified) (2 diffs)
• lib/protocols_transport.c (modified) (2 diffs)
• lib/trace.c (modified) (11 diffs)
• lib/trace_parallel.c (modified) (11 diffs)
• test/do-test-build-dpdk.sh (added)
• test/test-live.c (modified) (1 diff)
• test/test-tracetime-parallel.c (modified) (1 diff)
• tools/tracertstats/tracertstats.c (modified) (2 diffs)

.gitignore

@@ -81 +81 @@
 test/test-wireless
 test/test-write
+test/DPDK_source/
+test/DPDK_builds/
 tools/traceanon/traceanon
 tools/traceanon/traceanon_parallel

AUTHORS

@@ -15 +15 @@
 
  * Matt Brown for getting us into Debian and keeping us up to date.
+ * Brad Cowie for packaging libtrace4 for Debian and Ubuntu
  * Alistair King for reporting and suggesting fixes for numerous bugs, 
    particularly in libwandio
@@ -27 +28 @@
  * Robert Edmonds for tidying up the libpacketdump plugin install
  * Martin Bligh for patching in support for nanosecond pcap traces
+ * Teemu Rytilahti for adding support for SIT
+ * "EaseTheWorld" for their work with improving the packet statistics API, as
+   well as reporting several bugs
+ * Richard Cziva for contributing to the DPDK support
  * Jamie Curtis for fixing a couple of bugs many many years ago
  * Brendon Jones for creating the original Windows DLLs and writing bits of

README

@@ -1 @@
-libtrace 4.0.0
+libtrace 4.0.1
 
 ---------------------------------------------------------------------------
-Copyright (c) 2007-2016 The University of Waikato, Hamilton, New Zealand.
+Copyright (c) 2007-2017 The University of Waikato, Hamilton, New Zealand.
 All rights reserved.
 

configure.in

@@ -4 +4 @@
 # and in the README
 
-AC_INIT([libtrace],[4.0.0],[contact@wand.net.nz],[libtrace])
+AC_INIT([libtrace],[4.0.1],[contact@wand.net.nz],[libtrace])
 
 LIBTRACE_MAJOR=4
 LIBTRACE_MID=0
-LIBTRACE_MINOR=0
+LIBTRACE_MINOR=1
 
 # OpenSolaris hides libraries like libncurses in /usr/gnu/lib, which is not
@@ -387 +387 @@
 # Check for DPDK 
 AC_ARG_WITH(dpdk,
-            AS_HELP_STRING(--with-dpdk,include DPDK live capture support (Must set RTE_SDK/_TARGET environment variable)),
+            AS_HELP_STRING(--with-dpdk,include DPDK live capture support (From either the RTE_SDK/_TARGET environment variable or the dpdk-dev package)),
 [
         if test "$withval" = no
@@ -402 +402 @@
 libtrace_dpdk=false
 if test "$want_dpdk" != no; then
-    saved_ldflags="$LDFLAGS"
-    # Add to our libpath location
-    LDFLAGS="$LDFLAGS -L$RTE_SDK/$RTE_TARGET/lib/"
-    # Test that the library exists, we cannot be sure of what extra library flags need to be added
-    # this could change in the next version which would make the test fail
-    #AC_CHECK_LIB(intel_dpdk, rte_eal_init, dpdk_found=1, dpdk_found=0, -lpthread -lm -lrt)
-
-    # So instead simply check for existence
-    AC_CHECK_FILE("$RTE_SDK/$RTE_TARGET/lib/libintel_dpdk.a", dpdk_found=1, dpdk_found=0)
-    # Revert back
-    LDFLAGS="$saved_ldflags"
-    if test "$dpdk_found" = 1; then
-        # Save these now so that they can be re-exported later
+        # So instead simply check for existence
+        if test "$RTE_SDK" != ""; then
+                AC_CHECK_FILE("$RTE_SDK/$RTE_TARGET/lib/libintel_dpdk.a", dpdk_found="libintel_dpdk.a", dpdk_found=0)
+        fi
+        # DPDK 2.1.0+ renames this to libdpdk from libintel_dpdk
+        if test "$dpdk_found" = 0 -a "$RTE_SDK" != ""; then
+                AC_CHECK_FILE("$RTE_SDK/$RTE_TARGET/lib/libdpdk.a", dpdk_found="libdpdk.a", dpdk_found=0)
+        fi
+        if test "$dpdk_found" != 0 -a "$RTE_SDK" != ""; then
+                # Save these now so that they can be re-exported later
                 AC_SUBST([RTE_TARGET])
-        AC_SUBST([RTE_SDK])
-        # Force dpdk library to be statically linked to allow compiler optimisations
-        LIBTRACE_LIBS="$LIBTRACE_LIBS -Wl,--whole-archive -Wl,-l:libintel_dpdk.a -Wl,--no-whole-archive -Wl,-lm"
+                AC_SUBST([RTE_SDK])
+                # Force dpdk library to be statically linked to allow compiler optimisations
+                LIBTRACE_LIBS="$LIBTRACE_LIBS -Wl,--whole-archive -Wl,-l:$dpdk_found -Wl,--no-whole-archive -Wl,-lm"
                 AC_DEFINE(HAVE_DPDK,1,[conditional for building with DPDK live capture support])
                 libtrace_dpdk=true
+        fi
+        if test "$RTE_SDK" = ""; then
+                AC_MSG_NOTICE([No RTE_SDK given, checking for system dpdk-dev package])
+                # Search the system, maybe it is installed? Ethdev is one of the main libraries
+                AC_CHECK_LIB(ethdev, rte_eth_dev_configure, dpdk_found="system", dpdk_found=0)
+                # We also need to check that rte.vars.mk is installed from dpdk-dev (as well as libdpdk-dev)
+                if test "$dpdk_found" != 0 -a -e /usr/share/dpdk/mk/rte.vars.mk ; then
+                        RTE_TARGET="x86_64-default-linuxapp-gcc"
+                        RTE_SDK="/usr/share/dpdk/"
+                        RTE_INCLUDE="/usr/include/dpdk"
+                        AC_SUBST([RTE_TARGET])
+                        AC_SUBST([RTE_SDK])
+                        AC_SUBST([RTE_INCLUDE])
+                        # include libethdev this is part of DPDK
+                        LIBTRACE_LIBS="$LIBTRACE_LIBS -lethdev"
+                        # Include all rte libs note we need to escape '[' and ']'
+                        LIBTRACE_LIBS="$LIBTRACE_LIBS $(ldconfig -p | sed 's/.*lib\([[^.]]*\).*/-l\1/' | grep rte_ | tr '\n' ' ')"
+                        AC_MSG_NOTICE([Building against system DPDK])
+
+                        AC_DEFINE(HAVE_DPDK,1,[conditional for building with DPDK live capture support])
+                        libtrace_dpdk=true
+                fi
         fi
 fi
@@ -471 +490 @@
 if test "$ac_cv_search_dlopen" != "none required"; then 
         LIBPKTDUMP_LIBS="$LIBPKTDUMP_LIBS $ac_cv_search_dlopen"
-        if test "$dpdk_found" = 1; then
+        if test "$dpdk_found" != 0; then
                 LIBTRACE_LIBS="$LIBTRACE_LIBS -Wl,$ac_cv_search_dlopen"
         fi

lib/Makefile.am

@@ -1 +1 @@
 lib_LTLIBRARIES = libtrace.la
-include_HEADERS = libtrace.h dagformat.h lt_inttypes.h daglegacy.h \
-        rt_protocol.h erftypes.h libtrace_parallel.h \
+include_HEADERS = libtrace.h libtrace_parallel.h
+pkginclude_HEADERS = dagformat.h lt_inttypes.h daglegacy.h \
+        rt_protocol.h erftypes.h \
         data-struct/ring_buffer.h data-struct/object_cache.h \
-        data-struct/vector.h data-struct/message_queue.h \
+        data-struct/vector.h \
         data-struct/deque.h data-struct/linked_list.h \
-        data-struct/sliding_window.h hash_toeplitz.h \
-        data-struct/buckets.h
+        data-struct/buckets.h data-struct/sliding_window.h \
+        data-struct/message_queue.h hash_toeplitz.h
 
 AM_CFLAGS=@LIBCFLAGS@ @CFLAG_VISIBILITY@ -pthread

lib/format_bpf.c

@@ -497 +497 @@
 
                 /* Timed out -- check if we should halt */
-                if (libtrace_halt)
-                        return 0;
+                if ((ret=is_halted(libtrace)) != -1)
+                        return ret;
         }
         

lib/format_dag25.c

@@ -1304 +1304 @@
                                 return -2;
 
-                        if (libtrace_halt)
-                                return 0;
+                        if ((numbytes=is_halted(libtrace)) != -1)
+                                return numbytes;
                         /* Block until we see a packet */
                         continue;
@@ -1313 +1313 @@
 
         packet->trace = libtrace;
+
         /* Prepare the libtrace packet */
         if (dag_prepare_packet_stream(libtrace, stream_data, packet, erfptr,
@@ -1324 +1325 @@
         }
 
+        packet->order = erf_get_erf_timestamp(packet);
         packet->error = packet->payload ? htons(erfptr->rlen) :
                                           erf_get_framing_length(packet);

lib/format_dpdk.c

@@ -23 +23 @@
  *
  *
- */
+ *
  * Kit capture format.
  *
@@ -69 +69 @@
  * code (that we still attempt to support).
  *
- * DPDK v1.7.1 is recommended.
- * However 1.5 to 1.8 are likely supported.
+ * DPDK 16.04 or newer is recommended.
+ * However 1.6 and newer are still likely supported.
  */
 #include <rte_eal.h>
@@ -145 +145 @@
 #else
 #       define DPDK_USE_NULL_QUEUE_CONFIG 0
+#endif
+
+/* 2.0.0-rc1
+ * Unifies RSS hash between cards
+ */
+#if RTE_VERSION >= RTE_VERSION_NUM(2, 0, 0, 1)
+#       define RX_RSS_FLAGS (ETH_RSS_IP | ETH_RSS_UDP | ETH_RSS_TCP | \
+                             ETH_RSS_SCTP)
+#else
+#       define RX_RSS_FLAGS (ETH_RSS_IPV4_UDP | ETH_RSS_IPV6 | ETH_RSS_IPV4 | \
+                             ETH_RSS_IPV4_TCP | ETH_RSS_IPV6_TCP |\
+                             ETH_RSS_IPV6_UDP)
+#endif
+
+/* v16.07-rc1 - deprecated
+ * rte_mempool_avail_count to replace rte_mempool_count
+ * rte_mempool_in_use_count to replace rte_mempool_free_count
+ */
+#if RTE_VERSION < RTE_VERSION_NUM(16, 7, 0, 1)
+#define rte_mempool_avail_count rte_mempool_count
+#define rte_mempool_in_use_count rte_mempool_free_count
 #endif
 
@@ -170 +191 @@
 #endif
 
+/* 16.04-rc3 ETH_LINK_SPEED_X are replaced with ETH_SPEED_NUM_X.
+ * ETH_LINK_SPEED_ are reused as flags, ugly.
+ * We use the new way in this code.
+ */
+#ifndef ETH_SPEED_NUM_1G
+        #define ETH_SPEED_NUM_1G ETH_LINK_SPEED_1000
+        #define ETH_SPEED_NUM_10G ETH_LINK_SPEED_10G
+        #define ETH_SPEED_NUM_20G ETH_LINK_SPEED_20G
+        #define ETH_SPEED_NUM_40G ETH_LINK_SPEED_40G
+#endif
 
 /* The default size of memory buffers to use - This is the max size of standard
@@ -175 +206 @@
 #define RX_MBUF_SIZE 1514
 
-/* 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.
- */
-#define MIN_NB_BUF 64
+/* The minimum number of memory buffers per queue tx or rx. Based on
+ * the requirement of the memory pool with 128 per thread buffers, needing
+ * at least 128*1.5 = 192 buffers. Our code allocates 128*2 to be safe.
+ */
+#define MIN_NB_BUF 128
 
 /* Number of receive memory buffers to use
@@ -185 +217 @@
  * This can be increased in the driver and here.
  * Should be at least MIN_NB_BUF.
- */
-#define NB_RX_MBUF 4096
+ * We choose 2K rather than 4K because it enables the usage of sse vector
+ * drivers which are significantly faster than using the larger buffer.
+ */
+#define NB_RX_MBUF (4096/2)
 
 /* Number of send memory buffers to use.
@@ -203 +237 @@
 /* For single threaded libtrace we read packets as a batch/burst
  * this is the maximum size of said burst */
-#define BURST_SIZE 50
+#define BURST_SIZE 32
 
 #define MBUF(x) ((struct rte_mbuf *) x)
@@ -229 +263 @@
  *
  * Make sure you understand what these are doing before enabling them.
- * They might make traces incompatable with other builds etc.
+ * They might make traces incompatible with other builds etc.
  *
  * These are also included to show how to do somethings which aren't
@@ -240 +274 @@
 /* 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
+ * otherwise could be a large overhead. Again gettimeofday() should be
  * vsyscall also if it's not you should seriously consider updating your
  * kernel.
@@ -297 +331 @@
         int lcore;
 #if HAS_HW_TIMESTAMPS_82580
-        /* Timestamping only relevent to RX */
+        /* Timestamping only relevant 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 */
@@ -330 +364 @@
 #endif
         char mempool_name[MEMPOOL_NAME_LEN]; /* The name of the mempool that we are using */
-        uint8_t rss_key[40]; // This is the RSS KEY
+        enum hasher_types hasher_type;
+        uint8_t *rss_key;
         /* To improve single-threaded performance we always batch reading
          * packets, in a burst, otherwise the parallel library does this for us */
@@ -822 +857 @@
         FORMAT(libtrace)->burst_size = 0;
         FORMAT(libtrace)->burst_offset = 0;
+        FORMAT(libtrace)->hasher_type = HASHER_BALANCE;
+        FORMAT(libtrace)->rss_key = NULL;
 
         /* Make our first stream */
@@ -899 +936 @@
                 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);
+                        FORMAT(libtrace)->hasher_type = *(enum hasher_types*)data;
+                        if (FORMAT(libtrace)->rss_key)
+                                free(FORMAT(libtrace)->rss_key);
+                        FORMAT(libtrace)->rss_key = NULL;
                         return 0;
                 case HASHER_CUSTOM:
-                        // We don't support these
+                        // Let libtrace do this
                         return -1;
                 }
@@ -958 +996 @@
         .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,
+                        .rss_hf = RX_RSS_FLAGS,
                 },
         },
@@ -1087 +1124 @@
         int i;
         struct rte_config *cfg = rte_eal_get_configuration();
+        (void) socket;
 
         pthread_mutex_lock(&dpdk_lock);
@@ -1246 +1284 @@
                 fprintf(stderr, "DPDK memory pool not empty %d of %d, please "
                         "free all packets before finishing a trace\n",
-                        rte_mempool_count(mempool), mempool->size);
+                        rte_mempool_avail_count(mempool), mempool->size);
         }
 
@@ -1315 +1353 @@
                         return -1;
                 }
+        }
+
+        /* Generate the hash key, based on the device */
+        uint8_t rss_size = 52; // 52 for i40e, 40 for others, use the largest by default
+        // In new versions DPDK we can query the size
+#if RTE_VERSION >= RTE_VERSION_NUM(2, 1, 0, 0)
+        struct rte_eth_dev_info dev_info;
+        rte_eth_dev_info_get(format_data->port, &dev_info);
+        rss_size = dev_info.hash_key_size;
+#endif
+        if (rss_size != 0) {
+                format_data->rss_key = malloc(rss_size);
+                if (format_data->hasher_type == HASHER_BIDIRECTIONAL) {
+                        toeplitz_ncreate_bikey(format_data->rss_key, rss_size);
+                } else {
+                        toeplitz_ncreate_unikey(format_data->rss_key, rss_size);
+                }
+                port_conf.rx_adv_conf.rss_conf.rss_key = format_data->rss_key;
+#if RTE_VERSION >= RTE_VERSION_NUM(1, 7, 0, 1)
+                port_conf.rx_adv_conf.rss_conf.rss_key_len = rss_size;
+#endif
+        } else {
+                fprintf(stderr, "DPDK couldn't configure RSS hashing!");
         }
 
@@ -1720 +1781 @@
                 libtrace_list_deinit(FORMAT(libtrace)->per_stream);
                 /* filter here if we used it */
+                if (FORMAT(libtrace)->rss_key)
+                        free(FORMAT(libtrace)->rss_key);
                 free(libtrace->format_data);
         }
@@ -1842 +1905 @@
 retry_calc_wiretime:
         switch (format_data->link_speed) {
-        case ETH_LINK_SPEED_40G:
-                wire_time /=  ETH_LINK_SPEED_40G;
+        case ETH_SPEED_NUM_40G:
+                wire_time /=  ETH_SPEED_NUM_40G;
                 break;
-        case ETH_LINK_SPEED_20G:
-                wire_time /= ETH_LINK_SPEED_20G;
+        case ETH_SPEED_NUM_20G:
+                wire_time /= ETH_SPEED_NUM_20G;
                 break;
-        case ETH_LINK_SPEED_10G:
-                wire_time /= ETH_LINK_SPEED_10G;
+        case ETH_SPEED_NUM_10G:
+                wire_time /= ETH_SPEED_NUM_10G;
                 break;
-        case ETH_LINK_SPEED_1000:
-                wire_time /= ETH_LINK_SPEED_1000;
+        case ETH_SPEED_NUM_1G:
+                wire_time /= ETH_SPEED_NUM_1G;
                 break;
         case 0:
@@ -2072 +2135 @@
                 if (mesg && libtrace_message_queue_count(mesg) > 0)
                         return READ_MESSAGE;
-                if (libtrace_halt)
-                        return READ_EOF;
+                if ((nb_rx=is_halted(libtrace)) != (size_t) -1)
+                        return nb_rx;
                 /* Wait a while, polling on memory degrades performance
                  * This relieves the pressure on memory allowing the NIC to DMA */
@@ -2091 +2154 @@
         int i;
         dpdk_per_stream_t *stream = t->format_data;
+        struct dpdk_addt_hdr * hdr;
 
         nb_rx = dpdk_read_packet_stream (libtrace, stream, &t->messages,
@@ -2107 +2171 @@
                         packets[i]->trace = libtrace;
                         packets[i]->error = 1;
+                        hdr = (struct dpdk_addt_hdr *) 
+                                        ((struct rte_mbuf*) pkts_burst[i] + 1);
+                        packets[i]->order = hdr->timestamp;
                         dpdk_prepare_packet(libtrace, packets[i], packets[i]->buffer, packets[i]->type, 0);
                 }
@@ -2194 +2261 @@
         stats->captured = dev_stats.ipackets;
 
-        /* Not that we support adding filters but if we did this
-         * would work */
-        stats->filtered += dev_stats.fdirmiss;
-
         stats->dropped_valid = true;
         stats->dropped = dev_stats.imissed;
 
+#if RTE_VERSION >= RTE_VERSION_NUM(16, 4, 0, 2)
+        /* DPDK commit 86057c fixes ensures missed does not get counted as
+         * errors */
+        stats->errors_valid = true;
+        stats->errors = dev_stats.ierrors;
+#else
         /* DPDK errors includes drops */
         stats->errors_valid = true;
         stats->errors = dev_stats.ierrors - dev_stats.imissed;
-
+#endif
         stats->received_valid = true;
         stats->received = dev_stats.ipackets + dev_stats.imissed;
@@ -2217 +2286 @@
                                             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 */
+        size_t nb_rx; /* Number of received packets we've read */
 
         do {
 
-                /* See if we already have a packet waiting */
-                nb_rx = rte_eth_rx_burst(FORMAT(trace)->port,
-                                         FORMAT_DATA_FIRST(trace)->queue_id,
-                                         pkts_burst, 1);
-
-                if (nb_rx > 0) {
+                /* No packets waiting in our buffer? Try and read some more */
+                if (FORMAT(trace)->burst_size == FORMAT(trace)->burst_offset) {
+                        nb_rx = rte_eth_rx_burst(FORMAT(trace)->port,
+                                                 FORMAT_DATA_FIRST(trace)->queue_id,
+                                                 FORMAT(trace)->burst_pkts, BURST_SIZE);
+                        if (nb_rx > 0) {
+                                dpdk_ready_pkts(trace, FORMAT_DATA_FIRST(trace),
+                                                FORMAT(trace)->burst_pkts, nb_rx);
+                                FORMAT(trace)->burst_size = nb_rx;
+                                FORMAT(trace)->burst_offset = 0;
+                        }
+                }
+
+                /* Now do we have packets waiting? */
+                if (FORMAT(trace)->burst_size != FORMAT(trace)->burst_offset) {
                         /* Free the last packet buffer */
                         if (packet->buffer != NULL) {
@@ -2239 +2316 @@
                         packet->type = TRACE_RT_DATA_DPDK;
                         event.type = TRACE_EVENT_PACKET;
-                        dpdk_ready_pkts(trace, FORMAT_DATA_FIRST(trace), pkts_burst, 1);
-                        packet->buffer = FORMAT(trace)->burst_pkts[0];
+                        packet->buffer = FORMAT(trace)->burst_pkts[
+                                             FORMAT(trace)->burst_offset++];
                         dpdk_prepare_packet(trace, packet, packet->buffer, packet->type, 0);
                         event.size = 1; // TODO should be bytes read, which essentially useless anyway
@@ -2269 +2346 @@
 
 static void dpdk_help(void) {
-        printf("dpdk format module: $Revision: 1752 $\n");
+        printf("dpdk format module: %s (%d) \n", rte_version(), RTE_VERSION);
         printf("Supported input URIs:\n");
         printf("\tdpdk:<domain:bus:devid.func>-<coreid>\n");

lib/format_linux_common.c

@@ -315 +315 @@
         libtrace_filter_t *filter = FORMAT_DATA->filter;
 
+        stream->last_timestamp = 0;
+
         /* Create a raw socket for reading packets on */
         stream->fd = socket(PF_PACKET, SOCK_RAW, htons(ETH_P_ALL));

lib/format_linux_common.h

@@ -257 +257 @@
          * that every ring can get setup the same */
         libtrace_list_t *per_stream;
+
 };
 
@@ -287 +288 @@
         /* The ring buffer layout */
         struct tpacket_req req;
+        uint64_t last_timestamp;
 } ALIGN_STRUCT(CACHE_LINE_SIZE);
 
-#define ZERO_LINUX_STREAM {-1, MAP_FAILED, 0, {0,0,0,0}}
+#define ZERO_LINUX_STREAM {-1, MAP_FAILED, 0, {0,0,0,0}, 0}
 
 

lib/format_linux_int.c

@@ -215 +215 @@
                                 return -1;
                         } else {
-                                if (libtrace_halt)
-                                        return READ_EOF;
+                                if ((ret=is_halted(libtrace)) != -1)
+                                        return ret;
                         }
                 }
@@ -286 +286 @@
         }
 
+
         /* Buffer contains all of our packet (including our custom header) so
          * we just need to get prepare_packet to set all our packet pointers
@@ -294 +295 @@
                 return -1;
         
+        if (hdr->timestamptype == TS_TIMEVAL) {
+                packet->order = (((uint64_t)hdr->tv.tv_sec) << 32)
+                            + ((((uint64_t)hdr->tv.tv_usec) << 32) /1000000);
+        } else if (hdr->timestamptype == TS_TIMESPEC) {
+                packet->order = (((uint64_t)hdr->ts.tv_sec) << 32)
+                            + ((((uint64_t)hdr->ts.tv_nsec) << 32) /1000000000);
+        } else {
+                packet->order = 0;
+        }
+
+        if (packet->order <= stream->last_timestamp) {
+                packet->order = stream->last_timestamp + 1;
+        }
+
+        stream->last_timestamp = packet->order;
+
         return hdr->wirelen+sizeof(*hdr);
 }

lib/format_linux_ring.c

@@ -447 +447 @@
 #ifdef HAVE_NETPACKET_PACKET_H
 #define LIBTRACE_MIN(a,b) ((a)<(b) ? (a) : (b))
+/* We use TP_STATUS_LIBTRACE to ensure we don't loop back on ourself
+ * and read the same packet twice if an old packet has not yet been freed */
+#define TP_STATUS_LIBTRACE 0xFFFFFFFF
+
 inline static int linuxring_read_stream(libtrace_t *libtrace,
                                         libtrace_packet_t *packet,
@@ -470 +474 @@
          * ready for consumption.
          */
-        while (!(header->tp_status & TP_STATUS_USER)) {
+        while (!(header->tp_status & TP_STATUS_USER) ||
+               header->tp_status == TP_STATUS_LIBTRACE) {
+                if ((ret=is_halted(libtrace)) != -1)
+                        return ret;
                 pollset[0].fd = stream->fd;
                 pollset[0].events = POLLIN;
@@ -509 +516 @@
                         }
                 } else {
-                        /* Poll timed out - check if we should exit */
-                        if (libtrace_halt)
-                                return 0;
+                        /* Poll timed out - check if we should exit on next loop */
                         continue;
                 }
         }
-
         packet->buffer = header;
         packet->trace = libtrace;
+        
+        header->tp_status = TP_STATUS_LIBTRACE;
 
         /* If a snaplen was configured, automatically truncate the packet to
@@ -531 +537 @@
         stream->rxring_offset++;
         stream->rxring_offset %= stream->req.tp_frame_nr;
+
+        packet->order = (((uint64_t)TO_TP_HDR2(packet->buffer)->tp_sec) << 32)
+                        + ((((uint64_t)TO_TP_HDR2(packet->buffer)->tp_nsec)
+                        << 32) / 1000000000);
+
+        if (packet->order <= stream->last_timestamp) {
+                packet->order = stream->last_timestamp + 1;
+        }
+
+        stream->last_timestamp = packet->order;
+                
 
         /* We just need to get prepare_packet to set all our packet pointers
@@ -574 +591 @@
         /* Fetch the current frame */
         header = GET_CURRENT_BUFFER(FORMAT_DATA_FIRST);
-        if (header->tp_status & TP_STATUS_USER) {
+        if (header->tp_status & TP_STATUS_USER &&
+            header->tp_status != TP_STATUS_LIBTRACE) {
                 /* We have a frame waiting */
                 event.size = trace_read_packet(libtrace, packet);

lib/format_pcap.c

@@ -447 +447 @@
                         case 1: break; /* no error */
                         case 0: 
-                                if (libtrace_halt)
-                                        return 0;
+                                if ((ret=is_halted(libtrace)) != -1)
+                                        return ret;
                                 continue; /* timeout expired */
                         case -1: 

lib/hash_toeplitz.c

@@ -71 +71 @@
  * Creates a random unidirectional RSS key - a ip or ip+port combination in
  * the opposite directions will most likely get different hashes.
+ * @param key An array of bytes to retrieve the RSS key
+ * @param num The number of bytes in key
+ */
+void toeplitz_ncreate_unikey(uint8_t *key, size_t num) {
+        size_t i;
+        unsigned int seed = time(NULL);
+        for (i = 0; i < num; i++) {
+                key[i] = (uint8_t) rand_r(&seed);
+        }
+}
+
+/**
+ * Creates a random 40 byte unidirectional RSS key - a ip or ip+port combination
+ * in the opposite directions will most likely get different hashes.
  * @param key must have 40 bytes of space to retrieve random the key
- */ 
+ */
 void toeplitz_create_unikey(uint8_t *key) {
-        int i;
-        unsigned int seed = time(NULL);
-        for (i = 0; i < 40; i++) {
-                key[i] = (uint8_t) rand_r(&seed);
-        }
+        toeplitz_ncreate_unikey(key, 40);
 }
 
@@ -85 +95 @@
  * in opposite directions will receive the same hash
  * @param key must have 40 bytes of space to retrieve random the key
- */
-void toeplitz_create_bikey(uint8_t *key) {
+ * @param num The number of bytes in the key, must be a multiple of 2
+ */
+void toeplitz_ncreate_bikey(uint8_t *key, size_t num) {
         unsigned int seed = time(NULL);
-        int i;
+        size_t i;
+        if (num % 2 != 0) {
+                perror("Can not create a bidirectional key for an odd length key");
+        }
         // Every thing is 16bit (port=16, ipv4=32, ipv6=128 
         // aligned so this will make the hash bidirectional
         uint16_t bi_key = (uint16_t) rand_r(&seed);
         uint16_t *bi_rep = (uint16_t *) key;
-        for (i = 0; i < 20; i++) {
+        for (i = 0; i < num/2; i++) {
                 bi_rep[i] = bi_key;
         }
+}
+
+/**
+ * Create a 40 byte bidirectional RSS key, i.e. ip and ip+port configurations
+ * in opposite directions will receive the same hash
+ * @param key An array of bytes to retrieve the RSS key
+ */
+void toeplitz_create_bikey(uint8_t *key) {
+        toeplitz_ncreate_bikey(key, 40);
 }
 
@@ -106 +129 @@
         }
         toeplitz_hash_expand_key(conf);
+        conf->hash_ipv4 = 1;
+        conf->hash_ipv6 = 1;
+        conf->hash_tcp_ipv4 = 1;
+        conf->x_hash_udp_ipv4 = 1;
+        conf->hash_tcp_ipv6 = 1;
+        conf->x_hash_udp_ipv6 = 1;
 }
 

lib/hash_toeplitz.h

@@ -58 +58 @@
 void toeplitz_init_config(toeplitz_conf_t *conf, bool bidirectional);
 uint64_t toeplitz_hash_packet(const libtrace_packet_t * pkt, const toeplitz_conf_t *cnf);
+void toeplitz_ncreate_bikey(uint8_t *key, size_t num);
 void toeplitz_create_bikey(uint8_t *key);
+void toeplitz_ncreate_unikey(uint8_t *key, size_t num);
 void toeplitz_create_unikey(uint8_t *key);
 

lib/libtrace.h.in

@@ -260 +260 @@
 /** If the packet has allocated its own memory the buffer_control should be
  * set to TRACE_CTRL_PACKET, so that the memory will be freed when the packet
- * is destroyed. If the packet has been zerocopied out of memory owned by
+ * is destroyed. If the packet has been zero-copied out of memory owned by
  * something else, e.g. a DAG card, it should be TRACE_CTRL_EXTERNAL.
  *
@@ -282 +282 @@
 /** Enumeration of error codes */
 enum {
-        /** No Error has occured.... yet. */
+        /** No Error has occurred.... yet. */
         TRACE_ERR_NOERROR       = 0,
         /** The URI passed to trace_create() is unsupported, or badly formed */
@@ -336 +336 @@
 #endif
         TRACE_DLT_PPP_SERIAL = 50,
-        TRACE_DLT_LINKTYPE_RAW = 101, /**< See TRACE_DLT_RAW for explainations of pain. */
+        TRACE_DLT_LINKTYPE_RAW = 101, /**< See TRACE_DLT_RAW for explanations of pain. */
         TRACE_DLT_C_HDLC = 104,
         TRACE_DLT_IEEE802_11 = 105,
@@ -548 +548 @@
         uint64_t hash; /**< A hash of the packet as supplied by the user */
         int error; /**< The error status of pread_packet */
-        uint64_t internalid;            /** Internal indentifier for the pkt */
+        uint64_t internalid;            /** Internal identifier for the pkt */
         void *srcbucket;
 } libtrace_packet_t;
@@ -576 +576 @@
     TRACE_RADIOTAP_LOCK_QUALITY = 7, /**< Barker Code lock quality (uint16) */
     TRACE_RADIOTAP_TX_ATTENUATION = 8, /**< TX attenuation as unitless distance from max power (uint16) */
-    TRACE_RADIOTAP_DB_TX_ATTENUATION = 9, /**< TX attenutation as dB from max power (uint16) */
+    TRACE_RADIOTAP_DB_TX_ATTENUATION = 9, /**< TX attenuation as dB from max power (uint16) */
     TRACE_RADIOTAP_DBM_TX_POWER = 10, /**< TX Power in dBm (int8) */
     TRACE_RADIOTAP_ANTENNA = 11, /**< Antenna frame was rx'd or tx'd on (uint8) */
@@ -869 +869 @@
 #define LIBTRACE_GRE_FLAG_VERMASK  0x0007
 
+
+/* PPTP GRE (RFC2637) */
+#define LIBTRACE_GRE_FLAG_ACK      0x0080
+#define LIBTRACE_GRE_PPTP_VERSION  0x0001
+
 /** Libtrace local definition of VXLAN Header
  * (draft-mahalingam-dutt-dcops-vxlan)
@@ -1182 +1187 @@
  * @param [out] format  A pointer that will be updated to point to an allocated 
  *                      string holding the format component of the URI
- * @return NULL if an error occured, otherwise return a pointer to the uridata 
+ * @return NULL if an error occurred, otherwise return a pointer to the uridata 
  * component
  *
@@ -1207 +1212 @@
  *  - rt:hostname:port
  *
- *  If an error occured when attempting to open the trace file, a
+ *  If an error occurred when attempting to open the trace file, a
  *  trace is still returned so trace_is_err() should be called to find out
- *  if an error occured. The trace is created in the configuration state, you 
+ *  if an error occurred. The trace is created in the configuration state, you 
  *  must call trace_start before attempting to read packets from the trace.
  */
@@ -1238 +1243 @@
  *  - pcap:/path/to/pcap/file
  *
- *  If an error occured when attempting to open the output trace, a trace is 
+ *  If an error occurred when attempting to open the output trace, a trace is 
  *  still returned but trace_errno will be set. Use trace_is_err_out() and 
  *  trace_perror_output() to get more information.
@@ -1377 +1382 @@
 } trace_option_output_t;
 
-/* To add a new stat field update this list, and the relevent places in
+/* To add a new stat field update this list, and the relevant places in
  * libtrace_stat_t structure.
  */
@@ -1430 +1435 @@
         uint64_t filtered;
 
-        /** The total number of good packets which have been recevied. Including
+        /** The total number of good packets which have been received. Including
          * those which are dropped and filtered. This does not include errors.
          *
@@ -1610 +1615 @@
  * Returns statistic counters for a trace, for a parallel trace this is a
  * combined total.
- * Where possible these are retrived atomically, however this behaviour depends
+ * Where possible these are retrieved atomically, however this behaviour depends
  * on the underlying trace format.
  *
@@ -1631 +1636 @@
 /**
  * Returns statistic counters for a single thread of a trace.
- * Where possible these are retrived atomically, however this behaviour depends
+ * Where possible these are retrieved atomically, however this behaviour depends
  * on the underlying trace format.
  *
  * @param trace The input trace to examine.
- * @param t An optional thread to received stats for or NULL to retrive stats
+ * @param t An optional thread to received stats for or NULL to retrieve stats
  *          for the current thread
  * @param stats Filled upon return with statistics about the trace, check the
@@ -1734 +1739 @@
  * function should be avoided where possible.
  * 
- * @par The reason you would want to use this function is that a zerocopied
+ * @par The reason you would want to use this function is that a zero-copied
  * packet from a device will be stored using memory owned by the device which
  * may be a limited resource. Copying the packet will ensure that the packet
@@ -2693 +2698 @@
  *
  * @note This function only works for OSPF version 2 packets.
- * @note trace_get_next_ospf_lsa_v2() should be subequently used to process the LSAs
+ * @note trace_get_next_ospf_lsa_v2() should be subsequently used to process the LSAs
  */
 DLLEXPORT SIMPLE_FUNCTION 
@@ -2716 +2721 @@
  *
  * @note This function only works for OSPF version 2 packets.
- * @note trace_get_next_ospf_lsa_header_v2() should be subequently used to process the LSA headers
+ * @note trace_get_next_ospf_lsa_header_v2() should be subsequently used to process the LSA headers
  */
 DLLEXPORT SIMPLE_FUNCTION 
@@ -2739 +2744 @@
  *
  * @note This function only works for OSPF version 2 packets.
- * @note trace_get_next_ospf_link_v2() should be subequently used to process
+ * @note trace_get_next_ospf_link_v2() should be subsequently used to process
  * the links
  */

lib/libtrace_int.h

@@ -327 +327 @@
         /** Synchronise writes/reads across this format object and attached threads etc */
         pthread_mutex_t libtrace_lock;
+        /** Packet read lock, seperate from libtrace_lock as to not block while reading a burst */
+        pthread_mutex_t read_packet_lock;
         /** State */
         enum trace_state state;
@@ -606 +608 @@
          * @return The size of the packet read (in bytes) including the capture
          * framing header, or -1 if an error occurs. 0 is returned in the
-         * event of an EOF. 
+         * event of an EOF or -2 in the case of interrupting the parallel API.
          *
          * If no packets are available for reading, this function should block
@@ -1032 +1034 @@
  */
 extern volatile int libtrace_halt;
+
+/**
+ * Used by a format to check if trace_interrupt or if a trace_pause/stop has
+ * been called. Provides backwards compatibility with traditional read
+ * functions when trace_read_packet() is used by the parallel API.
+ *
+ * Returns -1 if not halting otherwise returns the code that the read
+ * operation should pass on.
+ */
+static inline int is_halted(libtrace_t *trace) {
+        if (!(libtrace_halt || trace->state == STATE_PAUSING)) {
+                return -1;
+        } else if (libtrace_halt) {
+                return READ_EOF;
+        } else {
+                return READ_MESSAGE;
+        }
+}
 
 /** Registers a new capture format module.

lib/protocols_l2.c

@@ -313 +313 @@
                 switch(ntohs(ppp->protocol)) {
                         case 0x0021: *type = TRACE_ETHERTYPE_IP; break;
+                        case 0x0057: *type = TRACE_ETHERTYPE_IPV6; break;                               
                         /* If it isn't IP, then it is probably PPP control and
                          * I can't imagine anyone caring about that too much
@@ -395 +396 @@
                         case 0x0021: /* IP */
                                 *type = TRACE_ETHERTYPE_IP;
+                                break;
+                        case 0x0057: /* IPV6 */
+                                *type = TRACE_ETHERTYPE_IPV6;
                                 break;
                         case 0xc021: /* Link Control Protocol */

lib/protocols_transport.c

@@ -549 +549 @@
         uint32_t *remaining)
 {
+    uint8_t flags = ntohs(gre->flags);
     uint32_t size = 4; /* GRE is 4 bytes long by default */
     if (remaining && *remaining < size) {
@@ -555 +556 @@
     }
 
-    if ((ntohs(gre->flags) & LIBTRACE_GRE_FLAG_CHECKSUM) != 0) {
-        size += 4;  /* An extra 4 bytes. */
-    }
-
-    if ((ntohs(gre->flags) & LIBTRACE_GRE_FLAG_KEY) != 0) {
-        size += 4;  /* An extra 4 bytes. */
-    }
-
-    if ((ntohs(gre->flags) & LIBTRACE_GRE_FLAG_SEQ) != 0) {
-        size += 4;  /* An extra 4 bytes. */
+    if((flags & LIBTRACE_GRE_FLAG_VERMASK) == LIBTRACE_GRE_PPTP_VERSION) {
+        size += 4;
+
+        if ((flags & LIBTRACE_GRE_FLAG_SEQ) != 0) {
+            size += 4;
+        }
+        if ((flags & LIBTRACE_GRE_FLAG_ACK) != 0) {
+            size += 4;
+        }
+    } else {
+
+        if ((ntohs(gre->flags) & LIBTRACE_GRE_FLAG_CHECKSUM) != 0) {
+            size += 4;  /* An extra 4 bytes. */
+        }
+
+        if ((ntohs(gre->flags) & LIBTRACE_GRE_FLAG_KEY) != 0) {
+           size += 4;  /* An extra 4 bytes. */
+        }
+
+        if ((ntohs(gre->flags) & LIBTRACE_GRE_FLAG_SEQ) != 0) {
+            size += 4;  /* An extra 4 bytes. */
+        }
     }
 

lib/trace.c

@@ -261 +261 @@
         /* Parallel inits */
         ASSERT_RET(pthread_mutex_init(&libtrace->libtrace_lock, NULL), == 0);
+        ASSERT_RET(pthread_mutex_init(&libtrace->read_packet_lock, NULL), == 0);
         ASSERT_RET(pthread_cond_init(&libtrace->perpkt_cond, NULL), == 0);
         libtrace->state = STATE_NEW;
@@ -385 +386 @@
         /* Parallel inits */
         ASSERT_RET(pthread_mutex_init(&libtrace->libtrace_lock, NULL), == 0);
+        ASSERT_RET(pthread_mutex_init(&libtrace->read_packet_lock, NULL), == 0);
         ASSERT_RET(pthread_cond_init(&libtrace->perpkt_cond, NULL), == 0);
         libtrace->state = STATE_NEW; // TODO MAYBE DEAD
@@ -548 +550 @@
 
         /* Finish the last packet we read - for backwards compatibility */
-        if (libtrace->last_packet)
+        if (!libtrace_parallel && libtrace->last_packet)
                 trace_fin_packet(libtrace->last_packet);
         assert(libtrace->last_packet == NULL);
@@ -680 +682 @@
 
         ASSERT_RET(pthread_mutex_destroy(&libtrace->libtrace_lock), == 0);
+        ASSERT_RET(pthread_mutex_destroy(&libtrace->read_packet_lock), == 0);
         ASSERT_RET(pthread_cond_destroy(&libtrace->perpkt_cond), == 0);
 
@@ -694 +697 @@
 
         /* Finish any the last packet we read - for backwards compatibility */
-        if (libtrace->last_packet) {
+        if (!libtrace_parallel && libtrace->last_packet) {
                 trace_fin_packet(libtrace->last_packet);
         }
@@ -758 +761 @@
 
         ASSERT_RET(pthread_mutex_destroy(&libtrace->libtrace_lock), == 0);
+        ASSERT_RET(pthread_mutex_destroy(&libtrace->read_packet_lock), == 0);
         ASSERT_RET(pthread_cond_destroy(&libtrace->perpkt_cond), == 0);
 
@@ -833 +837 @@
                 packet->trace->format->fin_packet(packet);
         }
-        if (packet->trace && packet->trace->last_packet == packet)
+        if (!libtrace_parallel && packet->trace &&
+             packet->trace->last_packet == packet) {
                 packet->trace->last_packet = NULL;
+        }
         
         if (packet->buf_control == TRACE_CTRL_PACKET && packet->buffer) {
@@ -864 +870 @@
 
                 if (packet->trace) {
-                        pthread_mutex_lock(&packet->trace->libtrace_lock);
-                        if (packet->trace->last_packet == packet)
-                                packet->trace->last_packet = NULL;
-                        pthread_mutex_unlock(&packet->trace->libtrace_lock);
+                        if (!libtrace_parallel && packet->trace->last_packet == packet)
+                                packet->trace->last_packet = NULL;
                 }
 
@@ -921 +925 @@
                         size_t ret;
                         int filtret;
+                        if ((ret=is_halted(libtrace)) != (size_t)-1)
+                                return ret;
                         /* Store the trace we are reading from into the packet opaque 
                          * structure */
                         packet->trace = libtrace;
                         ret=libtrace->format->read_packet(libtrace,packet);
-                        if (ret==(size_t)-1 || ret==0) {
+                        if (ret==(size_t)READ_MESSAGE ||
+                            ret==(size_t)-1 || ret==0) {
                                 packet->trace = NULL;
                                 return ret;
@@ -953 +960 @@
                         ++libtrace->accepted_packets;
                         ++libtrace->sequence_number;
-                        if (packet->trace == libtrace)
+                        if (!libtrace_parallel && packet->trace == libtrace)
                                 libtrace->last_packet = packet;
 
@@ -996 +1003 @@
 
         packet->trace = trace;
-        pthread_mutex_lock(&trace->libtrace_lock);
-        trace->last_packet = packet;
-        pthread_mutex_unlock(&trace->libtrace_lock);
+        if (!libtrace_parallel)
+                trace->last_packet = packet;
         /* Clear packet cache */
         trace_clear_cache(packet);

lib/trace_parallel.c

@@ -677 +677 @@
                         if (!trace->pread) {
                                 assert(packets[0]);
+                                ASSERT_RET(pthread_mutex_lock(&trace->libtrace_lock), == 0);
                                 nb_packets = trace_read_packet(trace, packets[0]);
+                                ASSERT_RET(pthread_mutex_unlock(&trace->libtrace_lock), == 0);
                                 packets[0]->error = nb_packets;
                                 if (nb_packets > 0)
@@ -787 +789 @@
                         libtrace_ocache_alloc(&trace->packet_freelist, (void **) &packet, 1, 1);
                 assert(packet);
-
-                if (libtrace_halt) {
-                        packet->error = 0;
-                        break;
-                }
 
                 // Check for messages that we expect MESSAGE_DO_PAUSE, (internal messages only)
@@ -821 +818 @@
 
                 if ((packet->error = trace_read_packet(trace, packet)) <1) {
-                        break; /* We are EOF or error'd either way we stop  */
+                        if (packet->error == READ_MESSAGE) {
+                                pkt_skipped = 1;
+                                continue;
+                        } else {
+                                break; /* We are EOF or error'd either way we stop  */
+                        }
                 }
 
@@ -889 +891 @@
         //bool tick_hit = false;
 
-        ASSERT_RET(pthread_mutex_lock(&libtrace->libtrace_lock), == 0);
+        ASSERT_RET(pthread_mutex_lock(&libtrace->read_packet_lock), == 0);
         /* Read nb_packets */
         for (i = 0; i < nb_packets; ++i) {
-                if (libtrace_halt) {
-                        break;
+                if (libtrace_message_queue_count(&t->messages) > 0) {
+                        if ( i==0 ) {
+                                ASSERT_RET(pthread_mutex_unlock(&libtrace->read_packet_lock), == 0);
+                                return READ_MESSAGE;
+                        } else {
+                                break;
+                        }
                 }
                 packets[i]->error = trace_read_packet(libtrace, packets[i]);
@@ -900 +907 @@
                         /* We'll catch this next time if we have already got packets */
                         if ( i==0 ) {
-                                ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
+                                ASSERT_RET(pthread_mutex_unlock(&libtrace->read_packet_lock), == 0);
                                 return packets[i]->error;
                         } else {
@@ -916 +923 @@
                 store_first_packet(libtrace, packets[0], t);
         }
-        ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
+        ASSERT_RET(pthread_mutex_unlock(&libtrace->read_packet_lock), == 0);
         /* XXX TODO this needs to be inband with packets, or we don't bother in this case
         if (tick_hit) {
@@ -1007 +1014 @@
                         /* Check if we are newer than the previous 'first' packet */
                         size_t first = libtrace->first_packets.first;
-                        if (trace_get_seconds(dup) <
-                                trace_get_seconds(libtrace->first_packets.packets[first].packet))
+                        struct timeval cur_ts = trace_get_timeval(dup);
+                        struct timeval first_ts = trace_get_timeval(libtrace->first_packets.packets[first].packet);
+                        if (timercmp(&cur_ts, &first_ts, <))
                                 libtrace->first_packets.first = t->perpkt_num;
                 }
@@ -1336 +1344 @@
                                         trace_set_capture_length(packets[i],
                                                         libtrace->snaplen);
-                                trace_packet_set_order(packets[i], trace_get_erf_timestamp(packets[i]));
                         }
                 } while(ret == 0);
@@ -1471 +1478 @@
                 libtrace->config.reporter_thold = 100;
         if (libtrace->config.burst_size <= 0)
-                libtrace->config.burst_size = 10;
+                libtrace->config.burst_size = 32;
         if (libtrace->config.thread_cache_size <= 0)
-                libtrace->config.thread_cache_size = 20;
+                libtrace->config.thread_cache_size = 64;
         if (libtrace->config.cache_size <= 0)
                 libtrace->config.cache_size = (libtrace->config.hasher_queue_size + 1) * libtrace->perpkt_thread_count;
@@ -1707 +1714 @@
         libtrace->perpkt_threads = NULL;
         /* Set a global which says we are using a parallel trace. This is
-         * for backwards compatability due to changes when destroying packets */
+         * for backwards compatibility due to changes when destroying packets */
         libtrace_parallel = 1;
 
@@ -1728 +1735 @@
                 if (libtrace->perpkt_thread_count > 1)
                         libtrace->pread = trace_pread_packet_first_in_first_served;
+                        /* Don't wait for a burst of packets if the format is
+                         * live as this could block ring based formats and
+                         * introduces delay. */
+                        if (libtrace->format->info.live) {
+                                libtrace->config.burst_size = 1;
+                        }
                 else
                         /* Use standard read_packet */

test/test-live.c

@@ -171 +171 @@
         if (!stat->received_valid) {
                 printf("\tInfo: trace does not support received counter\n");
-        } else if (stat->received != 100) {
-                ERROR("Trace received %zu/100 packets\n", stat->received);
+        } else if (stat->received != (uint32_t) test_size) {
+                ERROR("Trace received %zu/%u packets\n", stat->received,
+                                (uint32_t)test_size);
         }
 

test/test-tracetime-parallel.c

@@ -232 +232 @@
 
         trace_set_reporter_thold(trace, 1);
+        trace_set_burst_size(trace, 10);
 
         // Start it

tools/tracertstats/tracertstats.c

@@ -258 +258 @@
         }
         trace_set_combiner(trace, &combiner_ordered, (libtrace_generic_t){0});
-        trace_set_tracetime(trace, true);
         trace_set_perpkt_threads(trace, threadcount);
         trace_set_burst_size(trace, burstsize);
@@ -264 +263 @@
         if (trace_get_information(trace)->live) {
                 trace_set_tick_interval(trace, (int) (packet_interval * 1000));
+        } else {
+                trace_set_tracetime(trace, true);
         }