Changeset 29bbef0

Timestamp:

03/30/14 17:48:26 (8 years ago)

Author:

Richard Sanger <rsangerarj@…>

Branches:

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

Children:

f1015ad

Parents:

dad224b

Message:

My work from over summer, with a few things tidied up and updated to include recent commits/patches to bring this up to date. Still very much work in progress.

Files:

• lib/Makefile.am (modified) (3 diffs)
• lib/deque.c (added)
• lib/deque.h (added)
• lib/format_atmhdr.c (modified) (1 diff)
• lib/format_bpf.c (modified) (1 diff)
• lib/format_dag24.c (modified) (1 diff)
• lib/format_dag25.c (modified) (1 diff)
• lib/format_dpdk.c (modified) (21 diffs)
• lib/format_duck.c (modified) (1 diff)
• lib/format_erf.c (modified) (2 diffs)
• lib/format_legacy.c (modified) (1 diff)
• lib/format_linux.c (modified) (27 diffs)
• lib/format_pcap.c (modified) (2 diffs)
• lib/format_pcapfile.c (modified) (1 diff)
• lib/format_rt.c (modified) (1 diff)
• lib/format_tsh.c (modified) (2 diffs)
• lib/hash_toeplitz.c (added)
• lib/hash_toeplitz.h (added)
• lib/libtrace.h.in (modified) (3 diffs)
• lib/libtrace_int.h (modified) (5 diffs)
• lib/libtrace_message_queue.c (added)
• lib/libtrace_message_queue.h (added)
• lib/trace.c (modified) (17 diffs)
• lib/trace_parallel.c (added)
• lib/trace_ringbuffer.c (added)
• lib/trace_ringbuffer.h (added)
• lib/trace_sliding_window.c (added)
• lib/trace_sliding_window.h (added)
• lib/trace_vector.c (added)
• lib/trace_vector.h (added)
• tools/traceanon/Makefile.am (modified) (3 diffs)
• tools/traceanon/panon.c (modified) (2 diffs)
• 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)

lib/Makefile.am

@@ -2 +2 @@
 include_HEADERS = libtrace.h dagformat.h lt_inttypes.h daglegacy.h rt_protocol.h erftypes.h  
 
-AM_CFLAGS=@LIBCFLAGS@
-AM_CXXFLAGS=@LIBCXXFLAGS@
+AM_CFLAGS=@LIBCFLAGS@ -pthread
+AM_CXXFLAGS=@LIBCXXFLAGS@ -pthread
 
 extra_DIST = format_template.c
@@ -40 +40 @@
 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 \
@@ -52 +52 @@
                 $(DAGSOURCE) format_erf.h \
                 $(BPFJITSOURCE) \
-                libtrace_arphrd.h
+                libtrace_arphrd.h \
+                trace_ringbuffer.c trace_vector.c libtrace_message_queue.c deque.c trace_sliding_window.c hash_toeplitz.c
 
 if DAG2_4

lib/format_atmhdr.c

@@ -231 +231 @@
         trace_event_trace,              /* trace_event */
         NULL,                           /* help */
+        NULL, /* pstart_input */
+        NULL, /* pread_packet */
+        NULL, /* ppause_input */
+        NULL, /* pfin_input */
+        NULL, /* pconfig_input */
         NULL                            /* next pointer */
 };

lib/format_bpf.c

@@ -648 +648 @@
         NULL,                   /* trace_event */
         bpf_help,               /* help */
+        NULL, /* pstart_input */
+        NULL, /* pread_packet */
+        NULL, /* ppause_input */
+        NULL, /* pfin_input */
+        NULL, /* pconfig_input */
         NULL
 };

lib/format_dag24.c

@@ -557 +557 @@
         trace_event_dag,                /* trace_event */
         dag_help,                       /* help */
+        NULL, /* pstart_input */
+        NULL, /* pread_packet */
+        NULL, /* ppause_input */
+        NULL, /* pfin_input */
+        NULL, /* pconfig_input */
         NULL                            /* next pointer */
 };

lib/format_dag25.c

@@ -1210 +1210 @@
         trace_event_dag,                /* trace_event */
         dag_help,                       /* help */
+        NULL, /* pstart_input */
+        NULL, /* pread_packet */
+        NULL, /* ppause_input */
+        NULL, /* pfin_input */
+        NULL, /* pconfig_input */
         NULL                            /* next pointer */
 };

lib/format_dpdk.c

@@ -46 +46 @@
 #include "format_helper.h"
 #include "libtrace_arphrd.h"
+#include "hash_toeplitz.h"
 
 #ifdef HAVE_INTTYPES_H
@@ -72 +73 @@
 #include <rte_mempool.h>
 #include <rte_mbuf.h>
+#include <rte_launch.h>
+#include <rte_lcore.h>
+#include <rte_per_lcore.h>
 
 /* The default size of memory buffers to use - This is the max size of standard 
@@ -129 +133 @@
 
 /* Print verbose messages to stdout */
-#define DEBUG 0
+#define DEBUG 1
 
 /* Use clock_gettime() for nanosecond resolution rather than gettimeofday() 
@@ -176 +180 @@
     DPDK_RUNNING,
     DPDK_PAUSED,
+};
+
+struct per_lcore_t
+{
+        // TODO move time stamp stuff here
+        uint16_t queue_id;
+        uint8_t port;
+        uint8_t enabled;
 };
 
@@ -194 +206 @@
     char mempool_name[MEMPOOL_NAME_LEN]; /* The name of the mempool that we are using */
     unsigned int nb_blacklist; /* Number of blacklist items in are valid */
+    uint8_t rss_key[40]; // This is the RSS KEY
 #if HAS_HW_TIMESTAMPS_82580
     /* Timestamping only relevent to RX */
@@ -200 +213 @@
     uint32_t wrap_count; /* Number of times the NIC clock has wrapped around completely */
 #endif
+        // DPDK normally seems to have a limit of 
+        struct per_lcore_t per_lcore[RTE_MAX_LCORE];
 };
 
@@ -399 +414 @@
      * Basically binds this thread to a fixed core, which we choose as
      * the last core on the machine (assuming fewer interrupts mapped here).
-     * "-c" controls the cpu mask 0x1=1st core 0x2=2nd 0x4=3rd and so om
+     * "-c" controls the cpu mask 0x1=1st core 0x2=2nd 0x4=3rd and so on
      * "-n" the number of memory channels into the CPU (hardware specific)
      *      - Most likely to be half the number of ram slots in your machine.
@@ -436 +451 @@
     }
 
-    /* Make our mask */
-    snprintf(cpu_number, sizeof(cpu_number), "%x", 0x1 << (my_cpu - 1));
+    /* Make our mask */ //  0x1 << (my_cpu - 1)
+    snprintf(cpu_number, sizeof(cpu_number), "%x", 0x3);
     argv[2] = cpu_number;
 
@@ -478 +493 @@
         return -1;
     }
+    
+    struct rte_eth_dev_info dev_info;
+    rte_eth_dev_info_get(0, &dev_info);
+    printf("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;
@@ -485 +505 @@
     char err[500];
     err[0] = 0;
+    int i;
     
     libtrace->format_data = (struct dpdk_format_data_t *)
@@ -504 +525 @@
     FORMAT(libtrace)->wrap_count = 0;
 #endif
-
+        for (i = 0;i < RTE_MAX_LCORE; i++) {
+                // Disabled by default
+                FORMAT(libtrace)->per_lcore[i].enabled = 0;
+        }
+        
     if (dpdk_init_enviroment(libtrace->uridata, FORMAT(libtrace), err, sizeof(err)) != 0) {
         trace_set_err(libtrace, TRACE_ERR_INIT_FAILED, "%s", err);
@@ -547 +572 @@
 };
 
+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;
+                        }
+        }
+        return -1;
+}
 /**
  * Note here snaplen excludes the MAC checksum. Packets over
@@ -596 +643 @@
 static struct rte_eth_conf port_conf = {
         .rxmode = {
+                .mq_mode = ETH_RSS,
                 .split_hdr_size = 0,
                 .header_split   = 0, /**< Header Split disabled */
@@ -619 +667 @@
         .txmode = {
                 .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,
+                },
         },
 };
@@ -710 +764 @@
      */
     
+    
+    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 */
@@ -770 +827 @@
     return 0;
 }
+int mapper_start(void *data); // This actually a void*
+
+/* Attach memory to the port and start the port or restart the ports.
+ */
+static int dpdk_start_port_queues (libtrace_t *libtrace, 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 enviroment 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 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
+    printf("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,
+                       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,
+                       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", 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
+    printf("Doing dev configure\n");
+#endif
+    /* Initilise 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
+    printf("Doing queue configure\n");
+#endif  
+                /* Initilise the RX queue with some packets from memory */
+                ret = rte_eth_rx_queue_setup(format_data->port, i,
+                                                                format_data->nb_rx_buf, SOCKET_ID_ANY,
+                                                                &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;
+                }
+        }
+    
+#if DEBUG
+    printf("Doing start device\n");
+#endif  
+    /* Start device */
+    ret = rte_eth_dev_start(format_data->port);
+#if DEBUG
+    printf("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(mapper_start, (void *)libtrace, i);
+        }*/
+    
+    /* Wait for the link to come up */
+    rte_eth_link_get(format_data->port, &link_info);
+#if DEBUG
+    printf("Link status is %d %d %d\n", (int) link_info.link_status,
+            (int) link_info.link_duplex, (int) link_info.link_speed);
+#endif
+
+    return 0;
+}
 
 static int dpdk_start_input (libtrace_t *libtrace) {
@@ -782 +991 @@
     }
     return 0;
+}
+
+static int dpdk_pstart_input (libtrace_t *libtrace) {
+    char err[500];
+    int enabled_lcore_count = 0, i=0;
+    int tot = libtrace->mapper_thread_count;
+    err[0] = 0;
+        
+        libtrace->mapper_thread_count;
+        
+        for (i = 0; i < RTE_MAX_LCORE; i++)
+        {
+                if (rte_lcore_is_enabled(i))
+                        enabled_lcore_count++;
+        }
+        
+        tot = MIN(libtrace->mapper_thread_count, enabled_lcore_count);
+        tot = MIN(tot, 8);
+        printf("Running pstart DPDK %d %d %d %d\n", tot, libtrace->mapper_thread_count, enabled_lcore_count, rte_lcore_count());
+        
+    if (dpdk_start_port_queues(libtrace, 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;
+    }
+    
+    return 0;
+    return tot;
 }
 
@@ -1116 +1354 @@
 }
 
+
+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 */
@@ -1145 +1392 @@
             return dpdk_ready_pkt(libtrace, packet, pkts_burst[0]);
         }
+    }
+    
+    /* We'll never get here - but if we did it would be bad */
+    return -1;
+}
+libtrace_thread_t * get_thread_table(libtrace_t *libtrace);
+static int dpdk_pread_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) */
+
+    /* 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;
+    
+    /* Wait for a packet */
+    while (1) {
+        /* Poll for a single packet */
+        nb_rx = rte_eth_rx_burst(FORMAT(libtrace)->port,
+                            get_thread_table_num(libtrace), pkts_burst, 1);
+        if (nb_rx > 0) { /* Got a packet - otherwise we keep spining */
+                        printf("Doing P READ PACKET port=%d q=%d\n", (int) FORMAT(libtrace)->port, (int) get_thread_table_num(libtrace));
+            return dpdk_ready_pkt(libtrace, packet, pkts_burst[0]);
+        }
+        // Check the message queue this could be (Well it shouldn't but anyway) be less than 0
+        if (libtrace_message_queue_count(&(get_thread_table(libtrace)->messages)) > 0) {
+                        printf("Extra message yay");
+                        return -2;
+                }
     }
     
@@ -1325 +1613 @@
         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 */
@@ -1348 +1636 @@
         dpdk_trace_event,               /* trace_event */
     dpdk_help,              /* help */
+    dpdk_pstart_input, /* pstart_input */
+        dpdk_pread_packet, /* pread_packet */
+        dpdk_pause_input, /* ppause */
+        dpdk_fin_input, /* p_fin */
+        dpdk_pconfig_input, /* pconfig_input */
         NULL
 };

lib/format_duck.c

@@ -360 +360 @@
         NULL,                           /* trace_event */
         duck_help,                      /* help */
+        NULL, /* pstart_input */
+        NULL, /* pread_packet */
+        NULL, /* ppause_input */
+        NULL, /* pfin_input */
+        NULL, /* pconfig_input */
         NULL                            /* next pointer */
 };

lib/format_erf.c

@@ -828 +828 @@
         erf_event,                      /* trace_event */
         erf_help,                       /* help */
+        NULL, /* pstart_input */
+        NULL, /* pread_packet */
+        NULL, /* ppause_input */
+        NULL, /* pfin_input */
+        NULL, /* pconfig_input */
         NULL                            /* next pointer */
 };
@@ -871 +876 @@
         erf_event,                      /* trace_event */
         erf_help,                       /* help */
+        NULL, /* pstart_input */
+        NULL, /* pread_packet */
+        NULL, /* ppause_input */
+        NULL, /* pfin_input */
+        NULL, /* pconfig_input */
         NULL                            /* next pointer */
 };

lib/format_legacy.c

@@ -681 +681 @@
         trace_event_trace,              /* trace_event */
         legacynzix_help,                /* help */
+        NULL, /* pstart_input */
+        NULL, /* pread_packet */
+        NULL, /* ppause_input */
+        NULL, /* pfin_input */
+        NULL, /* pconfig_input */
         NULL,                           /* next pointer */
 };

lib/format_linux.c

@@ -72 +72 @@
 #include <sys/mman.h>
 
+#include <fcntl.h>
+
 /* MAX_ORDER is defined in linux/mmzone.h. 10 is default for 2.4 kernel.
  * max_order will be decreased by one if the ring buffer fails to allocate.
@@ -147 +149 @@
 #define PACKET_HDRLEN   11
 #define PACKET_TX_RING  13
+#define PACKET_FANOUT   18
 #define TP_STATUS_USER  0x1
 #define TP_STATUS_SEND_REQUEST  0x1
@@ -154 +157 @@
 #define TPACKET_ALIGN(x)        (((x)+TPACKET_ALIGNMENT-1)&~(TPACKET_ALIGNMENT-1))
 #define TPACKET_HDRLEN         (TPACKET_ALIGN(sizeof(struct tpacket2_hdr)) + sizeof(struct sockaddr_ll))
+
+/* Since 3.1 kernel we have packet_fanout support */
+// schedule to socket by skb's rxhash - the implementation is bi-directional
+#define PACKET_FANOUT_HASH              0
+// schedule round robin
+#define PACKET_FANOUT_LB                1
+// schedule to the same socket that received the packet
+#define PACKET_FANOUT_CPU               2
+// Something to do with fragmented packets and hashing problems !! TODO figure out if this needs to be on
+#define PACKET_FANOUT_FLAG_DEFRAG       0x8000
+/* Included but unused by libtrace since 3.10 */
+// if one socket if full roll over to the next
+#define PACKET_FANOUT_ROLLOVER          3
+// This flag makes any other system roll over
+#define PACKET_FANOUT_FLAG_ROLLOVER     0x1000
+/* Included but unused by libtrace since 3.12 */
+// schedule random
+#define PACKET_FANOUT_RND               4
+
 
 enum tpacket_versions {
@@ -184 +206 @@
         unsigned int tp_frame_size;  /* Size of frame */
         unsigned int tp_frame_nr;    /* Total number of frames */
+};
+
+struct linux_per_thread_t {
+        char *rx_ring;
+        int rxring_offset;
+        int fd;
+        // The flag layout should be the same for all (I Hope)
+        // max_order
 };
 
@@ -212 +242 @@
         /* Used to determine buffer size for the ring buffer */ 
         uint32_t max_order;
+        /* Used for the parallel case, fanout is the mode */
+        uint16_t fanout_flags;
+        /* The group lets Linux know which sockets to group together
+         * so we use a random here to try avoid collisions */
+        uint16_t fanout_group;
+        /* When running in parallel mode this is malloc'd with an array 
+         * file descriptors from packet fanout will use, here we assume/hope
+         * that every ring can get setup the same */
+        struct linux_per_thread_t *per_thread;
 };
 
@@ -367 +406 @@
         FORMAT(libtrace->format_data)->rxring_offset = 0;
         FORMAT(libtrace->format_data)->max_order = MAX_ORDER;
+        FORMAT(libtrace->format_data)->fanout_flags = PACKET_FANOUT_LB; // This might be best or alternatively PACKET_FANOUT_LB
+        // Some examples use pid for the group however that would limit a single
+        // application to use only int/ring format, instead using rand
+        FORMAT(libtrace->format_data)->fanout_group = (uint16_t) rand();
+        FORMAT(libtrace->format_data)->per_thread = NULL;
 }
 static int linuxring_init_input(libtrace_t *libtrace) 
@@ -582 +626 @@
         return 0;
 }
-static int linuxring_start_input(libtrace_t *libtrace){
-
-        char error[2048];       
+
+/**
+ * Converts a socket, either packet_mmap or standard raw socket into a 
+ * fanout socket.
+ * NOTE: This means we can read from the socket with multiple queues, 
+ * each must be setup (identically) and then this called upon them
+ * 
+ * @return 0 success, -1 error
+ */
+static inline int socket_to_packet_fanout(int fd, 
+                                        uint16_t fanout_flags,
+                                        uint16_t fanout_group) {
+        int fanout_opt = ((int)fanout_flags << 16) | (int)fanout_group;
+        if (setsockopt(fd, SOL_PACKET, PACKET_FANOUT,
+                        &fanout_opt, sizeof(fanout_opt)) == -1) {
+                return -1;
+        }
+        return 0;
+}
+
+static int linuxnative_ppause_input(libtrace_t *libtrace)
+{
+        int i;
+        int tot = libtrace->mapper_thread_count;
+        printf("CAlling native pause packet\n");
+        
+        for (i = 0; i < tot; i++) {
+                close(FORMAT(libtrace->format_data)->per_thread[i].fd);
+        }
+        
+        free(FORMAT(libtrace->format_data)->per_thread);
+        FORMAT(libtrace->format_data)->per_thread = NULL;
+        return 0;
+}
+
+static int linuxring_start_input(libtrace_t *libtrace)
+{
+        char error[2048];
 
         /* We set the socket up the same and then convert it to PACKET_MMAP */
@@ -609 +688 @@
 }
 
+static int linuxnative_pstart_input(libtrace_t *libtrace) {
+        int i = 0;
+        int tot = libtrace->mapper_thread_count;
+        int iserror = 0;
+        // We store this here otherwise it will be leaked if the memory doesn't know
+        struct linux_per_thread_t *per_thread = NULL;
+        
+        if (!FORMAT(libtrace->format_data)->per_thread) {
+                per_thread = calloc(tot, sizeof(struct linux_per_thread_t));
+                FORMAT(libtrace->format_data)->per_thread = per_thread;
+        } else {
+                // Whats going on this might not work 100%
+                // We assume all sockets have been closed ;)
+                printf("Pause and then start called again lets hope that mapper_thread_count hasn't changed\n");
+        }
+        
+        printf("Calling native pstart packet\n");
+        for (i = 0; i < tot; ++i)
+        {
+                if (FORMAT(libtrace->format_data)->format == TRACE_FORMAT_LINUX_NATIVE) {
+                        if (linuxnative_start_input(libtrace) != 0) {
+                                iserror = 1;
+                                break;
+                        }
+                } else {
+                        // This must be ring
+                        if (linuxring_start_input(libtrace) != 0) {
+                                iserror = 1;
+                                break;
+                        }
+                }
+                if (socket_to_packet_fanout(FORMAT(libtrace->format_data)->fd, FORMAT(libtrace->format_data)->fanout_flags, FORMAT(libtrace->format_data)->fanout_group) != 0)
+                {
+                        iserror = 1;
+                        // Clean up here to keep consistent with every one else
+                        trace_set_err(libtrace, TRACE_ERR_INIT_FAILED, "Converting the fd to a socket fanout failed");
+                        close(FORMAT(libtrace->format_data)->fd);
+                        free(libtrace->format_data);
+                        libtrace->format_data = NULL;
+                        break;
+                }
+                per_thread[i].fd = FORMAT(libtrace->format_data)->fd;
+                if (FORMAT(libtrace->format_data)->format == TRACE_FORMAT_LINUX_RING) {
+                        per_thread[i].rxring_offset = FORMAT(libtrace->format_data)->rxring_offset;
+                        per_thread[i].rx_ring = FORMAT(libtrace->format_data)->rx_ring;
+                }
+        }
+        
+        // Roll back those that failed - by this point in time the format_data
+        // has been freed
+        if (iserror) {
+                for (i = i - 1; i >= 0; i--) {
+                        close(per_thread[i].fd);
+                }
+                free(per_thread);
+                per_thread = NULL;
+                return -1;
+        }
+        
+        return 0;
+}
+
 static int linuxnative_start_output(libtrace_out_t *libtrace)
 {
@@ -615 +756 @@
                 free(DATAOUT(libtrace));
                 return -1;
-        }       
+        }
 
         return 0;
@@ -660 +801 @@
         return 0;
 }
+
 static int linuxring_pause_input(libtrace_t *libtrace)
 {
@@ -800 +942 @@
 #endif /* HAVE_NETPACKET_PACKET_H */
 
+
+static int linuxnative_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:
+                                        // Do fanout
+                                        FORMAT(libtrace->format_data)->fanout_flags = PACKET_FANOUT_LB;
+                                        // Or we could balance to the CPU
+                                        return 0;
+                                case HASHER_BIDIRECTIONAL:
+                                case HASHER_UNIDIRECTIONAL:
+                                        FORMAT(libtrace->format_data)->fanout_flags = PACKET_FANOUT_HASH;
+                                        return 0;
+                                case HASHER_CUSTOM:
+                                case HASHER_HARDWARE:
+                                        return -1;
+                        }
+                        break;
+                /* Avoid default: so that future options will cause a warning
+                 * here to remind us to implement it, or flag it as
+                 * unimplementable
+                 */
+        }
+        
+        /* Don't set an error - trace_config will try to deal with the
+         * option and will set an error if it fails */
+        return -1;
+}
+
+
 static int linuxnative_prepare_packet(libtrace_t *libtrace UNUSED, 
                 libtrace_packet_t *packet, void *buffer, 
@@ -871 +1047 @@
 
 #ifdef HAVE_NETPACKET_PACKET_H
-static int linuxnative_read_packet(libtrace_t *libtrace, libtrace_packet_t *packet) 
+libtrace_thread_t * get_thread_table(libtrace_t *libtrace) ;
+inline static int linuxnative_read_packet_fd(libtrace_t *libtrace, libtrace_packet_t *packet, int fd, const int check_queue) 
 {
         struct libtrace_linuxnative_header *hdr;
@@ -879 +1056 @@
         struct cmsghdr *cmsg;
         int snaplen;
+
         uint32_t flags = 0;
         
@@ -914 +1092 @@
         iovec.iov_base = (void*)(packet->buffer+sizeof(*hdr));
         iovec.iov_len = snaplen;
-
-        hdr->wirelen = recvmsg(FORMAT(libtrace->format_data)->fd, &msghdr, 0);
-
+        
+        if (check_queue) {
+                // Check for a packet - TODO only Linux has MSG_DONTWAIT should use fctl O_NONBLOCK
+                hdr->wirelen = recvmsg(fd, &msghdr, MSG_DONTWAIT);
+                if ((unsigned) hdr->wirelen == -1 && (errno == EAGAIN || errno == EWOULDBLOCK)) {
+                        // Do message queue check or select
+                        int ret;
+                        fd_set rfds;
+                        FD_ZERO(&rfds);
+                        FD_SET(fd, &rfds);
+                        FD_SET(get_thread_table(libtrace)->messages.pipefd[0], &rfds);
+                        int largestfd = fd > get_thread_table(libtrace)->messages.pipefd[0] ? fd : get_thread_table(libtrace)->messages.pipefd[0];
+                        
+                        do {
+                                ret = select(largestfd+1, &rfds, NULL, NULL, NULL);
+                                if (ret == -1 && errno != EINTR)
+                                        perror("Select() failed");
+                        }
+                        while (ret == -1);
+                        
+                        assert (ret == 1 || ret == 2); // No timeout 0 is not an option
+                        
+                        if (FD_ISSET(get_thread_table(libtrace)->messages.pipefd[0], &rfds)) {
+                                // Not an error but check the message queue we have something
+                                return -2;
+                        }
+                        // Otherwise we must have a packet
+                        hdr->wirelen = recvmsg(fd, &msghdr, 0);
+                }
+        } else {
+                hdr->wirelen = recvmsg(fd, &msghdr, 0);
+        }
+        
         if (hdr->wirelen==~0U) {
                 trace_set_err(libtrace,errno,"recvmsg");
@@ -964 +1172 @@
         if (cmsg == NULL) {
                 struct timeval tv;
-                if (ioctl(FORMAT(libtrace->format_data)->fd, 
-                                  SIOCGSTAMP,&tv)==0) {
+                if (ioctl(fd, SIOCGSTAMP,&tv)==0) {
                         hdr->tv.tv_sec = tv.tv_sec;
                         hdr->tv.tv_usec = tv.tv_usec;
@@ -986 +1193 @@
 }
 
+static int linuxnative_read_packet(libtrace_t *libtrace, libtrace_packet_t *packet) 
+{
+        int fd = FORMAT(libtrace->format_data)->fd;
+        return linuxnative_read_packet_fd(libtrace, packet, fd, 0);
+}
+
+static int linuxnative_pread_packet(libtrace_t *libtrace, libtrace_packet_t *packet)
+{
+        int fd = FORMAT(libtrace->format_data)->per_thread[get_thread_table_num(libtrace)].fd;
+        printf("Thread number is #%d point %p\n", get_thread_table_num(libtrace), FORMAT(libtrace->format_data)->per_thread);
+        return linuxnative_read_packet_fd(libtrace, packet, fd, 1);
+}
+
 #define LIBTRACE_BETWEEN(test,a,b) ((test) >= (a) && (test) < (b))
 static int linuxring_get_capture_length(const libtrace_packet_t *packet);
@@ -992 +1212 @@
 /* Release a frame back to the kernel or free() if it's a malloc'd buffer 
  */
-inline static void ring_release_frame(libtrace_t *libtrace, libtrace_packet_t *packet ){
+inline static void ring_release_frame(libtrace_t *libtrace, libtrace_packet_t *packet){
         /* Free the old packet */
         if(packet->buffer == NULL)
@@ -1004 +1224 @@
                 struct linux_format_data_t *ftd = FORMAT(libtrace->format_data);
                 
-                /* Check it's within our buffer first */
-                if(LIBTRACE_BETWEEN((char *) packet->buffer, 
+                /* Check it's within our buffer first - consider the pause resume case it might have already been free'd lets hope we get another buffer */
+                // For now let any one free anything
+                /*if(LIBTRACE_BETWEEN((char *) packet->buffer, 
                                 (char *) ftd->rx_ring,
                                 ftd->rx_ring
-                                + ftd->req.tp_block_size * ftd->req.tp_block_nr)){
+                                + ftd->req.tp_block_size * ftd->req.tp_block_nr)){*/
                         TO_TP_HDR(packet->buffer)->tp_status = 0;
                         packet->buffer = NULL;
-                }
-        }
-}
-
-static int linuxring_read_packet(libtrace_t *libtrace, libtrace_packet_t *packet) {
+                /*}*/
+        }
+}
+
+/**
+ * Free any resources being kept for this packet, Note: libtrace
+ * will ensure all fields are zeroed correctly.
+ */
+static void linuxring_fin_packet(libtrace_packet_t *packet)
+{
+        assert(packet->trace);
+        
+        // Started should always match the existence of the rx_ring
+        assert(!!FORMAT(packet->trace->format_data)->rx_ring == !!packet->trace->started);
+        
+        // Our packets are always under our control
+        assert(packet->buf_control == TRACE_CTRL_EXTERNAL);
+        
+        if (FORMAT(packet->trace->format_data)->rx_ring) // If we don't have a ring its already been destroyed or paused
+                ring_release_frame(packet->trace, packet);
+        else
+                packet->buffer = NULL;
+}
+
+inline static int linuxring_read_packet_fd(libtrace_t *libtrace, libtrace_packet_t *packet, int fd, int *rxring_offset, char *rx_ring, int message) {
 
         struct tpacket2_hdr *header;
-        struct pollfd pollset;  
         int ret;
         
@@ -1027 +1267 @@
         
         /* Fetch the current frame */
-        header = GET_CURRENT_BUFFER(libtrace);
+        header = ((void*) rx_ring) + *rxring_offset * FORMAT(libtrace->format_data)->req.tp_frame_size; // GET_CURRENT_BUFFER(libtrace);
         assert((((unsigned long) header) & (pagesize - 1)) == 0);
 
@@ -1035 +1275 @@
          */
         while (!(header->tp_status & TP_STATUS_USER)) {
-                pollset.fd = FORMAT(libtrace->format_data)->fd;
-                pollset.events = POLLIN;
-                pollset.revents = 0;
-                /* Wait for more data */
-                ret = poll(&pollset, 1, -1);
-                if (ret < 0) {
-                        if (errno != EINTR)
-                                trace_set_err(libtrace,errno,"poll()");
-                        return -1;
+                if (message) {
+                        struct pollfd pollset[2];
+                        pollset[0].fd = fd;
+                        pollset[0].events = POLLIN;
+                        pollset[0].revents = 0;
+                        pollset[1].fd = libtrace_message_queue_get_fd(&get_thread_table(libtrace)->messages);
+                        pollset[1].events = POLLIN;
+                        pollset[1].revents = 0;
+                        /* Wait for more data or a message*/
+                        ret = poll(pollset, 2, -1);
+                        if (ret < 0) {
+                                if (errno != EINTR)
+                                        trace_set_err(libtrace,errno,"poll()");
+                                return -1;
+                        }
+                        // Check for a message otherwise loop
+                        if (pollset[1].revents)
+                                return -2;
+                } else {
+                        struct pollfd pollset;
+                        pollset.fd = fd;
+                        pollset.events = POLLIN;
+                        pollset.revents = 0;
+
+                        /* Wait for more data or a message*/
+                        ret = poll(&pollset, 1, -1);
+                        if (ret < 0) {
+                                if (errno != EINTR)
+                                        trace_set_err(libtrace,errno,"poll()");
+                                return -1;
+                        }
                 }
         }
@@ -1050 +1312 @@
 
         /* Move to next buffer */
-        FORMAT(libtrace->format_data)->rxring_offset++;
-        FORMAT(libtrace->format_data)->rxring_offset %= FORMAT(libtrace->format_data)->req.tp_frame_nr;
+        (*rxring_offset)++;
+        *rxring_offset %= FORMAT(libtrace->format_data)->req.tp_frame_nr;
 
         /* We just need to get prepare_packet to set all our packet pointers
@@ -1061 +1323 @@
                                 linuxring_get_capture_length(packet);
 
+}
+
+static int linuxring_read_packet(libtrace_t *libtrace, libtrace_packet_t *packet) {
+        int fd = FORMAT(libtrace->format_data)->fd;
+        int *rxring_offset = &FORMAT(libtrace->format_data)->rxring_offset;
+        char *rx_ring = FORMAT(libtrace->format_data)->rx_ring;
+        return linuxring_read_packet_fd(libtrace, packet, fd, rxring_offset, rx_ring, 0);
+}
+
+static int linuxring_pread_packet(libtrace_t *libtrace, libtrace_packet_t *packet) {
+        int tnum = get_thread_table_num(libtrace);
+        int fd = FORMAT(libtrace->format_data)->per_thread[tnum].fd;
+        int *rxring_offset = &FORMAT(libtrace->format_data)->per_thread[tnum].rxring_offset;
+        char *rx_ring = FORMAT(libtrace->format_data)->per_thread[tnum].rx_ring;
+        printf("Thread number is #%d point %p\n", get_thread_table_num(libtrace), FORMAT(libtrace->format_data)->per_thread);
+        return linuxring_read_packet_fd(libtrace, packet, fd, rxring_offset, rx_ring, 1);
 }
 
@@ -1550 +1828 @@
         trace_event_device,             /* trace_event */
         linuxnative_help,               /* help */
+        linuxnative_pstart_input,                       /* pstart_input */
+        linuxnative_pread_packet,                       /* pread_packet */
+        linuxnative_ppause_input,                       /* ppause */
+        linuxnative_fin_input,                          /* p_fin */
+        linuxnative_pconfig_input,                      /* pconfig input */ //int (*pconfig_input)(libtrace_t *libtrace,trace_option_t option,void *value);
         NULL
 };
@@ -1570 +1853 @@
         linuxring_read_packet,  /* read_packet */
         linuxring_prepare_packet,       /* prepare_packet */
-        NULL,                           /* fin_packet */
+        linuxring_fin_packet,                           /* fin_packet */
         linuxring_write_packet, /* write_packet */
         linuxring_get_link_type,        /* get_link_type */
@@ -1593 +1876 @@
         linuxring_event,                /* trace_event */
         linuxring_help,         /* help */
+        linuxnative_pstart_input,                       /* pstart_input */
+        linuxring_pread_packet,                 /* pread_packet */
+        linuxnative_ppause_input,                       /* ppause */
+        linuxnative_fin_input,                          /* p_fin */
+        linuxnative_pconfig_input,
         NULL
 };
@@ -1645 +1933 @@
         trace_event_device,             /* trace_event */
         linuxnative_help,               /* help */
+        NULL,                   /* pstart_input */
+        NULL,                   /* pread_packet */
         NULL
 };
@@ -1688 +1978 @@
         NULL,                           /* trace_event */
         linuxring_help,                 /* help */
+        NULL,                   /* pstart_input */
+        NULL,                   /* pread_packet */
         NULL
 };

lib/format_pcap.c

@@ -801 +801 @@
         trace_event_trace,              /* trace_event */
         pcap_help,                      /* help */
+        NULL, /* pstart_input */
+        NULL, /* pread_packet */
+        NULL, /* ppause_input */
+        NULL, /* pfin_input */
+        NULL, /* pconfig_input */
         NULL                            /* next pointer */
 };
@@ -844 +849 @@
         trace_event_device,             /* trace_event */
         pcapint_help,                   /* help */
+        NULL, /* pstart_input */
+        NULL, /* pread_packet */
+        NULL, /* ppause_input */
+        NULL, /* pfin_input */
+        NULL, /* pconfig_input */
         NULL                            /* next pointer */
 };

lib/format_pcapfile.c

@@ -774 +774 @@
         pcapfile_event,         /* trace_event */
         pcapfile_help,                  /* help */
+        NULL, /* pstart_input */
+        NULL, /* pread_packet */
+        NULL, /* ppause_input */
+        NULL, /* pfin_input */
+        NULL, /* pconfig_input */
         NULL                            /* next pointer */
 };

lib/format_rt.c

@@ -858 +858 @@
         trace_event_rt,             /* trace_event */
         rt_help,                        /* help */
+        NULL, /* pstart_input */
+        NULL, /* pread_packet */
+        NULL, /* ppause_input */
+        NULL, /* pfin_input */
+        NULL, /* pconfig_input */
         NULL                            /* next pointer */
 };

lib/format_tsh.c

@@ -269 +269 @@
         trace_event_trace,              /* trace_event */
         tsh_help,                       /* help */
+        NULL, /* pstart_input */
+        NULL, /* pread_packet */
+        NULL, /* ppause_input */
+        NULL, /* pfin_input */
+        NULL, /* pconfig_input */
         NULL                            /* next pointer */
 };
@@ -317 +322 @@
         trace_event_trace,              /* trace_event */
         tsh_help,                       /* help */
+        NULL, /* pstart_input */
+        NULL, /* pread_packet */
+        NULL, /* ppause_input */
+        NULL, /* pfin_input */
+        NULL, /* pconfig_input */
         NULL                            /* next pointer */
 };

lib/libtrace.h.in

@@ -196 +196 @@
 typedef struct libtrace_filter_t libtrace_filter_t;
 
+/** Opaque structure holding information about a result */
+typedef struct libtrace_result_t libtrace_result_t;
+
+typedef struct libtrace_message_t libtrace_message_t;
+
+typedef struct libtrace_thread_t libtrace_thread_t;
+
 /** 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
@@ -481 +488 @@
         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;
 
@@ -3098 +3108 @@
 /*@}*/
 
+typedef void* (*fn_per_pkt)(libtrace_t* trace, libtrace_packet_t *p, libtrace_message_t *m, libtrace_thread_t *thread);
+typedef void* (*fn_reducer)(libtrace_t* trace, void* global_blob);
+typedef uint64_t (*fn_hasher)(libtrace_packet_t* packet, void *data);
+
+DLLEXPORT int trace_pstart(libtrace_t *libtrace, void* global_blob, fn_per_pkt per_pkt, fn_reducer reducer);
+DLLEXPORT int trace_pread_packet(libtrace_t *libtrace, libtrace_packet_t **packet);
+DLLEXPORT int trace_ppause(libtrace_t *libtrace);
+DLLEXPORT int trace_pstop(libtrace_t *libtrace);
+DLLEXPORT void trace_join(libtrace_t * trace);
+DLLEXPORT inline void print_contention_stats (libtrace_t *libtrace);
+DLLEXPORT libtrace_result_t *trace_create_result();
+
+typedef struct libtrace_result_t {
+        uint64_t key;
+        void * value;
+} libtrace_result_t;
+
+DLLEXPORT inline void libtrace_result_set_key(libtrace_result_t * result, uint64_t key);
+DLLEXPORT inline uint64_t libtrace_result_get_key(libtrace_result_t * result);
+DLLEXPORT inline void libtrace_result_set_value(libtrace_result_t * result, void * value);
+DLLEXPORT inline void* libtrace_result_get_value(libtrace_result_t * result);
+DLLEXPORT inline void libtrace_result_set_key_value(libtrace_result_t * result, uint64_t key, void * 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, uint64_t key, void * value);
+typedef struct libtrace_vector libtrace_vector_t;
+DLLEXPORT int trace_get_results(libtrace_t *libtrace, libtrace_vector_t * results);
+
+DLLEXPORT libtrace_result_t *trace_create_result();
+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_reducer(libtrace_t * libtrace, 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);
+
+
+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,
+        
+        /**
+         * See diagrams, this sets the maximum size of freelist used to 
+         * maintain packets and there 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.
+         */
+        TRACE_OPTION_SET_PACKET_FREELIST_SIZE,
+        
+        /**
+         * 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.
+         */
+        TRACE_OPTION_SET_MAPPER_BUFFER_SIZE,
+        
+        /**
+         * Libtrace set mapper thread count
+         */
+        TRACE_OPTION_SET_MAPPER_THREAD_COUNT,
+        
+        /**
+         * Libtrace should expect sequential keys from the output to count
+         * up starting numbered from 1, 2, 3 ... 
+         * such as is the case with numbered packets.
+         * 
+         * ALSO consider - TRACE_OPTIONS_ORDERED_RESULTS suitable for live formats
+         */
+         TRACE_OPTION_SEQUENTIAL,
+         
+         /**
+          * Libtrace ordered results, results in each queue are ordered by key
+          * however my not be sequential, a typically case is packet timestamps
+          * the reducer will receive packets in order - note threasholds
+          * will be used such that a empty queue wont break things
+          */
+         TRACE_OPTION_ORDERED,
+         
+         
+         /**
+          * When accepting ordered results if a threashold is meet before an
+          * older result is available from another queue drop that packet
+          */
+          TRACE_DROP_OUT_OF_ORDER,
+          
+          /**
+           * If set to true (i.e. 1) the trace starts a dedicated hasher thread
+           */
+          TRACE_OPTION_USE_DEDICATED_HASHER,
+          TRACE_OPTION_USE_SLIDING_WINDOW_BUFFER,
+          TRACE_OPTION_TRACETIME
+} trace_parallel_option_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);
+
 #ifdef __cplusplus
 } /* extern "C" */

lib/libtrace_int.h

@@ -165 +165 @@
         bool waiting;
 };
+
+enum thread_types {
+        THREAD_EMPTY,
+        THREAD_HASHER,
+        THREAD_MAPPER,
+        THREAD_REDUCER
+};
+
+enum thread_states {
+        THREAD_RUNNING,
+        THREAD_FINISHING,
+        THREAD_FINISHED,
+        THREAD_PAUSED
+};
+
+enum hasher_types {
+        HASHER_BALANCE, /** Balance load across CPUs best as possible this is basically to say don't care about hash, but this might still might be implemented using a hash or round robin etc.. */
+        HASHER_BIDIRECTIONAL, /** Use a hash which is uni-directional for TCP flows (IP src dest,TCP port src dest), non TCP
+                                                        Will be sent to the same place, with the exception of UDP which may or may not be sent to separate cores */
+        HASHER_UNIDIRECTIONAL, /** Use a hash which is uni-directional across TCP flow */
+        HASHER_CUSTOM, /** Always use the user supplied hasher */
+        HASHER_HARDWARE, /** Set by the format if the hashing is going to be done in hardware */
+};
+
+// Reduce expects sequential data
+#define REDUCE_SEQUENTIAL 0x1
+// Reduce is working on ordered data
+#define REDUCE_ORDERED 0x2
+// Reduce should sort the data
+#define REDUCE_SORT 0x4
+// Drop out of order valid with
+#define REDUCE_DROP_OOO 0x8
+// Reduce reads all queues with same key
+#define REDUCE_STEPPING 0x10
+
+#define MAPPER_USE_SLIDING_WINDOW 0x20
+
+
+#include "trace_ringbuffer.h"
+#include "trace_vector.h"
+#include "libtrace_message_queue.h"
+#include "deque.h"
+#include "trace_sliding_window.h"
+
+/**
+ * Information of this thread
+ */
+typedef struct libtrace_thread_t {
+        libtrace_t * trace;
+        void* ret;
+        enum thread_types type;
+        enum thread_states state;
+        void* user_data; // TLS for the user to use
+        pthread_t tid;
+        int map_num; // A number from 0-X that represents this mapper number
+                                // in the table, intended to quickly identify this thread
+                                // -1 represents NA (such as in the case this is not a mapper thread)
+        libtrace_ringbuffer_t rbuffer; // Input
+        libtrace_vector_t vector; // Output
+        libtrace_queue_t deque; // Real Output type makes more sense
+        libtrace_message_queue_t messages; // Message handling
+        // Temp storage for time sensitive results
+        uint64_t tmp_key;
+        void *tmp_data;
+        pthread_spinlock_t tmp_spinlock;
+        // 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;
+} libtrace_thread_t;
+
+/**
+ * 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 == mappers we have all
+        size_t first; // Valid if count != 0
+        struct __packet_storage_magic_type {
+                libtrace_packet_t * packet;
+                struct timeval tv;
+        } * packets;
+};
+
 
 /** A libtrace input trace 
@@ -187 +274 @@
         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;
+        
+        /** Use to control pausing threads and finishing threads etc always used with libtrace_lock */
+        pthread_cond_t perpkt_cond;
+        /** Set to the number of mapper threads that are finishing (or have finished), or to -1 once all have been joined, 0 implies all are running */
+        int mappers_finishing;
+        /** A count of mappers that are pausing */
+        int perpkt_pausing;
+        
+        /** For the sliding window hasher implementation */
+        pthread_rwlock_t window_lock;
+        /** Set once trace_join has been called */
+        bool joined;
+        /** Set to indicate a mappers queue is full and such the writing mapper cannot proceed */
+        bool mapper_queue_full;
+        /** Global storage for this trace, shared among all the threads  */
+        void* global_blob;
+        /** Requested size of the pkt buffer (currently only used if using dedicated hasher thread) */
+        int packet_freelist_size;
+        /** The actual freelist */
+        libtrace_ringbuffer_t packet_freelist;
+        /** The number of packets that can queue per mapper thread - XXX consider deadlocks with non malloc()'d packets that need to be released */
+        int mapper_buffer_size;
+        /** The reducer flags */
+        int reducer_flags;
+        /** Used to track the next expected key */
+        uint64_t expected_key;
+        /** User defined per_pkt function called when a pkt is ready */
+        fn_per_pkt per_pkt;
+        /** User defined reducer function entry point XXX not hooked up */
+        fn_reducer reducer;
+        /** 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;
+        
+        libtrace_thread_t hasher_thread;
+        libtrace_thread_t reducer_thread;
+        int mapper_thread_count;
+        libtrace_thread_t * mapper_threads; // All our mapper threads
+        libtrace_slidingwindow_t sliding_window;
+        sem_t sem;
+        // Used to keep track of the first packet seen on each thread
+        struct first_packets first_packets;
+        int tracetime;
 };
+
+inline void libtrace_zero_thread(libtrace_thread_t * t);
+inline void store_first_packet(libtrace_t *libtrace, libtrace_packet_t *packet, libtrace_thread_t *t);
 
 /** A libtrace output trace
@@ -201 +338 @@
 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;              
@@ -209 +346 @@
         libtrace_err_t err;
         /** Boolean flag indicating whether the trace has been started */
-        bool started;                   
+        bool started;
 };
 
@@ -733 +870 @@
         /** Prints some useful help information to standard output. */
         void (*help)(void);
-
+        
+        /** 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 If successful the number of threads started, 0 indicates
+         *                 no threads started and this should be done automatically.
+         *                 Otherwise in event of an error -1 is returned.
+         * 
+         */
+        int (*pstart_input)(libtrace_t *trace);
+        
+        /** Read a packet in the new parallel mode 
+         * @return same as read_packet, with the addition of return -2 to represent
+         * interrupted due to message waiting. */
+        int (*pread_packet)(libtrace_t *trace, libtrace_packet_t *packet);
+        
+        /** Pause a parallel trace */
+        int (*ppause_input)(libtrace_t *trace);
+        
+        /** Called after all threads have been paused, Finish (close) a parallel trace */
+        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);
+        
         /** Next pointer, should always be NULL - used by the format module
          * manager. */

lib/trace.c

@@ -98 +98 @@
 #include "format_helper.h"
 #include "rt_protocol.h"
+
+#include <pthread.h>
+#include <signal.h>
 
 #define MAXOPTS 1024
@@ -253 +256 @@
         libtrace->filtered_packets = 0;
         libtrace->accepted_packets = 0;
+        
+        /* Parallel inits */
+        // libtrace->libtrace_lock
+        // libtrace->perpkt_cond;
+        libtrace->perpkt_pausing = 0;
+        libtrace->mapper_queue_full = false;
+        libtrace->mappers_finishing = -1;
+        libtrace->reducer_flags = 0;
+        libtrace->joined = false;
+        libtrace->global_blob = NULL;
+        libtrace->per_pkt = NULL;
+        libtrace->reducer = NULL;
+        libtrace->hasher = NULL;
+        libtrace->packet_freelist_size = 0;
+        libtrace->mapper_buffer_size = 0;
+        libtrace->expected_key = 0;
+        libtrace_zero_ringbuffer(&libtrace->packet_freelist);
+        libtrace_zero_thread(&libtrace->hasher_thread);
+        libtrace_zero_thread(&libtrace->reducer_thread);
+        libtrace_zero_slidingwindow(&libtrace->sliding_window);
+        libtrace->reducer_thread.type = THREAD_EMPTY;
+        libtrace->mapper_thread_count = 0;
+        libtrace->mapper_threads = NULL;
 
         /* Parse the URI to determine what sort of trace we are dealing with */
@@ -348 +374 @@
         libtrace->io = NULL;
         libtrace->filtered_packets = 0;
+        
+        /* Parallel inits */
+        // libtrace->libtrace_lock
+        // libtrace->perpkt_cond;
+        libtrace->perpkt_pausing = 0;
+        libtrace->mapper_queue_full = false;
+        libtrace->mappers_finishing = -1;
+        libtrace->reducer_flags = 0;
+        libtrace->joined = false;
+        libtrace->global_blob = NULL;
+        libtrace->per_pkt = NULL;
+        libtrace->reducer = NULL;
+        libtrace->hasher = NULL;
+        libtrace->expected_key = 0;
+        libtrace->packet_freelist_size = 0;
+        libtrace->mapper_buffer_size = 0;
+        libtrace_zero_ringbuffer(&libtrace->packet_freelist);
+        libtrace_zero_thread(&libtrace->hasher_thread);
+        libtrace_zero_thread(&libtrace->reducer_thread);
+        libtrace_zero_slidingwindow(&libtrace->sliding_window);
+        libtrace->reducer_thread.type = THREAD_EMPTY;
+        libtrace->mapper_thread_count = 0;
+        libtrace->mapper_threads = NULL;
         
         for(tmp=formats_list;tmp;tmp=tmp->next) {
@@ -583 +632 @@
  */
 DLLEXPORT void trace_destroy(libtrace_t *libtrace) {
+    int i;
         assert(libtrace);
         if (libtrace->format) {
@@ -590 +640 @@
                         libtrace->format->fin_input(libtrace);
         }
-        /* Need to free things! */
-        if (libtrace->uridata)
+        /* Need to free things! */
+        if (libtrace->uridata)
                 free(libtrace->uridata);
+        
+        /* Empty any packet memory */
+
+        libtrace_packet_t * packet;
+        while (libtrace_ringbuffer_try_read(&libtrace->packet_freelist,(void **) &packet))
+                trace_destroy_packet(packet);
+        
+        libtrace_ringbuffer_destroy(&libtrace->packet_freelist);
+        
+        for (i = 0; i < libtrace->mapper_thread_count; ++i) {
+                        assert (libtrace_vector_get_size(&libtrace->mapper_threads[i].vector) == 0);
+                        libtrace_vector_destroy(&libtrace->mapper_threads[i].vector);
+        }
+        free(libtrace->mapper_threads);
+        libtrace->mapper_threads = NULL;
+        libtrace->mapper_thread_count = 0;
+        
         if (libtrace->event.packet) {
                 /* Don't use trace_destroy_packet here - there is almost
@@ -661 +728 @@
         dest->type=packet->type;
         dest->buf_control=TRACE_CTRL_PACKET;
+        dest->order = packet->order;
         /* Reset the cache - better to recalculate than try to convert
          * the values over to the new packet */
@@ -675 +743 @@
  */
 DLLEXPORT void trace_destroy_packet(libtrace_packet_t *packet) {
+        /* Free any resources possibly associated with the packet */
+        if (packet->trace && packet->trace->format->fin_packet) {
+                packet->trace->format->fin_packet(packet);
+        }
+        
         if (packet->buf_control == TRACE_CTRL_PACKET && packet->buffer) {
                 free(packet->buffer);
@@ -685 +758 @@
 }       
 
+/**
+ * Removes any possible data stored againt the trace and releases any data.
+ * This will not destroy a reusable good malloc'd buffer (TRACE_CTRL_PACKET)
+ * use trace_destroy_packet() for those diabolical purposes.
+ */
+void trace_fin_packet(libtrace_packet_t *packet);
+void trace_fin_packet(libtrace_packet_t *packet) {
+        if (packet)
+        {
+                if (packet->trace && packet->trace->format->fin_packet) {
+                        packet->trace->format->fin_packet(packet);
+                        //gettimeofday(&tv, NULL);
+                        //printf ("%d.%06d DESTROYED #%"PRIu64"\n", tv.tv_sec, tv.tv_usec, trace_packet_get(packet));
+                }
+
+                // No matter what we remove the header and link pointers
+                packet->trace = NULL;
+                packet->header = NULL;
+                packet->payload = NULL;
+
+                if (packet->buf_control != TRACE_CTRL_PACKET)
+                {
+                        //packet->buf_control = 0; // Invalid value this should be fixed
+                        packet->buffer = NULL;
+                }
+
+                packet->trace = NULL;
+                packet->hash = 0;
+                packet->order = 0;
+                trace_clear_cache(packet);
+        }
+}
+
 /* Read one packet from the trace into buffer. Note that this function will
  * block until a packet is read (or EOF is reached).
@@ -707 +813 @@
         }
         assert(packet);
-      
-        /* Store the trace we are reading from into the packet opaque 
-         * structure */
-        packet->trace = libtrace;
-
-        /* Finalise the packet, freeing any resources the format module
-         * may have allocated it
-         */
-        if (libtrace->format->fin_packet) {
-                libtrace->format->fin_packet(packet);
-        }
-
 
         if (libtrace->format->read_packet) {
                 do {
                         size_t ret;
-                        /* Clear the packet cache */
-                        trace_clear_cache(packet);
+                        /* Finalise the packet, freeing any resources the format module
+                         * may have allocated it and zeroing all data associated with it.
+                         */
+                        trace_fin_packet(packet);
+                        /* 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) {
@@ -743 +842 @@
                                                 libtrace->snaplen);
                         }
+                        trace_packet_set_order(packet, libtrace->accepted_packets);
                         ++libtrace->accepted_packets;
                         return ret;
@@ -946 +1046 @@
         }
 
-        return tv;
+    return tv;
 }
 
@@ -1199 +1299 @@
                 libtrace_linktype_t linktype    ) {
 #ifdef HAVE_BPF_FILTER
+        /* It just so happens that the underlying libs used by pthread arn't
+         * thread safe, namely lex/flex thingys, so single threaded compile
+         * multi threaded running should be safe.
+         */
+        static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
         assert(filter);
 
@@ -1220 +1325 @@
                                         "Unknown pcap equivalent linktype");
                         return -1;
+                }
+                assert (pthread_mutex_lock(&mutex) == 0);
+                /* Make sure not one bet us to this */
+                if (filter->flag) {
+                        printf("Someone bet us to compile the filter\n");
+                        assert (pthread_mutex_unlock(&mutex) == 0);
+                        return 1;
                 }
                 pcap=(pcap_t *)pcap_open_dead(
@@ -1233 +1345 @@
                                         pcap_geterr(pcap));
                         pcap_close(pcap);
+                        assert (pthread_mutex_unlock(&mutex) == 0);
                         return -1;
                 }
                 pcap_close(pcap);
                 filter->flag=1;
+                assert (pthread_mutex_unlock(&mutex) == 0);
         }
         return 0;
@@ -1256 +1370 @@
         libtrace_linktype_t linktype;
         libtrace_packet_t *packet_copy = (libtrace_packet_t*)packet;
+#ifdef HAVE_LLVM
+        static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER;
+#endif
 
         assert(filter);
@@ -1306 +1423 @@
          * what the link type was
          */
+        // Note internal mutex locking used here
         if (trace_bpf_compile(filter,packet_copy,linkptr,linktype)==-1) {
                 if (free_packet_needed) {
@@ -1316 +1434 @@
 #if HAVE_LLVM
         if (!filter->jitfilter) {
-                filter->jitfilter = compile_program(filter->filter.bf_insns, filter->filter.bf_len);
+                assert(pthread_mutex_lock(&mutex) == 0);
+                /* Again double check here like the bpf filter */
+                if(filter->jitfilter) 
+                        printf("Someone bet us to compile the JIT thingy\n");
+                else
+                /* Looking at compile_program source this appears to be thread safe 
+                 * however if this gets called twice we will leak this memory :(
+                 * as such lock here anyways */
+                        filter->jitfilter = compile_program(filter->filter.bf_insns, filter->filter.bf_len);
+                assert(pthread_mutex_unlock(&mutex) == 0);
         }
 #endif

tools/traceanon/Makefile.am

@@ -1 @@
-bin_PROGRAMS = traceanon
+bin_PROGRAMS = traceanon traceanon_parallel
 
 man_MANS = traceanon.1
@@ -6 +6 @@
 include ../Makefile.tools
 traceanon_SOURCES = traceanon.c rijndael.h rijndael.c panon.h panon.c ipenc.c ipenc.h 
+traceanon_parallel_SOURCES = traceanon_parallel.c rijndael.h rijndael.c panon.h panon.c ipenc.c ipenc.h 
 
 # rijndael.c does lots of nasty casting that is going to be a nightmare to fix
@@ -11 +12 @@
 # messy and hopefully isn't actually an issue. 
 traceanon_CFLAGS = $(AM_CFLAGS)
+traceanon_parallel_CFLAGS = $(AM_CFLAGS)

tools/traceanon/panon.c

@@ -8 +8 @@
 #include "panon.h"
 
-static uint8_t m_key[16];
-static uint8_t m_pad[16];
+static __thread uint8_t m_key[16];
+static __thread  uint8_t m_pad[16];
 
 #define CACHEBITS 20
@@ -16 +16 @@
 //static uint32_t enc_cache[CACHESIZE];
 
-static uint32_t *enc_cache = 0;
-static uint32_t fullcache[2][2];
+static __thread  uint32_t *enc_cache = 0; // Should be ok shared across multiple
+static __thread  uint32_t fullcache[2][2]; // Needs to be against on thread
 
 

tools/tracertstats/Makefile.am

@@ -1 @@
-bin_PROGRAMS = tracertstats
+bin_PROGRAMS = tracertstats tracertstats_parallel
 man_MANS = tracertstats.1
 EXTRA_DIST = $(man_MANS)
@@ -16 +16 @@
 tracertstats_SOURCES = tracertstats.c output.h output.c $(OUTPUT_MODULES)
 tracertstats_LDADD = -ltrace $(OUTPUT_PNG_LD)
+tracertstats_parallel_SOURCES = tracertstats_parallel.c output.h output.c $(OUTPUT_MODULES)
+tracertstats_parallel_LDADD = -ltrace $(OUTPUT_PNG_LD)

tools/tracestats/Makefile.am

@@ -1 @@
-bin_PROGRAMS = tracestats
+bin_PROGRAMS = tracestats tracestats_parallel
 bin_SCRIPTS = tracesummary
 
@@ -7 +7 @@
 include ../Makefile.tools
 tracestats_SOURCES = tracestats.c
+tracestats_parallel_SOURCES = tracestats_parallel.c