Changeset f625817

Timestamp:

09/11/15 15:00:27 (5 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:

322c516

Parents:

8c7490fe

Message:

Reworked callback API and removed old per_msg and reporter functions

Updated tracertstats to use the new callback API.

Extended the callback approach to the reporter thread as well as the per
packet threads.

Added libtrace_callback_set_t structure, which is used to register the
user callback functions.

Added callback functionality for MESSAGE_RESULT (needed now that reporter
threads also do callbacks) and MESSAGE_USER (for user-defined messages). The
MESSAGE_USER callback is essentially the same as the old per_msg function
style.

Updated combiners to use send_message to pass results to the reporter thread.
send_message itself is now no longer static, so that combiners can use it.

Disabled building of tracestats_parallel as it was using the older version
of the callback API. Will update in a future commit.

Files:

• lib/combiner_ordered.c (modified) (3 diffs)
• lib/combiner_sorted.c (modified) (1 diff)
• lib/combiner_unordered.c (modified) (1 diff)
• lib/libtrace.h.in (modified) (2 diffs)
• lib/libtrace_int.h (modified) (4 diffs)
• lib/libtrace_parallel.h (modified) (4 diffs)
• lib/trace.c (modified) (5 diffs)
• lib/trace_parallel.c (modified) (16 diffs)
• tools/tracertstats/tracertstats.c (modified) (5 diffs)
• tools/tracestats/Makefile.am (modified) (2 diffs)

lib/combiner_ordered.c

@@ -65 +65 @@
                                 libtrace_generic_t gt = {.res = &r};
                                 ASSERT_RET (libtrace_deque_pop_front(v, (void *) &r), == 1);
-                                trace->reporter(trace, MESSAGE_RESULT, gt, &trace->reporter_thread);
+                                send_message(trace, &trace->reporter_thread,
+                                                MESSAGE_RESULT, gt,
+                                                &trace->reporter_thread);
                                 return 0;
                         }
@@ -88 +90 @@
                                 libtrace_generic_t gt = {.res = &r};
                                 ASSERT_RET (libtrace_deque_pop_front(v, (void *) &r), == 1);
-                                trace->reporter(trace, MESSAGE_RESULT, gt, &trace->reporter_thread);
+                                send_message(trace, &trace->reporter_thread,
+                                                MESSAGE_RESULT, gt,
+                                                &trace->reporter_thread);
                                 return 0;
                         }
@@ -152 +156 @@
 
                 ASSERT_RET (libtrace_deque_pop_front(&queues[min_queue], (void *) &r), == 1);
-                trace->reporter(trace, MESSAGE_RESULT, gt, &trace->reporter_thread);
+                send_message(trace, &trace->reporter_thread,
+                                MESSAGE_RESULT, gt,
+                                NULL);
 
                 // Now update the one we just removed

lib/combiner_sorted.c

@@ -67 +67 @@
                         continue;
                 }
-                trace->reporter(trace, MESSAGE_RESULT, gt, &trace->reporter_thread);
+                send_message(trace, &trace->reporter_thread, MESSAGE_RESULT,
+                                gt, NULL);
         }
         libtrace_vector_empty(&queues[0]);

lib/combiner_unordered.c

@@ -49 +49 @@
                                 c->last_count_tick = r.key;
                         }
-                        trace->reporter(trace, MESSAGE_RESULT, gt, &trace->reporter_thread);
+                        send_message(trace, &trace->reporter_thread,
+                                MESSAGE_RESULT, gt, NULL);
                 }
         }

lib/libtrace.h.in

@@ -252 +252 @@
 /** Opaque structure holding information about libtrace thread */
 typedef struct libtrace_thread_t libtrace_thread_t;
+
+/** Opaque structure holding callback functions for libtrace threads */
+typedef struct callback_set libtrace_callback_set_t;
 
 /** If the packet has allocated its own memory the buffer_control should be
@@ -1480 +1483 @@
 DLLEXPORT void trace_destroy_output(libtrace_out_t *trace);
 
+/** Create a callback set that can be used to define callbacks for parallel
+  * libtrace threads.
+  *
+  * @return A pointer to a freshly allocated callback set.
+  */
+DLLEXPORT libtrace_callback_set_t *trace_create_callback_set();
+
+/** Destroys a callback set, freeing up an resources it was using.
+ *
+ * @param cbset         The callback set to be destroyed.
+ */
+DLLEXPORT void trace_destroy_callback_set(libtrace_callback_set_t *cbset);
+
+
 /** Check (and clear) the current error state of an input trace
  * @param trace         The input trace to check the error state on

lib/libtrace_int.h

@@ -284 +284 @@
 };
 #define ZERO_USER_CONFIG(config) memset(&config, 0, sizeof(struct user_configuration));
+
+struct callback_set {
+
+        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_result message_result;
+        fn_cb_first_packet message_first_packet;
+        fn_cb_tick message_tick_count;
+        fn_cb_tick message_tick_interval;
+        fn_cb_usermessage message_user;
+};
 
 /** A libtrace input trace 
@@ -333 +347 @@
         /** The actual freelist */
         libtrace_ocache_t packet_freelist;
-        /** 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 */
         enum hasher_types hasher_type;
@@ -361 +371 @@
         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;
+        
+        /* Set of callbacks to be executed by per packet threads in response
+         * to various messages. */
+        struct callback_set *perpkt_cbs;
+        /* Set of callbacks to be executed by the reporter thread in response
+         * to various messages. */
+        struct callback_set *reporter_cbs;
 };
 
@@ -380 +387 @@
 libtrace_thread_t * get_thread_table(libtrace_t *libtrace);
 
+
+void send_message(libtrace_t *trace, libtrace_thread_t *target,
+                const enum libtrace_messages type,
+                libtrace_generic_t data, libtrace_thread_t *sender);
 
 /** A libtrace output trace

lib/libtrace_parallel.h

@@ -460 +460 @@
  * @param libtrace The input trace to start
  * @param global_blob Global data related to this trace accessible using trace_get_global()
- * @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.
+ * @param per_packet_cbs A set of user supplied functions to be called in response to events being observed by the per_pkt threads.
+ * @param reporter_cbs A set of user supplied functions to be called in response to events / results being seen by the reporter thread.
  * Optional if NULL the reporter thread will not be started.
  * @return 0 on success, otherwise -1 to indicate an error has occurred
@@ -467 +467 @@
  * 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_msg and reporter will only be updated
- * if they are non-null. Otherwise their previous values will be maintained.
+ * In this case global_blob, per_packet_cbs and reporter_cbs 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_cb_msg per_msg, fn_reporter reporter);
+                           libtrace_callback_set_t *per_packet_cbs,
+                           libtrace_callback_set_t *reporter_cbs);
 
 /**
@@ -539 +541 @@
                                            libtrace_packet_t *packet);
 
+/**
+ * Callback for handling a result message. Should only be required by the
+ * reporter thread.
+ *
+ * @param libtrace The parallel trace
+ * @param sender The thread that generated this result
+ * @param global The global storage
+ * @param tls The thread local storage
+ * @param result The result associated with the message
+ *
+ */
+typedef void (*fn_cb_result)(libtrace_t *libtrace, libtrace_thread_t *sender,
+                void *global, void *tls, libtrace_result_t *result);
+
+
+/**
+ * Callback for handling any user-defined message types. This will handle
+ * any messages with a type >= MESSAGE_USER.
+ *
+ * @param libtrace The parallel trace
+ * @param t The thread
+ * @param global The global storage
+ * @param tls The thread local storage
+ * @param mesg The code identifying the message type
+ * @param data The data associated with the message
+ *
+ */
+typedef void (*fn_cb_usermessage) (libtrace_t *libtrace, libtrace_thread_t *t,
+                void *global, void *tls, int mesg, libtrace_generic_t data);
+
 /** Registers a built-in message with a handler.
  * Note we do not include the sending thread as an argument to the reporter.
@@ -548 +580 @@
  */
 
-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);
+DLLEXPORT int trace_set_starting_cb(libtrace_callback_set_t *cbset,
+                fn_cb_starting handler);
+
+DLLEXPORT int trace_set_stopping_cb(libtrace_callback_set_t *cbset,
+                fn_cb_dataless handler);
+
+DLLEXPORT int trace_set_resuming_cb(libtrace_callback_set_t *cbset,
+                fn_cb_dataless handler);
+
+DLLEXPORT int trace_set_pausing_cb(libtrace_callback_set_t *cbset,
+                fn_cb_dataless handler);
+
+DLLEXPORT int trace_set_packet_cb(libtrace_callback_set_t *cbset,
+                fn_cb_packet handler);
+
+DLLEXPORT int trace_set_first_packet_cb(libtrace_callback_set_t *cbset,
+                fn_cb_first_packet handler);
+
+DLLEXPORT int trace_set_result_cb(libtrace_callback_set_t *cbset,
+                fn_cb_result handler);
+
+DLLEXPORT int trace_set_tick_count_cb(libtrace_callback_set_t *cbset,
+                fn_cb_tick handler);
+
+DLLEXPORT int trace_set_tick_interval_cb(libtrace_callback_set_t *cbset,
+                fn_cb_tick handler);
+
+DLLEXPORT int trace_set_user_message_cb(libtrace_callback_set_t *cbset,
+                fn_cb_usermessage handler);
 
 /** Pauses a trace previously started with trace_pstart()

lib/trace.c

@@ -269 +269 @@
         libtrace->perpkt_queue_full = false;
         libtrace->global_blob = NULL;
-        libtrace->per_msg = NULL;
-        libtrace->reporter = NULL;
         libtrace->hasher = NULL;
         libtrace_zero_ocache(&libtrace->packet_freelist);
@@ -288 +286 @@
         ZERO_USER_CONFIG(libtrace->config);
         memset(&libtrace->combiner, 0, sizeof(libtrace->combiner));
-        memset(&libtrace->callbacks, 0, sizeof(libtrace->callbacks));
+        libtrace->perpkt_cbs = NULL;
+        libtrace->reporter_cbs = NULL;
 
         /* Parse the URI to determine what sort of trace we are dealing with */
@@ -393 +392 @@
         libtrace->perpkt_queue_full = false;
         libtrace->global_blob = NULL;
-        libtrace->per_msg = NULL;
-        libtrace->reporter = NULL;
         libtrace->hasher = NULL;
         libtrace_zero_ocache(&libtrace->packet_freelist);
@@ -409 +406 @@
         ZERO_USER_CONFIG(libtrace->config);
         memset(&libtrace->combiner, 0, sizeof(libtrace->combiner));
-        memset(&libtrace->callbacks, 0, sizeof(libtrace->callbacks));
+        libtrace->perpkt_cbs = NULL;
+        libtrace->reporter_cbs = NULL;
         
         for(tmp=formats_list;tmp;tmp=tmp->next) {
@@ -721 +719 @@
                 // This has all of our packets
                 libtrace_ocache_destroy(&libtrace->packet_freelist);
-                if (libtrace->combiner.destroy && libtrace->reporter)
+                if (libtrace->combiner.destroy && libtrace->reporter_cbs)
                         libtrace->combiner.destroy(libtrace, &libtrace->combiner);
                 free(libtrace->perpkt_threads);
                 libtrace->perpkt_threads = NULL;
                 libtrace->perpkt_thread_count = 0;
-        }
+
+        }
+
+        if (libtrace->perpkt_cbs)
+                trace_destroy_callback_set(libtrace->perpkt_cbs);
+        if (libtrace->reporter_cbs)
+                trace_destroy_callback_set(libtrace->reporter_cbs);
+
         
         if (libtrace->event.packet) {

lib/trace_parallel.c

@@ -80 +80 @@
 
 #include "libtrace.h"
-#include "libtrace_int.h"
+#include "libtrace_parallel.h"
 
 #ifdef HAVE_PCAP_BPF_H
@@ -173 +173 @@
 
 /* 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) {
+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) {
+        enum libtrace_messages switchtype;
+        libtrace_callback_set_t *cbs = NULL;
+
+        if (thread == &trace->reporter_thread) {
+                cbs = trace->reporter_cbs;
+        } else {
+                cbs = trace->perpkt_cbs;
+        }
+
+        if (cbs == NULL)
+                return;
+
+        if (type >= MESSAGE_USER)
+                switchtype = MESSAGE_USER;
+        else
+                switchtype = (enum libtrace_messages) type;
+
+        switch (switchtype) {
         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);
+                if (cbs->message_starting)
+                        thread->user_data = (*cbs->message_starting)(trace,
+                                        thread, trace->global_blob);
                 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);
+                if (cbs->message_first_packet)
+                                (*cbs->message_first_packet)(trace, thread,
+                                trace->global_blob, thread->user_data,
+                                data.pkt, 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);
+                if (cbs->message_tick_count)
+                        (*cbs->message_tick_count)(trace, thread,
+                                        trace->global_blob, thread->user_data,
+                                        data.uint64);
                 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);
+                if (cbs->message_tick_interval)
+                        (*cbs->message_tick_interval)(trace, thread,
+                                        trace->global_blob, thread->user_data,
+                                        data.uint64);
                 return;
         case MESSAGE_STOPPING:
-                fn = trace->callbacks.message_stopping;
+                fn = cbs->message_stopping;
                 break;
         case MESSAGE_RESUMING:
-                fn = trace->callbacks.message_resuming;
+                fn = cbs->message_resuming;
                 break;
         case MESSAGE_PAUSING:
-                fn = trace->callbacks.message_pausing;
+                fn = cbs->message_pausing;
                 break;
+        case MESSAGE_USER:
+                if (cbs->message_user)
+                        (*cbs->message_user)(trace, thread, trace->global_blob,
+                                        thread->user_data, type, data);
+                return;
+        case MESSAGE_RESULT:
+                if (cbs->message_result)
+                        (*cbs->message_result)(trace, thread,
+                                        trace->global_blob, thread->user_data,
+                                        data.res);
 
         /* These should be unused */
@@ -215 +243 @@
         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);
+}
+
+DLLEXPORT libtrace_callback_set_t *trace_create_callback_set() {
+        libtrace_callback_set_t *cbset;
+
+        cbset = (libtrace_callback_set_t *)malloc(sizeof(libtrace_callback_set_t));
+        memset(cbset, 0, sizeof(libtrace_callback_set_t));
+        return cbset;
+}
+
+DLLEXPORT void trace_destroy_callback_set(libtrace_callback_set_t *cbset) {
+        free(cbset);
 }
 
@@ -239 +275 @@
 {
         assert(libtrace->state != STATE_NEW);
-        return libtrace->reporter_thread.type == THREAD_REPORTER && libtrace->reporter;
+        return libtrace->reporter_thread.type == THREAD_REPORTER && libtrace->reporter_cbs;
 }
 
@@ -441 +477 @@
                 }
                 t->accepted_packets++;
-                libtrace_generic_t data = {.pkt = *packet};
-                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);
+                if (trace->perpkt_cbs->message_packet)
+                        *packet = (*trace->perpkt_cbs->message_packet)(trace, t, trace->global_blob, t->user_data, *packet);
                 trace_fin_packet(*packet);
         } else {
@@ -624 +657 @@
                                         goto eof;
                         }
-                        (*trace->per_msg)(trace, t, message.code, message.data, message.sender);
-                        (*trace->per_msg)(trace, t, message.code, message.data, message.sender);
+                        send_message(trace, t, message.code, message.data, 
+                                        message.sender);
                         /* Continue and the empty messages out before packets */
                         continue;
@@ -1067 +1100 @@
         }
 
-        (*trace->reporter)(trace, MESSAGE_STARTING, (libtrace_generic_t) {0}, t);
-        (*trace->reporter)(trace, MESSAGE_RESUMING, (libtrace_generic_t) {0}, t);
+        send_message(trace, t, MESSAGE_STARTING, (libtrace_generic_t){0}, t);
+        send_message(trace, t, MESSAGE_RESUMING, (libtrace_generic_t){0}, t);
 
         while (!trace_has_finished(trace)) {
@@ -1085 +1118 @@
                                 assert(trace->combiner.pause);
                                 trace->combiner.pause(trace, &trace->combiner);
-                                (*trace->reporter)(trace, MESSAGE_PAUSING, (libtrace_generic_t) {0}, t);
+                                send_message(trace, t, MESSAGE_PAUSING,
+                                                (libtrace_generic_t) {0}, t);
                                 trace_thread_pause(trace, t);
-                                (*trace->reporter)(trace, MESSAGE_RESUMING, (libtrace_generic_t) {0}, t);
+                                send_message(trace, t, MESSAGE_RESUMING,
+                                                (libtrace_generic_t) {0}, t);
                                 break;
                 default:
-                        (*trace->reporter)(trace, message.code, message.data, message.sender);
+                        send_message(trace, t, message.code, message.data,
+                                        message.sender);
                 }
         }
@@ -1098 +1134 @@
 
         // GOODBYE
-        (*trace->reporter)(trace, MESSAGE_PAUSING, (libtrace_generic_t) {0}, t);
-        (*trace->reporter)(trace, MESSAGE_STOPPING, (libtrace_generic_t) {0}, t);
+        send_message(trace, t, MESSAGE_PAUSING,(libtrace_generic_t) {0}, t);
+        send_message(trace, t, MESSAGE_STOPPING,(libtrace_generic_t) {0}, t);
 
         thread_change_state(trace, &trace->reporter_thread, THREAD_FINISHED, true);
@@ -1315 +1351 @@
  */
 static int trace_prestart(libtrace_t * libtrace, void *global_blob,
-                          fn_cb_msg per_msg, fn_reporter reporter) {
+                          libtrace_callback_set_t *per_packet_cbs, 
+                          libtrace_callback_set_t *reporter_cbs) {
         int i, err = 0;
         if (libtrace->state != STATE_PAUSED) {
@@ -1358 +1395 @@
 
         /* Update functions if requested */
-        if (per_msg)
-                libtrace->per_msg = per_msg;
-        assert(libtrace->per_msg);
-        if (reporter)
-                libtrace->reporter = reporter;
         if(global_blob)
                 libtrace->global_blob = global_blob;
+
+        if (per_packet_cbs) {
+                if (libtrace->perpkt_cbs)
+                        trace_destroy_callback_set(libtrace->perpkt_cbs);
+                libtrace->perpkt_cbs = trace_create_callback_set();
+                memcpy(libtrace->perpkt_cbs, per_packet_cbs, 
+                                sizeof(libtrace_callback_set_t));
+        }
+
+        if (reporter_cbs) {
+                if (libtrace->reporter_cbs)
+                        trace_destroy_callback_set(libtrace->reporter_cbs);
+
+                libtrace->reporter_cbs = trace_create_callback_set();
+                memcpy(libtrace->reporter_cbs, reporter_cbs, 
+                                sizeof(libtrace_callback_set_t));
+        }
 
         if (trace_is_parallel(libtrace)) {
@@ -1586 +1635 @@
 
 DLLEXPORT int trace_pstart(libtrace_t *libtrace, void* global_blob,
-                           fn_cb_msg per_msg, fn_reporter reporter) {
+                           libtrace_callback_set_t *per_packet_cbs,
+                           libtrace_callback_set_t *reporter_cbs) {
         int i;
         int ret = -1;
@@ -1599 +1649 @@
 
         if (libtrace->state == STATE_PAUSED) {
-                ret = trace_prestart(libtrace, global_blob, per_msg, reporter);
+                ret = trace_prestart(libtrace, global_blob, per_packet_cbs, 
+                                reporter_cbs);
                 ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
                 return ret;
@@ -1614 +1665 @@
         /* Store the user defined things against the trace */
         libtrace->global_blob = global_blob;
-        libtrace->per_msg = per_msg;
-        libtrace->reporter = reporter;
+
+        /* Save a copy of the callbacks in case the user tries to change them
+         * on us later */
+        if (!per_packet_cbs) {
+                trace_set_err(libtrace, TRACE_ERR_INIT_FAILED, "trace_pstart "
+                                "requires a non-NULL set of per packet "
+                                "callbacks.");
+                goto cleanup_none;
+        }
+
+        if (per_packet_cbs->message_packet == NULL) {
+                trace_set_err(libtrace, TRACE_ERR_INIT_FAILED, "The per "
+                                "packet callbacks must include a handler "
+                                "for a packet. Please set this using "
+                                "trace_set_packet_cb().");
+                goto cleanup_none;
+        }
+
+        libtrace->perpkt_cbs = trace_create_callback_set();
+        memcpy(libtrace->perpkt_cbs, per_packet_cbs, sizeof(libtrace_callback_set_t));
+        
+        if (reporter_cbs) {
+                libtrace->reporter_cbs = trace_create_callback_set();
+                memcpy(libtrace->reporter_cbs, reporter_cbs, sizeof(libtrace_callback_set_t));
+        }
+
+        
+
+
         /* And zero other fields */
         for (i = 0; i < THREAD_STATE_MAX; ++i) {
@@ -1694 +1772 @@
 
         /* Start the reporter thread */
-        if (reporter) {
+        if (reporter_cbs) {
                 if (libtrace->combiner.initialise)
                         libtrace->combiner.initialise(libtrace, &libtrace->combiner);
@@ -1793 +1871 @@
 }
 
-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;
+DLLEXPORT int trace_set_starting_cb(libtrace_callback_set_t *cbset,
+                fn_cb_starting handler) {
+        cbset->message_starting = handler;
+        return 0;
+}
+
+DLLEXPORT int trace_set_pausing_cb(libtrace_callback_set_t *cbset,
+                fn_cb_dataless handler) {
+        cbset->message_pausing = handler;
+        return 0;
+}
+
+DLLEXPORT int trace_set_resuming_cb(libtrace_callback_set_t *cbset,
+                fn_cb_dataless handler) {
+        cbset->message_resuming = handler;
+        return 0;
+}
+
+DLLEXPORT int trace_set_stopping_cb(libtrace_callback_set_t *cbset,
+                fn_cb_dataless handler) {
+        cbset->message_stopping = handler;
+        return 0;
+}
+
+DLLEXPORT int trace_set_packet_cb(libtrace_callback_set_t *cbset,
+                fn_cb_packet handler) {
+        cbset->message_packet = handler;
+        return 0;
+}
+
+DLLEXPORT int trace_set_first_packet_cb(libtrace_callback_set_t *cbset,
+                fn_cb_first_packet handler) {
+        cbset->message_first_packet = handler;
+        return 0;
+}
+
+DLLEXPORT int trace_set_tick_count_cb(libtrace_callback_set_t *cbset,
+                fn_cb_tick handler) {
+        cbset->message_tick_count = handler;
+        return 0;
+}
+
+DLLEXPORT int trace_set_tick_interval_cb(libtrace_callback_set_t *cbset,
+                fn_cb_tick handler) {
+        cbset->message_tick_interval = handler;
+        return 0;
+}
+
+DLLEXPORT int trace_set_result_cb(libtrace_callback_set_t *cbset,
+                fn_cb_result handler) {
+        cbset->message_result = handler;
+        return 0;
+}
+
+DLLEXPORT int trace_set_user_message_cb(libtrace_callback_set_t *cbset,
+                fn_cb_usermessage handler) {
+        cbset->message_user = handler;
         return 0;
 }

tools/tracertstats/tracertstats.c

@@ -130 +130 @@
 
 static uint64_t glob_last_ts = 0;
-static void process_result(libtrace_t *trace, int mesg,
-                           libtrace_generic_t data,
-                           libtrace_thread_t *sender UNUSED) {
+static void cb_result(libtrace_t *trace, libtrace_thread_t *sender UNUSED,
+                void *global UNUSED, void *tls UNUSED,
+                libtrace_result_t *result) {
         uint64_t ts = 0;
         static bool stopped = false;
@@ -142 +142 @@
                 return;
 
-        switch (mesg) {
-                case MESSAGE_RESULT:
-                ts = data.res->key;
-                res = data.res->value.ptr;
-                if (glob_last_ts == 0)
-                        glob_last_ts = ts;
-                while ((glob_last_ts >> 32) < (ts >> 32)) {
-                        report_results(glob_last_ts >> 32, count, bytes);
-                        count = 0;
-                        bytes = 0;
-                        for (j = 0; j < filter_count; j++)
-                                filters[j].count = filters[j].bytes = 0;
-                        glob_last_ts = ts;
-                }
-                count += res->total.count;
-                packets_seen += res->total.count;
-                bytes += res->total.bytes;
-                for (j = 0; j < filter_count; j++) {
-                        filters[j].count += res->filters[j].count;
-                        filters[j].bytes += res->filters[j].bytes;
-                }
-                free(res);
-        }
+        ts = result->key;
+        res = result->value.ptr;
+        if (glob_last_ts == 0)
+                glob_last_ts = ts;
+        while ((glob_last_ts >> 32) < (ts >> 32)) {
+                report_results(glob_last_ts >> 32, count, bytes);
+                count = 0;
+                bytes = 0;
+                for (j = 0; j < filter_count; j++)
+                        filters[j].count = filters[j].bytes = 0;
+                glob_last_ts = ts;
+        }
+        count += res->total.count;
+        packets_seen += res->total.count;
+        bytes += res->total.bytes;
+        for (j = 0; j < filter_count; j++) {
+                filters[j].count += res->filters[j].count;
+                filters[j].bytes += res->filters[j].bytes;
+        }
+        free(res);
 
         /* Be careful to only call pstop once from within this thread! */
@@ -246 +243 @@
 {
         libtrace_t *trace = NULL;
-        if (!merge_inputs) 
+        libtrace_callback_set_t *pktcbs, *repcbs;
+
+        if (!merge_inputs) 
                 create_output(uri);
 
@@ -268 +267 @@
         }
 
-        trace_cb_starting(trace, cb_starting);
-        trace_cb_stopping(trace, cb_stopping);
-        trace_cb_packet(trace, cb_packet);
-        trace_cb_tick_count(trace, cb_tick);
-        trace_cb_tick_interval(trace, cb_tick);
-
-        if (trace_pstart(trace, NULL, NULL, process_result)==-1) {
+        pktcbs = trace_create_callback_set();
+        trace_set_starting_cb(pktcbs, cb_starting);
+        trace_set_stopping_cb(pktcbs, cb_stopping);
+        trace_set_packet_cb(pktcbs, cb_packet);
+        trace_set_tick_count_cb(pktcbs, cb_tick);
+        trace_set_tick_interval_cb(pktcbs, cb_tick);
+
+        repcbs = trace_create_callback_set();
+        trace_set_result_cb(repcbs, cb_result);
+
+        if (trace_pstart(trace, NULL, pktcbs, repcbs)==-1) {
                 trace_perror(trace,"Failed to start trace");
                 trace_destroy(trace);
+                trace_destroy_callback_set(pktcbs);
+                trace_destroy_callback_set(repcbs);
                 if (!merge_inputs)
                         output_destroy(output);
@@ -292 +297 @@
 
         trace_destroy(trace);
+        trace_destroy_callback_set(pktcbs);
+        trace_destroy_callback_set(repcbs);
 
         if (!merge_inputs)

tools/tracestats/Makefile.am

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