Changeset f17b546 for lib

Timestamp:

08/26/15 10:18:31 (6 years ago)

Author:

Shane Alcock <salcock@…>

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:

285acaa

Parents:

7c17e4a (diff), 9a3a846 (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 remote branch 'parallel/develop' into libtrace4

Location:

lib

Files:

• data-struct/linked_list.c (modified) (1 diff)
• data-struct/object_cache.c (modified) (14 diffs)
• format_linux_common.c (modified) (1 diff)
• format_linux_int.c (modified) (1 diff)
• format_linux_ring.c (modified) (1 diff)
• libtrace_int.h (modified) (3 diffs)
• libtrace_parallel.h (modified) (13 diffs)
• trace.c (modified) (11 diffs)
• trace_parallel.c (modified) (25 diffs)

lib/data-struct/linked_list.c

@@ -52 +52 @@
 
         new->prev = NULL;
-        memcpy(&new->data, item, l->element_size);
+        memcpy(new->data, item, l->element_size);
 
         if (l->head == NULL) {

lib/data-struct/object_cache.c

@@ -27 +27 @@
 };
 
+#ifdef ENABLE_MEM_STATS
 extern __thread struct mem_stats mem_hits;
+#endif
 
 struct local_caches {
@@ -123 +125 @@
 /* Get TLS for the list of local_caches */
 static inline struct local_caches *get_local_caches() {
+#if HAVE_TLS
         static __thread struct local_caches *lcs = NULL;
         if (lcs) {
                 return lcs;
-        } else {
+        }
+#else
+        struct local_caches *lcs;
+        pthread_once(&memory_destructor_once, &once_memory_cache_key_init);
+        if ((lcs=pthread_getspecific(memory_destructor_key)) != 0) {
+                return lcs;
+        }
+#endif
+        else {
                 /* This thread has not been used with a memory pool before */
                 /* Allocate our TLS */
@@ -280 +291 @@
                 memcpy(values, &lc->cache[lc->used - nb_buffers], sizeof(void *) * nb_buffers);
                 lc->used -= nb_buffers;
+#ifdef ENABLE_MEM_STATS
                 mem_hits.read.cache_hit += nb_buffers;
                 mem_hits.readbulk.cache_hit += 1;
+#endif
                 return nb_buffers;
         }
@@ -287 +300 @@
         else if (nb_buffers > lc->total) {
                 i = libtrace_ringbuffer_sread_bulk(rb, values, nb_buffers, min_nb_buffers);
+#ifdef ENABLE_MEM_STATS
                 if (i)
                         mem_hits.readbulk.ring_hit += 1;
@@ -292 +306 @@
                         mem_hits.readbulk.miss += 1;
                 mem_hits.read.ring_hit += i;
+#endif
         } else { // Not enough cached
                 // Empty the cache and re-fill it and then see what we're left with
                 i = lc->used;
                 memcpy(values, lc->cache, sizeof(void *) * lc->used);
+#ifdef ENABLE_MEM_STATS
                 mem_hits.read.cache_hit += i;
+#endif
 
                 // Make sure we still meet the minimum requirement
@@ -303 +320 @@
                 else
                         lc->used = libtrace_ringbuffer_sread_bulk(rb, lc->cache, lc->total, 0);
-
+#ifdef ENABLE_MEM_STATS
                 if (lc->used == lc->total)
                         mem_hits.readbulk.ring_hit += 1;
@@ -309 +326 @@
                         mem_hits.readbulk.miss += 1;
                 mem_hits.read.ring_hit += lc->used;
+#endif
         }
 
@@ -319 +337 @@
                 i += remaining;
         }
+#ifdef ENABLE_MEM_STATS
         mem_hits.read.miss += nb_buffers - i;
+#endif
         assert(i >= min_nb_buffers);
         return i;
@@ -379 +399 @@
                 memcpy(&lc->cache[lc->used], values, sizeof(void *) * nb_buffers);
                 lc->used += nb_buffers;
+#ifdef ENABLE_MEM_STATS
                 mem_hits.write.cache_hit += nb_buffers;
                 mem_hits.writebulk.cache_hit += 1;
+#endif
                 return nb_buffers;
         }
@@ -386 +408 @@
         else if (nb_buffers > lc->total) {
                 i = libtrace_ringbuffer_swrite_bulk(rb, values, nb_buffers, min_nb_buffers);
+#ifdef ENABLE_MEM_STATS
                 if (i)
                         mem_hits.writebulk.ring_hit += 1;
@@ -391 +414 @@
                         mem_hits.writebulk.miss += 1;
                 mem_hits.write.ring_hit += i;
+#endif
         } else { // Not enough cache space but there might later
                 // Fill the cache and empty it and then see what we're left with
                 i = (lc->total - lc->used);
                 memcpy(&lc->cache[lc->used], values, sizeof(void *) * i);
+#ifdef ENABLE_MEM_STATS
                 mem_hits.write.cache_hit += i;
+#endif
 
                 // Make sure we still meet the minimum requirement
@@ -407 +433 @@
                         memmove(lc->cache, &lc->cache[lc->total - lc->used], sizeof(void *) * lc->used);
 
+#ifdef ENABLE_MEM_STATS
                 if (lc->used)
                         mem_hits.writebulk.miss += 1;
@@ -412 +439 @@
                         mem_hits.writebulk.ring_hit += 1;
                 mem_hits.write.ring_hit += lc->total - lc->used;
+#endif
         }
 
@@ -422 +450 @@
                 i += remaining;
         }
+#ifdef ENABLE_MEM_STATS
         mem_hits.write.miss += nb_buffers - i;
+#endif
         return i;
 }

lib/format_linux_common.c

@@ -661 +661 @@
                         linuxcommon_close_input_stream(libtrace, stream);
                 }
-                libtrace_list_deinit(FORMAT_DATA->per_stream);
-                free(libtrace->format_data);
-                libtrace->format_data = NULL;
                 return -1;
         }

lib/format_linux_int.c

@@ -69 +69 @@
 {
         int ret = linuxcommon_start_input_stream(libtrace, FORMAT_DATA_FIRST);
-        if (ret != 0) {
-                libtrace_list_deinit(FORMAT_DATA->per_stream);
-                free(libtrace->format_data);
-                libtrace->format_data = NULL;
-        }
         return ret;
 }

lib/format_linux_ring.c

@@ -280 +280 @@
 {
         int ret = linuxring_start_input_stream(libtrace, FORMAT_DATA_FIRST);
-        if (ret != 0) {
-                libtrace_list_deinit(FORMAT_DATA->per_stream);
-                free(libtrace->format_data);
-                libtrace->format_data = NULL;
-        }
         return ret;
 }

lib/libtrace_int.h

@@ -307 +307 @@
         /** The sequence is like accepted_packets but we don't reset this after a pause. */
         uint64_t sequence_number;
+        /** The packet read out by the trace, backwards compatibility to allow us to finalise
+         * a packet when the trace is destroyed */
+        libtrace_packet_t *last_packet;
         /** The filename from the uri for the trace */
         char *uridata;
@@ -330 +333 @@
         /** The actual freelist */
         libtrace_ocache_t packet_freelist;
-        /** User defined per_pkt function called when a pkt is ready */
-        fn_per_pkt per_pkt;
-        /** User defined reporter function entry point XXX not hooked up */
+        /** User defined per_msg function called when a message is ready */
+        fn_cb_msg per_msg;
+        /** User defined reporter function entry point */
         fn_reporter reporter;
         /** The hasher function */
@@ -358 +361 @@
         struct user_configuration config;
         libtrace_combine_t combiner;
+        struct {
+                fn_cb_starting message_starting;
+                fn_cb_dataless message_stopping;
+                fn_cb_dataless message_resuming;
+                fn_cb_dataless message_pausing;
+                fn_cb_packet message_packet;
+                fn_cb_first_packet message_first_packet;
+                fn_cb_tick message_tick_count;
+                fn_cb_tick message_tick_interval;
+        } callbacks;
 };
 

lib/libtrace_parallel.h

@@ -154 +154 @@
          * @param data unused, do not use this
          * @param sender The sender will be set as the current thread
+         * @return When using a function callback for starting, the returned
+         * value is stored against the thread tls. Otherwise the return is ignored.
          */
         MESSAGE_STARTING,
@@ -211 +213 @@
          * sent once a new packet is encountered
          *
-         * @
+         * @see trace_get_first_packet()
          */
         MESSAGE_FIRST_PACKET,
@@ -225 +227 @@
          * trace_post_reporter()
          *
-         * @see trace_get_first_packet()
          */
         MESSAGE_POST_REPORTER,
@@ -236 +237 @@
          * with an earlier time-stamp.
          *
-         * @param data data.uint64_t holds the system time-stamp in the
+         * @param data data.uint64 holds the system time-stamp in the
          * erf format
          * @param sender should be ignored
@@ -248 +249 @@
          * threads will see it between the same packets.
          *
-         * @param data The number of packets seen so far across all threads
+         * @param data data.uint64 holds the number of packets seen so far across all threads
          * @param sender Set to the current per-packet thread
          */
@@ -397 +398 @@
 /**
  * The definition for the main function that the user supplies to process
- * packets.
+ * messages.
  *
  * @param trace The trace the packet is related to.
@@ -412 +413 @@
  * documentation for the message as to what value these will contain.
  */
-typedef void* (*fn_per_pkt)(libtrace_t* trace,
-                            libtrace_thread_t *thread,
-                            int mesg_code,
-                            libtrace_generic_t data,
-                            libtrace_thread_t *sender);
+typedef void* (*fn_cb_msg)(libtrace_t* trace,
+                           libtrace_thread_t *thread,
+                           int mesg_code,
+                           libtrace_generic_t data,
+                           libtrace_thread_t *sender);
 
 /**
@@ -452 +453 @@
  * @param libtrace The input trace to start
  * @param global_blob Global data related to this trace accessible using trace_get_global()
- * @param per_pkt A user supplied function called when a packet is ready
+ * @param per_msg A user supplied function called when a message is ready
  * @param reporter A user supplied function called when a result is ready.
  * Optional if NULL the reporter thread will not be started.
@@ -459 +460 @@
  * This can also be used to restart an existing parallel trace,
  * that has previously been paused using trace_ppause().
- * In this case global_blob,per_pkt and reporter will only be updated
+ * In this case global_blob,per_msg and reporter will only be updated
  * if they are non-null. Otherwise their previous values will be maintained.
  *
  */
 DLLEXPORT int trace_pstart(libtrace_t *libtrace, void* global_blob,
-                           fn_per_pkt per_pkt, fn_reporter reporter);
+                           fn_cb_msg per_msg, fn_reporter reporter);
+
+/**
+ *
+ * @param libtrace The parallel trace
+ * @param t The thread that is running
+ * @param global The global storage
+ * @return The returned value is stored against the threads tls.
+ *         This is typically passed as tls argument to other messages.
+ */
+typedef void* (*fn_cb_starting)(libtrace_t *libtrace,
+                                     libtrace_thread_t *t,
+                                     void *global);
+
+/**
+ * @param libtrace The parallel trace
+ * @param t The thread that is running
+ * @param global The global storage
+ * @param tls The thread local storage
+ */
+typedef void (*fn_cb_dataless)(libtrace_t *libtrace,
+                                    libtrace_thread_t *t,
+                                    void *global,
+                                    void *tls);
+
+/**
+ * @param libtrace The parallel trace
+ * @param t The thread that is running
+ * @param global The global storage
+ * @param tls The thread local storage
+ */
+typedef void (*fn_cb_first_packet)(libtrace_t *libtrace,
+                                   libtrace_thread_t *t,
+                                   void *global,
+                                   void *tls,
+                                   libtrace_packet_t *first_packet,
+                                   libtrace_thread_t *sender);
+
+/**
+ * @param libtrace The parallel trace
+ * @param t The thread that is running
+ * @param global The global storage
+ * @param tls The thread local storage
+ * @param uint64_t Either the timestamp or packet count depending on message type
+ */
+typedef void (*fn_cb_tick)(libtrace_t *libtrace,
+                           libtrace_thread_t *t,
+                           void *global,
+                           void *tls,
+                           uint64_t order);
+
+/**
+ * @param libtrace The parallel trace
+ * @param t The thread
+ * @param packet The packet associated with the message
+ * @param global The global storage
+ * @param tls The thread local storage
+ *
+ * @return optionally a packet which is handed back to the library,
+ *         typically this is the packet supplied. Otherwise NULL.
+ */
+typedef libtrace_packet_t* (*fn_cb_packet)(libtrace_t *libtrace,
+                                           libtrace_thread_t *t,
+                                           void *global,
+                                           void *tls,
+                                           libtrace_packet_t *packet);
+
+/** Registers a built-in message with a handler.
+ * Note we do not include the sending thread as an argument to the reporter.
+ * If set to NULL, the message will be sent to default perpkt handler.
+ *
+ * @param libtrace The input trace to start
+ * @param handler the handler to be called when the message is received
+ * @return 0 if successful otherwise -1.
+ */
+
+DLLEXPORT int trace_cb_starting(libtrace_t *libtrace, fn_cb_starting handler);
+DLLEXPORT int trace_cb_stopping(libtrace_t *libtrace, fn_cb_dataless handler);
+DLLEXPORT int trace_cb_resuming(libtrace_t *libtrace, fn_cb_dataless handler);
+DLLEXPORT int trace_cb_pausing(libtrace_t *libtrace, fn_cb_dataless handler);
+DLLEXPORT int trace_cb_packet(libtrace_t *libtrace, fn_cb_packet handler);
+DLLEXPORT int trace_cb_first_packet(libtrace_t *libtrace, fn_cb_first_packet handler);
+DLLEXPORT int trace_cb_tick_count(libtrace_t *libtrace, fn_cb_tick handler);
+DLLEXPORT int trace_cb_tick_interval(libtrace_t *libtrace, fn_cb_tick handler);
 
 /** Pauses a trace previously started with trace_pstart()
@@ -542 +626 @@
 /** Store data against a thread.
  *
- * @param The parallel trace.
- * @param data The new value to save against the trace
+ * @param thread The thread
+ * @param data The new value to save against the thread
  * @return The previously stored value
  *
@@ -628 +712 @@
  * as possible (real-time).
  *
- * @param A parallel input trace
+ * @param trace A parallel input trace
  * @param tracetime If true packets are released with time intervals matching
  * the original trace. Otherwise packets are read as fast as possible.
@@ -810 +894 @@
          *
          * @param key (Typically) The packets order, see trace_packet_get_order()
-         * @param
          */
         RESULT_PACKET,
@@ -966 +1049 @@
  */
 DLLEXPORT bool trace_has_finished(libtrace_t * libtrace);
+
+
+/** Check if libtrace is directly reading from multiple queues
+ * from the format (such as a NICs hardware queues).
+ *
+ * When a parallel trace is running, or if checked after its completion
+ * this returns true if a trace was able to run natively parallel
+ * from the format. Otherwise false is returned, meaning libtrace is
+ * distibuting packets across multiple threads from a single source.
+ *
+ * Factors that may stop this happening despite the format supporting
+ * native parallel reads include: the choice of hasher function,
+ * the number of threads choosen (such as 1 or more than the trace supports)
+ * or another error when trying to start the parallel format.
+ *
+ * If this is called before the trace is started. I.e. before pstart
+ * this returns an indication that the trace has the possiblity to support
+ * native parallel reads. After trace pstart is called this should be
+ * checked again to confirm this has happened.
+ *
+ *
+ * @return true if the trace is parallel or false if the library is splitting
+ * the trace into multiple threads.
+ */
+DLLEXPORT bool trace_is_parallel(libtrace_t * libtrace);
 
 /** Returns either the sequence number or erf timestamp of a packet.

lib/trace.c

@@ -261 +261 @@
         libtrace->filtered_packets = 0;
         libtrace->accepted_packets = 0;
+        libtrace->last_packet = NULL;
         
         /* Parallel inits */
@@ -268 +269 @@
         libtrace->perpkt_queue_full = false;
         libtrace->global_blob = NULL;
-        libtrace->per_pkt = NULL;
+        libtrace->per_msg = NULL;
         libtrace->reporter = NULL;
         libtrace->hasher = NULL;
@@ -287 +288 @@
         ZERO_USER_CONFIG(libtrace->config);
         memset(&libtrace->combiner, 0, sizeof(libtrace->combiner));
+        memset(&libtrace->callbacks, 0, sizeof(libtrace->callbacks));
 
         /* Parse the URI to determine what sort of trace we are dealing with */
@@ -382 +384 @@
         libtrace->io = NULL;
         libtrace->filtered_packets = 0;
+        libtrace->accepted_packets = 0;
+        libtrace->last_packet = NULL;
         
         /* Parallel inits */
@@ -389 +393 @@
         libtrace->perpkt_queue_full = false;
         libtrace->global_blob = NULL;
-        libtrace->per_pkt = NULL;
+        libtrace->per_msg = NULL;
         libtrace->reporter = NULL;
         libtrace->hasher = NULL;
@@ -405 +409 @@
         ZERO_USER_CONFIG(libtrace->config);
         memset(&libtrace->combiner, 0, sizeof(libtrace->combiner));
+        memset(&libtrace->callbacks, 0, sizeof(libtrace->callbacks));
         
         for(tmp=formats_list;tmp;tmp=tmp->next) {
@@ -547 +552 @@
                 return -1;
         }
+
+        /* Finish the last packet we read - for backwards compatibility */
+        if (libtrace->last_packet)
+                trace_fin_packet(libtrace->last_packet);
+        assert(libtrace->last_packet == NULL);
+
         if (libtrace->format->pause_input)
                 libtrace->format->pause_input(libtrace);
+
         libtrace->started=false;
         return 0;
@@ -687 +699 @@
         }
 
+        /* Finish any the last packet we read - for backwards compatibility */
+        if (libtrace->last_packet)
+                trace_fin_packet(libtrace->last_packet);
+        assert(libtrace->last_packet == NULL);
+
         if (libtrace->format) {
                 if (libtrace->started && libtrace->format->pause_input)
@@ -803 +820 @@
                 packet->trace->format->fin_packet(packet);
         }
+        if (packet->trace && packet->trace->last_packet == packet)
+                packet->trace->last_packet = NULL;
         
         if (packet->buf_control == TRACE_CTRL_PACKET && packet->buffer) {
@@ -825 +844 @@
                         packet->trace->format->fin_packet(packet);
                 }
+                if (packet->trace && packet->trace->last_packet == packet)
+                        packet->trace->last_packet = NULL;
 
                 // No matter what we remove the header and link pointers
@@ -903 +924 @@
                         ++libtrace->accepted_packets;
                         ++libtrace->sequence_number;
+                        libtrace->last_packet = packet;
                         return ret;
                 } while(1);

lib/trace_parallel.c

@@ -113 +113 @@
 };
 
+
+#ifdef ENABLE_MEM_STATS
 // Grrr gcc wants this spelt out
 __thread struct mem_stats mem_hits = {{0},{0},{0},{0}};
 
+
 static void print_memory_stats() {
-#if 0
         uint64_t total;
 #if defined(HAVE_PTHREAD_SETNAME_NP) && defined(__linux__)
@@ -163 +165 @@
                                 total, (double) mem_hits.writebulk.miss / (double) total * 100.0);
         }
+}
+#else
+static void print_memory_stats() {}
 #endif
+
+static const libtrace_generic_t gen_zero = {0};
+
+/* This should optimise away the switch to nothing in the explict cases */
+static inline void send_message(libtrace_t *trace, libtrace_thread_t *thread, const enum libtrace_messages type,
+                                libtrace_generic_t data, libtrace_thread_t *sender) {
+        fn_cb_dataless fn = NULL;
+        switch (type) {
+        case MESSAGE_STARTING:
+                if (trace->callbacks.message_starting)
+                        thread->user_data = (*trace->callbacks.message_starting)(trace, thread, trace->global_blob);
+                else if (trace->per_msg)
+                        (*trace->per_msg)(trace, thread, type, data, sender);
+                return;
+        case MESSAGE_FIRST_PACKET:
+                if (trace->callbacks.message_first_packet)
+                        (*trace->callbacks.message_first_packet)(trace, thread, trace->global_blob, thread->user_data, data.pkt, sender);
+                else if (trace->per_msg)
+                        (*trace->per_msg)(trace, thread, type, data, sender);
+                return;
+        case MESSAGE_TICK_COUNT:
+                if (trace->callbacks.message_tick_count)
+                        (*trace->callbacks.message_tick_count)(trace, thread, trace->global_blob, thread->user_data, data.uint64);
+                else if (trace->per_msg)
+                        (*trace->per_msg)(trace, thread, type, data, sender);
+                return;
+        case MESSAGE_TICK_INTERVAL:
+                if (trace->callbacks.message_tick_interval)
+                        (*trace->callbacks.message_tick_interval)(trace, thread, trace->global_blob, thread->user_data,  data.uint64);
+                else if (trace->per_msg)
+                        (*trace->per_msg)(trace, thread, type, data, sender);
+                return;
+        case MESSAGE_STOPPING:
+                fn = trace->callbacks.message_stopping;
+                break;
+        case MESSAGE_RESUMING:
+                fn = trace->callbacks.message_resuming;
+                break;
+        case MESSAGE_PAUSING:
+                fn = trace->callbacks.message_pausing;
+                break;
+
+        /* These should be unused */
+        case MESSAGE_DO_PAUSE:
+        case MESSAGE_DO_STOP:
+        case MESSAGE_POST_REPORTER:
+        case MESSAGE_RESULT:
+        case MESSAGE_PACKET:
+                return;
+        case MESSAGE_USER:
+                break;
+        }
+        if (fn)
+                (*fn)(trace, thread, trace->global_blob, thread->user_data);
+        else if (trace->per_msg)
+                (*trace->per_msg)(trace, thread, type, data, sender);
 }
 
@@ -304 +365 @@
                 pthread_t tid = pthread_self();
                 // Check if we are reporter or something else
-                if (pthread_equal(tid, libtrace->reporter_thread.tid))
+                if (libtrace->hasher_thread.type == THREAD_REPORTER &&
+                                pthread_equal(tid, libtrace->reporter_thread.tid))
                         ret = &libtrace->reporter_thread;
-                else if (pthread_equal(tid, libtrace->hasher_thread.tid))
+                else if (libtrace->hasher_thread.type == THREAD_HASHER &&
+                         pthread_equal(tid, libtrace->hasher_thread.tid))
                         ret = &libtrace->hasher_thread;
                 else
@@ -316 +379 @@
 DLLEXPORT void libtrace_make_packet_safe(libtrace_packet_t *pkt) {
         // Duplicate the packet in standard malloc'd memory and free the
-        // original, This is a 1:1 exchange so is ocache count remains unchanged.
+        // original, This is a 1:1 exchange so the ocache count remains unchanged.
         if (pkt->buf_control != TRACE_CTRL_PACKET) {
                 libtrace_packet_t *dup;
@@ -324 +387 @@
                 /* Copy the duplicated packet over the existing */
                 memcpy(pkt, dup, sizeof(libtrace_packet_t));
+                /* Free the packet structure */
+                free(dup);
         }
 }
@@ -377 +442 @@
                 t->accepted_packets++;
                 libtrace_generic_t data = {.pkt = *packet};
-                *packet = (*trace->per_pkt)(trace, t, MESSAGE_PACKET, data, t);
+                if (trace->callbacks.message_packet)
+                        *packet = (*trace->callbacks.message_packet)(trace, t, trace->global_blob, t->user_data, *packet);
+                else if (trace->per_msg)
+                        *packet = (*trace->per_msg)(trace, t, MESSAGE_PACKET, data, t);
                 trace_fin_packet(*packet);
         } else {
                 assert((*packet)->error == READ_TICK);
                 libtrace_generic_t data = {.uint64 = trace_packet_get_order(*packet)};
-                (*trace->per_pkt)(trace, t, MESSAGE_TICK_COUNT, data, t);
+                send_message(trace, t, MESSAGE_TICK_COUNT, data, t);
         }
         return 0;
@@ -449 +517 @@
 
         /* Let the user thread know we are going to pause */
-        (*trace->per_pkt)(trace, t, MESSAGE_PAUSING, (libtrace_generic_t){0}, t);
+        send_message(trace, t, MESSAGE_PAUSING, gen_zero, t);
 
         /* Send through any remaining packets (or messages) without delay */
@@ -490 +558 @@
         /* Now we do the actual pause, this returns when we resumed */
         trace_thread_pause(trace, t);
-        (*trace->per_pkt)(trace, t, MESSAGE_RESUMING, (libtrace_generic_t){0}, t);
+        send_message(trace, t, MESSAGE_RESUMING, gen_zero, t);
         return 1;
 }
@@ -523 +591 @@
 
         if (trace->format->pregister_thread) {
-                trace->format->pregister_thread(trace, t, !trace_has_dedicated_hasher(trace));
+                trace->format->pregister_thread(trace, t, trace_is_parallel(trace));
         }
 
@@ -535 +603 @@
 
         /* Let the per_packet function know we have started */
-        (*trace->per_pkt)(trace, t, MESSAGE_STARTING, (libtrace_generic_t){0}, t);
-        (*trace->per_pkt)(trace, t, MESSAGE_RESUMING, (libtrace_generic_t){0}, t);
+        send_message(trace, t, MESSAGE_STARTING, gen_zero, t);
+        send_message(trace, t, MESSAGE_RESUMING, gen_zero, t);
 
         for (;;) {
@@ -556 +624 @@
                                         goto eof;
                         }
-                        (*trace->per_pkt)(trace, t, message.code, message.data, message.sender);
+                        (*trace->per_msg)(trace, t, message.code, message.data, message.sender);
+                        (*trace->per_msg)(trace, t, message.code, message.data, message.sender);
                         /* Continue and the empty messages out before packets */
                         continue;
@@ -619 +688 @@
 
         // Let the per_packet function know we have stopped
-        (*trace->per_pkt)(trace, t, MESSAGE_PAUSING, (libtrace_generic_t){0}, t);
-        (*trace->per_pkt)(trace, t, MESSAGE_STOPPING, (libtrace_generic_t){0}, t);
+        send_message(trace, t, MESSAGE_PAUSING, gen_zero, t);
+        send_message(trace, t, MESSAGE_STOPPING, gen_zero, t);
 
         // Free any remaining packets
@@ -1246 +1315 @@
  */
 static int trace_prestart(libtrace_t * libtrace, void *global_blob,
-                          fn_per_pkt per_pkt, fn_reporter reporter) {
+                          fn_cb_msg per_msg, fn_reporter reporter) {
         int i, err = 0;
         if (libtrace->state != STATE_PAUSED) {
@@ -1289 +1358 @@
 
         /* Update functions if requested */
-        if (per_pkt)
-                libtrace->per_pkt = per_pkt;
-        assert(libtrace->per_pkt);
+        if (per_msg)
+                libtrace->per_msg = per_msg;
+        assert(libtrace->per_msg);
         if (reporter)
                 libtrace->reporter = reporter;
@@ -1297 +1366 @@
                 libtrace->global_blob = global_blob;
 
-        if (libtrace->perpkt_thread_count > 1 &&
-            trace_supports_parallel(libtrace) &&
-            !trace_has_dedicated_hasher(libtrace)) {
+        if (trace_is_parallel(libtrace)) {
                 err = libtrace->format->pstart_input(libtrace);
         } else {
@@ -1512 +1579 @@
 }
 
+DLLEXPORT bool trace_is_parallel(libtrace_t * libtrace) {
+        if (libtrace->state == STATE_NEW)
+                return trace_supports_parallel(libtrace);
+        return libtrace->pread == trace_pread_packet_wrapper;
+}
+
 DLLEXPORT int trace_pstart(libtrace_t *libtrace, void* global_blob,
-                           fn_per_pkt per_pkt, fn_reporter reporter) {
+                           fn_cb_msg per_msg, fn_reporter reporter) {
         int i;
         int ret = -1;
@@ -1526 +1599 @@
 
         if (libtrace->state == STATE_PAUSED) {
-                ret = trace_prestart(libtrace, global_blob, per_pkt, reporter);
+                ret = trace_prestart(libtrace, global_blob, per_msg, reporter);
                 ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
                 return ret;
@@ -1541 +1614 @@
         /* Store the user defined things against the trace */
         libtrace->global_blob = global_blob;
-        libtrace->per_pkt = per_pkt;
+        libtrace->per_msg = per_msg;
         libtrace->reporter = reporter;
         /* And zero other fields */
@@ -1559 +1632 @@
         verify_configuration(libtrace);
 
+        ret = -1;
         /* Try start the format - we prefer parallel over single threaded, as
          * these formats should support messages better */
@@ -1565 +1639 @@
                 ret = libtrace->format->pstart_input(libtrace);
                 libtrace->pread = trace_pread_packet_wrapper;
-        } else {
+        }
+        if (ret != 0) {
                 if (libtrace->format->start_input) {
                         ret = libtrace->format->start_input(libtrace);
@@ -1703 +1778 @@
         libtrace->perpkt_thread_states[THREAD_FINISHED] = 0;
 cleanup_started:
-        if (trace_supports_parallel(libtrace) &&
-            !trace_has_dedicated_hasher(libtrace)
-            && libtrace->perpkt_thread_count > 1) {
+        if (libtrace->pread == trace_pread_packet_wrapper) {
                 if (libtrace->format->ppause_input)
                         libtrace->format->ppause_input(libtrace);
@@ -1720 +1793 @@
 }
 
+DLLEXPORT int trace_cb_starting(libtrace_t *libtrace, fn_cb_starting handler) {
+        libtrace->callbacks.message_starting = handler;
+        return 0;
+}
+
+DLLEXPORT int trace_cb_pausing(libtrace_t *libtrace, fn_cb_dataless handler) {
+        libtrace->callbacks.message_pausing = handler;
+        return 0;
+}
+
+DLLEXPORT int trace_cb_resuming(libtrace_t *libtrace, fn_cb_dataless handler) {
+        libtrace->callbacks.message_resuming = handler;
+        return 0;
+}
+
+DLLEXPORT int trace_cb_stopping(libtrace_t *libtrace, fn_cb_dataless handler) {
+        libtrace->callbacks.message_stopping = handler;
+        return 0;
+}
+
+DLLEXPORT int trace_cb_packet(libtrace_t *libtrace, fn_cb_packet handler) {
+        libtrace->callbacks.message_packet = handler;
+        return 0;
+}
+
+DLLEXPORT int trace_cb_first_packet(libtrace_t *libtrace, fn_cb_first_packet handler) {
+        libtrace->callbacks.message_first_packet = handler;
+        return 0;
+}
+
+DLLEXPORT int trace_cb_tick_count(libtrace_t *libtrace, fn_cb_tick handler) {
+        libtrace->callbacks.message_tick_count = handler;
+        return 0;
+}
+
+DLLEXPORT int trace_cb_tick_interval(libtrace_t *libtrace, fn_cb_tick handler) {
+        libtrace->callbacks.message_tick_interval = handler;
+        return 0;
+}
+
 /*
  * Pauses a trace, this should only be called by the main thread
@@ -1728 +1841 @@
  *
  * Once done you should be able to modify the trace setup and call pstart again
- * TODO handle changing thread numbers
+ * TODO add support to change the number of threads.
  */
 DLLEXPORT int trace_ppause(libtrace_t *libtrace)
@@ -1826 +1939 @@
         // Save the statistics against the trace
         trace_get_statistics(libtrace, NULL);
-        if (trace_supports_parallel(libtrace) && !trace_has_dedicated_hasher(libtrace) && libtrace->perpkt_thread_count > 1) {
+        if (trace_is_parallel(libtrace)) {
                 libtrace->started = false;
                 if (libtrace->format->ppause_input)
@@ -2248 +2361 @@
         return 0;
 }
-
-
 
 static bool config_bool_parse(char *value, size_t nvalue) {