Changeset 1b59edf for lib

Timestamp:

02/11/15 11:13:27 (6 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:

0b01fea

Parents:

fed9152 (diff), 12ae766 (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 'tidy_parallel_libtrace' into develop

Location:

lib

Files:

• format_dpdk.c (modified) (14 diffs)
• libtrace_int.h (modified) (3 diffs)
• trace.c (modified) (3 diffs)
• trace_parallel.c (modified) (19 diffs)

lib/format_dpdk.c

@@ -201 +201 @@
  */
 
-/* Print verbose messages to stdout */
-#define DEBUG 1
+/* Print verbose messages to stderr */
+#define DEBUG 0
 
 /* Use clock_gettime() for nanosecond resolution rather than gettimeofday()
@@ -351 +351 @@
                 if (format_data->nb_blacklist >= sizeof (format_data->blacklist)
                                 / sizeof (format_data->blacklist[0])) {
-                        printf("Warning: too many devices to blacklist consider"
+                        fprintf(stderr, "Warning: too many devices to blacklist consider"
                                         " increasing BLACK_LIST_SIZE");
                         break;
@@ -625 +625 @@
         if (my_cpu < 0) {
                 /* If we can assign to a core on the same numa node */
-                printf("Using pci card on numa_node%d\n", format_data->nic_numa_node);
+                fprintf(stderr, "Using pci card on numa_node%d\n", format_data->nic_numa_node);
                 if(format_data->nic_numa_node >= 0) {
                         int max_node_cpu = -1;
@@ -735 +735 @@
     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",
+    fprintf(stderr, "Device port=0\n\tmin_rx_bufsize=%d\n\tmax_rx_pktlen=%d\n\tmax rx queues=%d\n\tmax tx queues=%d",
                 (int) dev_info.min_rx_bufsize, (int) dev_info.max_rx_pktlen, (int) dev_info.max_rx_queues, (int) dev_info.max_tx_queues);
 
@@ -1206 +1206 @@
          */
 #if DEBUG
-    printf("Creating mempool named %s\n", format_data->mempool_name);
+    fprintf(stderr, "Creating mempool named %s\n", format_data->mempool_name);
 #endif
     format_data->pktmbuf_pool =
@@ -1243 +1243 @@
     }
 #if DEBUG
-    printf("Doing dev configure\n");
+    fprintf(stderr, "Doing dev configure\n");
 #endif
     /* Initialise the TX queue a minimum value if using this port for
@@ -1259 +1259 @@
     for (i=0; i < rx_queues; i++) {
 #if DEBUG
-    printf("Doing queue configure\n");
+    fprintf(stderr, "Doing queue configure\n");
 #endif
 
@@ -1419 +1419 @@
  * gives it.
  *
- * We then allow a mapper thread to be started on every real core as DPDK would
+ * We then allow a mapper thread to be started on every real core as DPDK would,
  * we also bind these to the corresponding CPU cores.
  *
@@ -1437 +1437 @@
     // in this case physical cores on the system will not exist so we don't bind
     // these to any particular physical core
+    pthread_mutex_lock(&libtrace->libtrace_lock);
     if (reading) {
 #if HAVE_LIBNUMA
@@ -1472 +1473 @@
         // TODO proper libtrace style error here!!
         fprintf(stderr, "Too many threads for DPDK!!\n");
+        pthread_mutex_unlock(&libtrace->libtrace_lock);
         return -1;
     }
@@ -1496 +1498 @@
         if (i != 0) {
             fprintf(stderr, "Warning pthread_setaffinity_np failed\n");
+            pthread_mutex_unlock(&libtrace->libtrace_lock);
             return -1;
         }
@@ -1508 +1511 @@
         }
     }
+    pthread_mutex_unlock(&libtrace->libtrace_lock);
     return 0;
 }
@@ -1523 +1527 @@
 
     assert(rte_lcore_id() < RTE_MAX_LCORE);
-
+    pthread_mutex_lock(&libtrace->libtrace_lock);
     // Skip if master!!
     if (rte_lcore_id() == rte_get_master_lcore()) {
         fprintf(stderr, "INFO: we are skipping unregistering the master lcore\n");
+        pthread_mutex_unlock(&libtrace->libtrace_lock);
         return;
     }
@@ -1535 +1540 @@
     RTE_PER_LCORE(_lcore_id) = -1; // Might make the world burn if used again
     assert(cfg->lcore_count >= 1); // We cannot unregister the master LCORE!!
+    pthread_mutex_unlock(&libtrace->libtrace_lock);
     return;
 }

lib/libtrace_int.h

@@ -193 +193 @@
  */
 struct libtrace_thread_t {
-        int accepted_packets; // The number of packets accepted only used if pread
+        uint64_t accepted_packets; // The number of packets accepted only used if pread
+        uint64_t filtered_packets;
         // is retreving packets
         // Set to true once the first packet has been stored
@@ -307 +308 @@
         fn_hasher hasher; // If valid using a separate thread
         void *hasher_data;
-        
+        /** The pread_packet choosen path for the configuration */
+        int (*pread)(libtrace_t *, libtrace_thread_t *, libtrace_packet_t **, size_t);
+
         libtrace_thread_t hasher_thread;
         libtrace_thread_t reporter_thread;
@@ -929 +932 @@
         /**
          * Register a thread for use with the format or using the packets produced
-         * by it. This is NOT only used for threads reading packets infact all
+         * by it. This is NOT only used for threads reading packets in fact all
          * threads use this.
+         *
+         * The libtrace lock is not held by this format but can be aquired
+         * by the format.
          *
          * Some use cases include setting up any thread local storage required for

lib/trace.c

@@ -282 +282 @@
         libtrace->dropped_packets = UINT64_MAX;
         libtrace->received_packets = UINT64_MAX;
+        libtrace->pread = NULL;
         ZERO_USER_CONFIG(libtrace->config);
 
@@ -396 +397 @@
         libtrace->perpkt_threads = NULL;
         libtrace->tracetime = 0;
+        libtrace->pread = NULL;
         ZERO_USER_CONFIG(libtrace->config);
         
@@ -1957 +1959 @@
 {
         assert(trace);
+        int i = 0;
+        uint64_t ret = trace->filtered_packets;
+        for (i = 0; i < trace->perpkt_thread_count; i++) {
+                ret += trace->perpkt_threads[i].filtered_packets;
+        }
         if (trace->format->get_filtered_packets) {
                 return trace->format->get_filtered_packets(trace)+
-                        trace->filtered_packets;
-        }
-        return trace->filtered_packets;
+                        ret;
+        }
+        return ret;
 }
 

lib/trace_parallel.c

@@ -101 +101 @@
 #include <unistd.h>
 
-
-static size_t trace_pread_packet(libtrace_t *libtrace, libtrace_thread_t *t, libtrace_packet_t *packets[], size_t nb_packets);
-
+static inline int delay_tracetime(libtrace_t *libtrace, libtrace_packet_t *packet, libtrace_thread_t *t);
 extern int libtrace_parallel;
 
@@ -226 +224 @@
 
         if (trace->config.debug_state)
-                fprintf(stderr, "Thread %d state changed from %d to %d\n", (int) t->tid,
-                        prev_state, t->state);
+                fprintf(stderr, "Thread %d state changed from %d to %d\n",
+                        (int) t->tid, prev_state, t->state);
 
         pthread_cond_broadcast(&trace->perpkt_cond);
@@ -293 +291 @@
 void libtrace_zero_thread(libtrace_thread_t * t) {
         t->accepted_packets = 0;
+        t->filtered_packets = 0;
         t->recorded_first = false;
         t->tracetime_offset_usec = 0;
@@ -387 +386 @@
 }
 
-
-
-/**
- * Dispatches packets to their correct place and applies any translations
- * as needed.
- *
+/**
+ * Sends a packet to the user, expects either a valid packet or a TICK packet.
+ *
+ * Note READ_MESSAGE will only be returned if tracetime is true.
+ *
+ * @brief dispatch_packet
  * @param trace
  * @param t
- * @param packet (in, out) this will be set to NULL if the user doesn't return the packet for reuse
- * @return -1 if an error or EOF has occured and the trace should end, otherwise a postive number (or 0)
- * representing the number of packets returned, these will be at the beginning of the array.
- */
-static inline int dispatch_packets(libtrace_t *trace, libtrace_thread_t *t, libtrace_packet_t **packets,
-                                   size_t nb_packets) {
-        libtrace_message_t message;
-        size_t i, empty = 0;
-        for (i = 0; i < nb_packets; ++i) {
-                if (packets[i]->error > 0) {
-                        packets[i] = (*trace->per_pkt)(trace, packets[i], NULL, t);
-                        trace_fin_packet(packets[i]);
-                } else if (packets[i]->error == READ_TICK) {
-                        message.code = MESSAGE_TICK;
-                        message.additional.uint64 = trace_packet_get_order(packets[i]);
-                        message.sender = t;
-                        (*trace->per_pkt)(trace, NULL, &message, t);
-                } else if (packets[i]->error != READ_MESSAGE) {
-                        // An error this should be the last packet we read
-                        size_t z;
-                        // We could have an eof or error and a message such as pause
-                        for (z = i + 1 ; z < nb_packets; ++z) {
-                                fprintf(stderr, "i=%d nb_packets=%d err=%d, seq=%d\n", (int) z, (int) nb_packets, packets[z]->error, (int) packets[z]->order);
-                                assert (packets[z]->error <= 0);
-                        }
-                        return -1;
-                }
-                if (packets[i]) {
-                        // Move full slots to front
-                        if (empty != i) {
-                                packets[empty] = packets[i];
-                                packets[i] = NULL;
-                        }
-                        ++empty;
-                        // Finish packets while still in CPU cache
-                }
-        }
-        return empty;
-}
-
-static inline int dispatch_packet(libtrace_t *trace, libtrace_thread_t *t, libtrace_packet_t **packet) {
-        libtrace_message_t message;
+ * @param packet A pointer to the packet storage, which may be set to null upon
+ *               return, or a packet to be finished.
+ * @return 0 is successful, otherwise if playing back in tracetime
+ *         READ_MESSAGE(-2) can be returned in which case the packet is not sent.
+ */
+static inline int dispatch_packet(libtrace_t *trace,
+                                                 libtrace_thread_t *t,
+                                                 libtrace_packet_t **packet,
+                                                 bool tracetime) {
         if ((*packet)->error > 0) {
+                if (tracetime) {
+                        if (delay_tracetime(trace, packet[0], t) == READ_MESSAGE)
+                                return READ_MESSAGE;
+                }
                 *packet = (*trace->per_pkt)(trace, *packet, NULL, t);
                 trace_fin_packet(*packet);
-        } else if ((*packet)->error == READ_TICK) {
+        } else {
+                libtrace_message_t message;
+                assert((*packet)->error == READ_TICK);
                 message.code = MESSAGE_TICK;
                 message.additional.uint64 = trace_packet_get_order(*packet);
                 message.sender = t;
                 (*trace->per_pkt)(trace, NULL, &message, t);
-        } else if ((*packet)->error != READ_MESSAGE) {
-                return -1;
         }
         return 0;
+}
+
+/**
+ * Pauses a per packet thread, messages will not be processed when the thread
+ * is paused.
+ *
+ * This process involves reading packets if a hasher thread is used. As such
+ * this function can fail to pause due to errors when reading in which case
+ * the thread should be stopped instead.
+ *
+ *
+ * @brief trace_perpkt_thread_pause
+ * @return READ_ERROR(-1) or READ_EOF(0) or 1 if successfull
+ */
+static int trace_perpkt_thread_pause(libtrace_t *trace, libtrace_thread_t *t,
+                                     libtrace_packet_t *packets[],
+                                     int *nb_packets, int *empty, int *offset) {
+        libtrace_message_t message = {0};
+        libtrace_packet_t * packet = NULL;
+
+        /* Let the user thread know we are going to pause */
+        message.code = MESSAGE_PAUSING;
+        message.sender = t;
+        (*trace->per_pkt)(trace, NULL, &message, t);
+
+        /* Send through any remaining packets (or messages) without delay */
+
+        /* First send those packets already read, as fast as possible
+         * This should never fail or check for messages etc. */
+        for (;*offset < *nb_packets; ++*offset) {
+                ASSERT_RET(dispatch_packet(trace, t, &packets[*offset], false), == 0);
+                /* Move full slots to front as we go */
+                if (packets[*offset]) {
+                        if (*empty != *offset) {
+                                packets[*empty] = packets[*offset];
+                                packets[*offset] = NULL;
+                        }
+                        ++*empty;
+                }
+        }
+
+        libtrace_ocache_alloc(&trace->packet_freelist, (void **) &packet, 1, 1);
+        /* If a hasher thread is running, empty input queues so we don't lose data */
+        if (trace_has_dedicated_hasher(trace)) {
+                fprintf(stderr, "Trace is using a hasher thread emptying queues\n");
+                // The hasher has stopped by this point, so the queue shouldn't be filling
+                while(!libtrace_ringbuffer_is_empty(&t->rbuffer) || t->format_data) {
+                        int ret = trace->pread(trace, t, &packet, 1);
+                        if (ret == 1) {
+                                if (packet->error > 0) {
+                                        store_first_packet(trace, packet, t);
+                                }
+                                ASSERT_RET(dispatch_packet(trace, t, &packet, 1), == 0);
+                                if (packet == NULL)
+                                        libtrace_ocache_alloc(&trace->packet_freelist, (void **) &packet, 1, 1);
+                        } else if (ret != READ_MESSAGE) {
+                                /* Ignore messages we pick these up next loop */
+                                assert (ret == READ_EOF || ret == READ_ERROR);
+                                /* Verify no packets are remaining */
+                                /* TODO refactor this sanity check out!! */
+                                while (!libtrace_ringbuffer_is_empty(&t->rbuffer)) {
+                                        ASSERT_RET(trace->pread(trace, t, &packet, 1), <= 0);
+                                        // No packets after this should have any data in them
+                                        assert(packet->error <= 0);
+                                }
+                                fprintf(stderr, "PREAD_FAILED %d\n", ret);
+                                libtrace_ocache_free(&trace->packet_freelist, (void **) &packet, 1, 1);
+                                return -1;
+                        }
+                }
+        }
+        libtrace_ocache_free(&trace->packet_freelist, (void **) &packet, 1, 1);
+
+        /* Now we do the actual pause, this returns when we resumed */
+        trace_thread_pause(trace, t);
+        message.code = MESSAGE_RESUMING;
+        (*trace->per_pkt)(trace, NULL, &message, t);
+        return 1;
 }
 
@@ -456 +504 @@
 static void* perpkt_threads_entry(void *data) {
         libtrace_t *trace = (libtrace_t *)data;
-        libtrace_thread_t * t;
+        libtrace_thread_t *t;
         libtrace_message_t message = {0};
         libtrace_packet_t *packets[trace->config.burst_size];
-        size_t nb_packets;
         size_t i;
-        int ret;
+        //int ret;
+        /* The current reading position into the packets */
+        int offset = 0;
+        /* The number of packets last read */
+        int nb_packets = 0;
+        /* The offset to the first NULL packet upto offset */
+        int empty = 0;
 
         /* Wait until trace_pstart has been completed */
@@ -473 +526 @@
         }
         //printf("Yay Started perpkt thread #%d\n", (int) get_thread_table_num(trace));
+        ASSERT_RET(pthread_mutex_unlock(&trace->libtrace_lock), == 0);
+
         if (trace->format->pregister_thread) {
                 trace->format->pregister_thread(trace, t, !trace_has_dedicated_hasher(trace));
         }
-        ASSERT_RET(pthread_mutex_unlock(&trace->libtrace_lock), == 0);
 
         /* Fill our buffer with empty packets */
@@ -498 +552 @@
 
                 if (libtrace_message_queue_try_get(&t->messages, &message) != LIBTRACE_MQ_FAILED) {
+                        int ret;
                         switch (message.code) {
                                 case MESSAGE_DO_PAUSE: // This is internal
-                                        // Send message to say we are pausing, TODO consider sender
-                                        message.code = MESSAGE_PAUSING;
-                                        message.sender = t;
-                                        (*trace->per_pkt)(trace, NULL, &message, t);
-                                        // If a hasher thread is running empty input queues so we don't lose data
-                                        if (trace_has_dedicated_hasher(trace)) {
-                                                fprintf(stderr, "Trace is using a hasher thread emptying queues\n");
-                                                // The hasher has stopped by this point, so the queue shouldn't be filling
-                                                while(!libtrace_ringbuffer_is_empty(&t->rbuffer)) {
-                                                        ASSERT_RET(trace_pread_packet(trace, t, packets, 1), == 1);
-                                                        if (dispatch_packets(trace, t, packets, 1) == -1) {
-                                                                // EOF or error, either way we'll stop
-                                                                while (!libtrace_ringbuffer_is_empty(&t->rbuffer)) {
-                                                                        ASSERT_RET(trace_pread_packet(trace, t, packets, 1), == 1);
-                                                                        // No packets after this should have any data in them
-                                                                        assert(packets[0]->error <= 0);
-                                                                }
-                                                                goto stop;
-                                                        }
-                                                }
+                                        ret = trace_perpkt_thread_pause(trace, t, packets, &nb_packets, &empty, &offset);
+                                        if (ret == READ_EOF) {
+                                                fprintf(stderr, "PAUSE stop eof!!\n");
+                                                goto eof;
+                                        } else if (ret == READ_ERROR) {
+                                                fprintf(stderr, "PAUSE stop error!!\n");
+                                                goto error;
                                         }
-                                        // Now we do the actual pause, this returns when we are done
-                                        trace_thread_pause(trace, t);
-                                        message.code = MESSAGE_RESUMING;
-                                        (*trace->per_pkt)(trace, NULL, &message, t);
-                                        // Check for new messages as soon as we return
+                                        assert(ret == 1);
                                         continue;
                                 case MESSAGE_DO_STOP: // This is internal
-                                        goto stop;
+                                        fprintf(stderr, "DO_STOP stop!!\n");
+                                        goto eof;
                         }
                         (*trace->per_pkt)(trace, NULL, &message, t);
+                        /* Continue and the empty messages out before packets */
                         continue;
                 }
 
-                if (trace->perpkt_thread_count == 1) {
-                        assert(packets[0]);
-                        packets[0]->error = trace_read_packet(trace, packets[0]);
-                        if (dispatch_packet(trace, t, &packets[0]) != 0)
-                                break;
-                        if (!packets[0]) {
-                                libtrace_ocache_alloc(&trace->packet_freelist, (void **) &packets[0], 1, 1);
+
+                /* Do we need to read a new set of packets MOST LIKELY we do */
+                if (offset == nb_packets) {
+                        /* Refill the packet buffer */
+                        if (empty != nb_packets) {
+                                // Refill the empty packets
+                                libtrace_ocache_alloc(&trace->packet_freelist,
+                                                      (void **) &packets[empty],
+                                                      nb_packets - empty,
+                                                      nb_packets - empty);
+                        }
+                        if (!trace->pread) {
+                                assert(packets[0]);
+                                nb_packets = trace_read_packet(trace, packets[0]);
+                                packets[0]->error = nb_packets;
+                                if (nb_packets > 0)
+                                        nb_packets = 1;
+                        } else {
+                                nb_packets = trace->pread(trace, t, packets, trace->config.burst_size);
+                        }
+                        offset = 0;
+                        empty = 0;
+                }
+
+                /* Handle error/message cases */
+                if (nb_packets > 0) {
+                        /* Store the first packet */
+                        if (packets[0]->error > 0) {
+                                store_first_packet(trace, packets[0], t);
+                        }
+                        for (;offset < nb_packets; ++offset) {
+                                int ret;
+                                ret = dispatch_packet(trace, t, &packets[offset], trace->tracetime);
+                                if (ret == 0) {
+                                        /* Move full slots to front as we go */
+                                        if (packets[offset]) {
+                                                if (empty != offset) {
+                                                        packets[empty] = packets[offset];
+                                                        packets[offset] = NULL;
+                                                }
+                                                ++empty;
+                                        }
+                                } else {
+                                        assert(ret == READ_MESSAGE);
+                                        /* Loop around and process the message, note */
+                                        continue;
+                                }
                         }
                 } else {
-                        nb_packets = trace_pread_packet(trace, t, packets, trace->config.burst_size);
-                        // Loop through the packets we just read and refill
-                        ret = dispatch_packets(trace, t, packets, nb_packets);
-                        if (ret == -1)
-                                break;
-                        else if (ret != nb_packets) {
-                                // Refill the empty packets
-                                //printf("Refilling packets ret=%d nb_packets=%zd\n", ret, nb_packets);
-                                libtrace_ocache_alloc(&trace->packet_freelist, (void **) &packets[ret], nb_packets - ret, nb_packets - ret);
+                        switch (nb_packets) {
+                        case READ_EOF:
+                                fprintf(stderr, "EOF stop %d!!\n", nb_packets);
+                                goto eof;
+                        case READ_ERROR:
+                                fprintf(stderr, "ERROR stop %d!!\n", nb_packets);
+                                goto error;
+                        case READ_MESSAGE:
+                                nb_packets = 0;
+                                continue;
+                        default:
+                                fprintf(stderr, "Unexpected error %d!!\n", nb_packets);
+                                goto error;
                         }
                 }
-        }
-
-
-stop:
+
+        }
+
+error:
+        fprintf(stderr, "An error occured in trace\n");
+eof:
+        fprintf(stderr, "An eof occured in trace\n");
         /* ~~~~~~~~~~~~~~ Trace is finished do tear down ~~~~~~~~~~~~~~~~~~~~~ */
 
@@ -610 +697 @@
         libtrace_packet_t * packet;
         libtrace_message_t message = {0};
+        int pkt_skipped = 0;
 
         assert(trace_has_dedicated_hasher(trace));
@@ -623 +711 @@
 
         printf("Hasher Thread started\n");
+        ASSERT_RET(pthread_mutex_unlock(&trace->libtrace_lock), == 0);
+
         if (trace->format->pregister_thread) {
                 trace->format->pregister_thread(trace, t, true);
         }
-        ASSERT_RET(pthread_mutex_unlock(&trace->libtrace_lock), == 0);
-        int pkt_skipped = 0;
+
         /* Read all packets in then hash and queue against the correct thread */
         while (1) {
@@ -750 +839 @@
 }
 
-/**
- * @brief Move NULLs to the end of an array.
- * @param values
- * @param len
- * @return The location the first NULL, aka the number of non NULL elements
- */
-static inline size_t move_nulls_back(void *arr[], size_t len) {
-        size_t fr=0, en = len-1;
-        // Shift all non NULL elements to the front of the array, and NULLs to the
-        // end, traverses every element at most once
-        for (;fr < en; ++fr) {
-                if (arr[fr] == NULL) {
-                        for (;en > fr; --en) {
-                                if(arr[en]) {
-                                        arr[fr] = arr[en];
-                                        arr[en] = NULL;
-                                        break;
-                                }
-                        }
-                }
-        }
-        // This is the index of the first NULL
-        en = MIN(fr, en);
-        // Or the end of the array if this special case
-        if (arr[en])
-                en++;
-        return en;
-}
-
-/** returns the number of packets successfully allocated in the final array
- these will all be at the front of the array */
-inline static size_t fill_array_with_empty_packets(libtrace_t *libtrace, libtrace_packet_t *packets[], size_t nb_packets) {
-        size_t nb;
-        nb = move_nulls_back((void **) packets, nb_packets);
-        mem_hits.read.recycled += nb;
-        nb += libtrace_ocache_alloc(&libtrace->packet_freelist, (void **) &packets[nb], nb_packets - nb, nb_packets - nb);
-        assert(nb_packets == nb);
-        return nb;
-}
-
-
-inline static size_t empty_array_of_packets(libtrace_t *libtrace, libtrace_packet_t *packets[], size_t nb_packets) {
-        size_t nb;
-        nb = move_nulls_back((void **) packets, nb_packets);
-        mem_hits.write.recycled += nb_packets - nb;
-        nb += nb_packets - libtrace_ocache_free(&libtrace->packet_freelist, (void **)packets, nb, nb);
-        memset(packets, 0, nb); // XXX make better, maybe do this in ocache??
-        return nb;
-}
-
 /* Our simplest case when a thread becomes ready it can obtain an exclusive
  * lock to read packets from the underlying trace.
  */
-inline static size_t trace_pread_packet_first_in_first_served(libtrace_t *libtrace, libtrace_thread_t *t, libtrace_packet_t *packets[], size_t nb_packets)
-{
+static int trace_pread_packet_first_in_first_served(libtrace_t *libtrace,
+                                                    libtrace_thread_t *t,
+                                                    libtrace_packet_t *packets[],
+                                                    size_t nb_packets) {
         size_t i = 0;
         //bool tick_hit = false;
@@ -812 +853 @@
         for (i = 0; i < nb_packets; ++i) {
                 packets[i]->error = trace_read_packet(libtrace, packets[i]);
+
                 if (packets[i]->error <= 0) {
-                        ++i;
-                        break;
+                        /* We'll catch this next time if we have already got packets */
+                        if ( i==0 ) {
+                                ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
+                                return packets[i]->error;
+                        } else {
+                                break;
+                        }
                 }
                 /*
@@ -842 +889 @@
  * 2. Move that into the packet provided (packet)
  */
-inline static size_t trace_pread_packet_hasher_thread(libtrace_t *libtrace, libtrace_thread_t *t, libtrace_packet_t **packets, size_t nb_packets)
-{
+inline static int trace_pread_packet_hasher_thread(libtrace_t *libtrace,
+                                                      libtrace_thread_t *t,
+                                                      libtrace_packet_t *packets[],
+                                                      size_t nb_packets) {
         size_t i;
+
+        /* We store the last error message here */
+        if (t->format_data) {
+                fprintf(stderr, "Hit me, ohh yeah got error %d\n",
+                        ((libtrace_packet_t *)t->format_data)->error);
+                return ((libtrace_packet_t *)t->format_data)->error;
+        }
 
         // Always grab at least one
@@ -851 +907 @@
         packets[0] = libtrace_ringbuffer_read(&t->rbuffer);
 
-        if (packets[0]->error < 0)
-                return 1;
+        if (packets[0]->error <= 0 && packets[0]->error != READ_TICK) {
+                fprintf(stderr, "Hit me, ohh yeah returning error %d\n", packets[0]->error);
+                return packets[0]->error;
+        }
 
         for (i = 1; i < nb_packets; i++) {
@@ -861 +919 @@
                         break;
                 }
-                // These are typically urgent
-                if (packets[i]->error < 0)
+
+                /* We will return an error or EOF the next time around */
+                if (packets[i]->error <= 0 && packets[0]->error != READ_TICK) {
+                        /* The message case will be checked automatically -
+                           However other cases like EOF and error will only be
+                           sent once*/
+                        if (packets[i]->error != READ_MESSAGE) {
+                                assert(t->format_data == NULL);
+                                t->format_data = packets[i];
+                                fprintf(stderr, "Hit me, ohh yeah set error %d\n",
+                                        ((libtrace_packet_t *)t->format_data)->error);
+                        }
                         break;
+                }
         }
 
@@ -1222 +1291 @@
 
 /**
- * Delays a packets playback so the playback will be in trace time
- */
-static inline void delay_tracetime(libtrace_t *libtrace, libtrace_packet_t *packet, libtrace_thread_t *t) {
+ * Delays a packets playback so the playback will be in trace time.
+ * This may break early if a message becomes available.
+ *
+ * Requires the first packet for this thread to be received.
+ * @param libtrace  The trace
+ * @param packet    The packet to delay
+ * @param t         The current thread
+ * @return Either READ_MESSAGE(-2) or 0 is successful
+ */
+static inline int delay_tracetime(libtrace_t *libtrace, libtrace_packet_t *packet, libtrace_thread_t *t) {
         struct timeval curr_tv, pkt_tv;
-        uint64_t next_release = t->tracetime_offset_usec; // Time at which to release the packet
+        uint64_t next_release = t->tracetime_offset_usec;
         uint64_t curr_usec;
-        /* Tracetime we might delay releasing this packet */
+
         if (!t->tracetime_offset_usec) {
-                libtrace_packet_t * first_pkt;
+                libtrace_packet_t *first_pkt;
                 struct timeval *sys_tv;
                 int64_t initial_offset;
@@ -1237 +1313 @@
                 pkt_tv = trace_get_timeval(first_pkt);
                 initial_offset = (int64_t)tv_to_usec(sys_tv) - (int64_t)tv_to_usec(&pkt_tv);
+                /* In the unlikely case offset is 0, change it to 1 */
                 if (stable)
-                        // 0->1 because 0 is used to mean unset
                         t->tracetime_offset_usec = initial_offset ? initial_offset: 1;
                 next_release = initial_offset;
@@ -1248 +1324 @@
         curr_usec = tv_to_usec(&curr_tv);
         if (next_release > curr_usec) {
+                int ret, mesg_fd = libtrace_message_queue_get_fd(&t->messages);
+                struct timeval delay_tv = usec_to_tv(next_release-curr_usec);
+                fd_set rfds;
+                FD_ZERO(&rfds);
+                FD_SET(mesg_fd, &rfds);
                 // We need to wait
-                struct timeval delay_tv = usec_to_tv(next_release-curr_usec);
+
                 //printf("WAITING for %d.%d next=%"PRIu64" curr=%"PRIu64" seconds packettime %f\n", delay_tv.tv_sec, delay_tv.tv_usec, next_release, curr_usec, trace_get_seconds(packet));
-                select(0, NULL, NULL, NULL, &delay_tv);
-        }
-}
-
-/* Read one packet from the trace into a buffer. Note that this function will
- * block until a packet is read (or EOF is reached).
- *
- * @param libtrace      the trace
- * @param t     The thread
- * @param packets       an array of packets
- * @param nb_packets
- * @returns The number of packets read or 0 on EOF, negative value on error
- *
- * Note this is identical to read_packet but calls pread_packet instead of
- * read packet in the format.
- *
- */
-static inline int trace_pread_packet_wrapper(libtrace_t *libtrace, libtrace_thread_t *t, libtrace_packet_t **packets, size_t nb_packets) {
+                ret = select(mesg_fd+1, &rfds, NULL, NULL, &delay_tv);
+                if (ret == 0) {
+                        return 0;
+                } else if (ret > 0) {
+                        return READ_MESSAGE;
+                } else {
+                        fprintf(stderr, "I thnik we broke select\n");
+                }
+        }
+        return 0;
+}
+
+/* Discards packets that don't match the filter.
+ * Discarded packets are emptied and then moved to the end of the packet list.
+ *
+ * @param trace       The trace format, containing the filter
+ * @param packets     An array of packets
+ * @param nb_packets  The number of valid items in packets
+ *
+ * @return The number of packets that passed the filter, which are moved to
+ *          the start of the packets array
+ */
+static inline size_t filter_packets(libtrace_t *trace,
+                                    libtrace_packet_t **packets,
+                                    size_t nb_packets) {
+        size_t offset = 0;
         size_t i;
+
+        for (i = 0; i < nb_packets; ++i) {
+                // The filter needs the trace attached to receive the link type
+                packets[i]->trace = trace;
+                if (trace_apply_filter(trace->filter, packets[i])) {
+                        libtrace_packet_t *tmp;
+                        tmp = packets[offset];
+                        packets[offset++] = packets[i];
+                        packets[i] = tmp;
+                } else {
+                        trace_fin_packet(packets[i]);
+                }
+        }
+
+        return offset;
+}
+
+/* Read a batch of packets from the trace into a buffer.
+ * Note that this function will block until a packet is read (or EOF is reached)
+ *
+ * @param libtrace    The trace
+ * @param t           The thread
+ * @param packets     An array of packets
+ * @param nb_packets  The number of empty packets in packets
+ * @return The number of packets read, 0 on EOF (or an error/message -1,-2).
+ */
+static int trace_pread_packet_wrapper(libtrace_t *libtrace,
+                                      libtrace_thread_t *t,
+                                      libtrace_packet_t *packets[],
+                                      size_t nb_packets) {
+        int i;
         assert(nb_packets);
-        assert(libtrace && "You called trace_read_packet() with a NULL libtrace parameter!\n");
+        assert(libtrace && "libtrace is NULL in trace_read_packet()");
         if (trace_is_err(libtrace))
                 return -1;
         if (!libtrace->started) {
-                trace_set_err(libtrace,TRACE_ERR_BAD_STATE,"You must call libtrace_start() before trace_read_packet()\n");
+                trace_set_err(libtrace, TRACE_ERR_BAD_STATE,
+                              "You must call libtrace_start() before trace_read_packet()\n");
                 return -1;
         }
 
-
         if (libtrace->format->pread_packets) {
-                for (i = 0; i < nb_packets; ++i) {
-                        assert(packets[i]);
-                        if (!(packets[i]->buf_control==TRACE_CTRL_PACKET || packets[i]->buf_control==TRACE_CTRL_EXTERNAL)) {
-                                trace_set_err(libtrace,TRACE_ERR_BAD_STATE,"Packet passed to trace_read_packet() is invalid\n");
+                int ret;
+                for (i = 0; i < (int) nb_packets; ++i) {
+                        assert(i[packets]);
+                        if (!(packets[i]->buf_control==TRACE_CTRL_PACKET ||
+                              packets[i]->buf_control==TRACE_CTRL_EXTERNAL)) {
+                                trace_set_err(libtrace,TRACE_ERR_BAD_STATE,
+                                              "Packet passed to trace_read_packet() is invalid\n");
                                 return -1;
                         }
-                        /* Finalise the packets, freeing any resources the format module
-                         * may have allocated it and zeroing all data associated with it.
-                         */
-                        //trace_fin_packet(packets[i]);
-                        /* Store the trace we are reading from into the packet opaque
-                         * structure */
-                        packets[i]->trace = libtrace;
                 }
                 do {
-                        int ret;
-                        ret=libtrace->format->pread_packets(libtrace, t, packets, nb_packets);
+                        ret=libtrace->format->pread_packets(libtrace, t,
+                                                            packets,
+                                                            nb_packets);
+                        /* Error, EOF or message? */
                         if (ret <= 0) {
                                 return ret;
                         }
+
                         if (libtrace->filter) {
-                                /*
-                                 * Discard packets that don't match the filter
-                                 * If that is all of the packets then pread again
-                                 */
-                                int nb_filtered = 0;
-                                libtrace_packet_t *filtered_pkts[ret];
-                                int offset;
-                                for (i = 0; i < ret; ++i) {
-                                        if (!trace_apply_filter(libtrace->filter, packets[i])){
-                                                trace_fin_packet(packets[i]);
-                                                packets[i]->trace = libtrace;
-                                                filtered_pkts[nb_filtered++] = packets[i];
-                                                packets[i] = NULL;
-                                        } else {
-                                                if (libtrace->snaplen>0)
-                                                        /* Snap the packet */
-                                                        trace_set_capture_length(packets[i],
-                                                                        libtrace->snaplen);
-                                                trace_packet_set_order(packets[i], trace_get_erf_timestamp(packets[i]));
-                                        }
-                                }
-                                // TODO this aint thread safe
-                                libtrace->filtered_packets += nb_filtered;
-                                for (i = 0, offset = 0; i < ret; ++i) {
-                                        if (packets[i])
-                                                packets[offset++] = packets[i];
-                                }
-                                assert (ret - offset == nb_filtered);
-                                memcpy(&packets[offset], filtered_pkts, nb_filtered * sizeof(libtrace_packet_t *));
-                                t->accepted_packets -= nb_filtered;
-                        } else {
-                                for (i = 0; i < ret; ++i) {
-                                        if (libtrace->snaplen>0)
-                                                trace_set_capture_length(packets[i],
-                                                                libtrace->snaplen);
-                                        trace_packet_set_order(packets[i], trace_get_erf_timestamp(packets[i]));
-                                }
+                                int remaining;
+                                remaining = filter_packets(libtrace,
+                                                           packets, ret);
+                                t->filtered_packets += ret - remaining;
+                                ret = remaining;
+                        }
+                        for (i = 0; i < ret; ++i) {
+                                packets[i]->trace = libtrace;
+                                /* TODO IN FORMAT?? Like traditional libtrace */
+                                if (libtrace->snaplen>0)
+                                        trace_set_capture_length(packets[i],
+                                                        libtrace->snaplen);
+                                trace_packet_set_order(packets[i], trace_get_erf_timestamp(packets[i]));
                         }
                         t->accepted_packets += ret;
-                        //++libtrace->accepted_packets;
-                        return ret;
-                } while(1);
-        }
-        trace_set_err(libtrace,TRACE_ERR_UNSUPPORTED,"This format does not support reading packets\n");
+                } while(ret == 0);
+                return ret;
+        }
+        trace_set_err(libtrace, TRACE_ERR_UNSUPPORTED,
+                      "This format does not support reading packets\n");
         return ~0U;
-}
-
-/**
- * Selects the correct source for packets, either a parallel source
- * or internal splitting
- *
- * @param libtrace
- * @param t
- * @param packets An array pre-filled with empty finilised packets
- * @param nb_packets The number of packets in the array
- *
- * @return the number of packets read, null packets indicate messages. Check packet->error before
- * assuming a packet is valid.
- */
-static size_t trace_pread_packet(libtrace_t *libtrace, libtrace_thread_t *t,
-                                 libtrace_packet_t *packets[], size_t nb_packets)
-{
-        size_t ret;
-        size_t i;
-        assert(nb_packets);
-
-        if (trace_supports_parallel(libtrace) && !trace_has_dedicated_hasher(libtrace)) {
-                ret = trace_pread_packet_wrapper(libtrace, t, packets, nb_packets);
-                /* Put the error into the first packet */
-                if ((int) ret <= 0) {
-                        packets[0]->error = ret;
-                        ret = 1;
-                }
-        } else if (trace_has_dedicated_hasher(libtrace)) {
-                ret = trace_pread_packet_hasher_thread(libtrace, t, packets, nb_packets);
-        } else if (!trace_has_dedicated_hasher(libtrace)) {
-                /* We don't care about which core a packet goes to */
-                ret = trace_pread_packet_first_in_first_served(libtrace, t, packets, nb_packets);
-        } /* else {
-                ret = trace_pread_packet_hash_locked(libtrace, packet);
-        }*/
-
-        // Formats can also optionally do this internally to ensure the first
-        // packet is always reported correctly
-        assert(ret);
-        assert(ret <= nb_packets);
-        if (packets[0]->error > 0) {
-                store_first_packet(libtrace, packets[0], t);
-                if (libtrace->tracetime)
-                        delay_tracetime(libtrace, packets[0], t);
-        }
-
-        return ret;
 }
 
@@ -1588 +1635 @@
                 printf("This format has direct support for p's\n");
                 ret = libtrace->format->pstart_input(libtrace);
+                libtrace->pread = trace_pread_packet_wrapper;
         } else {
                 if (libtrace->format->start_input) {
                         ret = libtrace->format->start_input(libtrace);
                 }
+                if (libtrace->perpkt_thread_count > 1)
+                        libtrace->pread = trace_pread_packet_first_in_first_served;
+                else
+                        libtrace->pread = NULL;
         }
 
@@ -1616 +1668 @@
                         goto cleanup_started;
                 }
+                libtrace->pread = trace_pread_packet_hasher_thread;
         } else {
                 libtrace->hasher_thread.type = THREAD_EMPTY;