Changeset 0862c20

Timestamp:

08/12/14 15:22:24 (7 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:

957a72a

Parents:

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

Message:

Merge branch 'develop' of ​https://github.com/rsanger/libtrace into develop

Files:

• lib/libtrace.h.in (modified) (8 diffs)
• lib/libtrace_int.h (modified) (4 diffs)
• lib/trace.c (modified) (2 diffs)
• lib/trace_parallel.c (modified) (46 diffs)
• test/Makefile (modified) (1 diff)
• test/do-tests-parallel.sh (modified) (1 diff)
• test/test-format-parallel-hasher.c (modified) (6 diffs)
• test/test-format-parallel-reporter.c (added)
• test/test-format-parallel-singlethreaded-hasher.c (modified) (5 diffs)
• test/test-format-parallel-singlethreaded.c (modified) (5 diffs)
• test/test-format-parallel-stressthreads.c (modified) (5 diffs)
• test/test-format-parallel.c (modified) (5 diffs)
• tools/traceanon/traceanon_parallel.c (modified) (10 diffs)
• tools/tracertstats/tracertstats_parallel.c (modified) (3 diffs)
• tools/tracestats/tracestats_parallel.c (modified) (5 diffs)

lib/libtrace.h.in

@@ -244 +244 @@
         uint64_t key;
         void * value;
-        int is_packet;
+        int type;
 } libtrace_result_t;
+#define RESULT_NORMAL 0
+#define RESULT_PACKET 1
+#define RESULT_TICK   2
 
 typedef struct libtrace_thread_t libtrace_thread_t;
@@ -3196 +3199 @@
 
 typedef void* (*fn_per_pkt)(libtrace_t* trace, libtrace_packet_t *p, libtrace_message_t *m, libtrace_thread_t *thread);
-typedef void* (*fn_reducer)(libtrace_t* trace, void* global_blob);
+typedef void* (*fn_reporter)(libtrace_t* trace, libtrace_result_t *r, libtrace_message_t *m);
 typedef uint64_t (*fn_hasher)(const libtrace_packet_t* packet, void *data);
 
-DLLEXPORT int trace_pstart(libtrace_t *libtrace, void* global_blob, fn_per_pkt per_pkt, fn_reducer reducer);
+DLLEXPORT int trace_pstart(libtrace_t *libtrace, void* global_blob, fn_per_pkt per_pkt, fn_reporter reporter);
 DLLEXPORT int trace_ppause(libtrace_t *libtrace);
 DLLEXPORT int trace_pstop(libtrace_t *libtrace);
@@ -3219 +3222 @@
 
 
-DLLEXPORT void trace_publish_result(libtrace_t *libtrace, uint64_t key, void * value);
-DLLEXPORT void trace_publish_packet(libtrace_t *libtrace, libtrace_packet_t *packet);
+DLLEXPORT void trace_publish_result(libtrace_t *libtrace, libtrace_thread_t *t, uint64_t key, void * value, int type);
 typedef struct libtrace_vector libtrace_vector_t;
 DLLEXPORT int trace_get_results(libtrace_t *libtrace, libtrace_vector_t * results);
 
-DLLEXPORT int trace_post_reduce(libtrace_t *libtrace);
+DLLEXPORT int trace_post_reporter(libtrace_t *libtrace);
 DLLEXPORT int libtrace_thread_get_message_count(libtrace_t * libtrace);
 DLLEXPORT int libtrace_thread_get_message(libtrace_t * libtrace, libtrace_message_t * message);
 DLLEXPORT int libtrace_thread_try_get_message(libtrace_t * libtrace, libtrace_message_t * message);
-DLLEXPORT int trace_send_message_to_reducer(libtrace_t * libtrace, libtrace_message_t * message);
+DLLEXPORT int trace_send_message_to_reporter(libtrace_t * libtrace, libtrace_message_t * message);
 DLLEXPORT int trace_send_message_to_perpkts(libtrace_t * libtrace, libtrace_message_t * message);
 DLLEXPORT int trace_send_message_to_thread(libtrace_t * libtrace, libtrace_thread_t *t, libtrace_message_t * message);
@@ -3246 +3248 @@
          */
         TRACE_OPTION_SET_HASHER,
-        
-        /**
-         * See diagrams, this sets the maximum size of freelist used to 
-         * maintain packets and there memory buffers.
-         * NOTE setting this to less than recommend could cause deadlock a
-         * trace that manages its own packets.
-         * A unblockable error message will be printed.
-         */
-        TRACE_OPTION_SET_PACKET_FREELIST_SIZE,
-        
-        /**
-         * See diagrams, this sets the maximum size of buffers used between
-         * the single hasher thread and the buffer.
-         * NOTE setting this to less than recommend could cause deadlock a
-         * trace that manages its own packets.
-         * A unblockable warning message will be printed to stderr in this case.
-         */
-        TRACE_OPTION_SET_PERPKT_BUFFER_SIZE,
         
         /**
@@ -3282 +3266 @@
           * Libtrace ordered results, results in each queue are ordered by key
           * however my not be sequential, a typically case is packet timestamps
-          * the reducer will receive packets in order - note threasholds
+          * the reporter will receive packets in order - note threasholds
           * will be used such that a empty queue wont break things
           */
@@ -3308 +3292 @@
 
 enum libtrace_messages {
-        MESSAGE_STARTED,
+        MESSAGE_STARTING,
+        MESSAGE_RESUMING,
+        MESSAGE_STOPPING,
+        MESSAGE_PAUSING,
         MESSAGE_DO_PAUSE,
-        MESSAGE_PAUSING,
-        MESSAGE_PAUSED,
         MESSAGE_DO_STOP,
-        MESSAGE_STOPPED,
         MESSAGE_FIRST_PACKET,
         MESSAGE_PERPKT_ENDED,
@@ -3319 +3303 @@
         MESSAGE_PERPKT_PAUSED,
         MESSAGE_PERPKT_EOF,
-        MESSAGE_POST_REDUCE,
+        MESSAGE_POST_REPORTER,
         MESSAGE_POST_RANGE,
         MESSAGE_TICK,
@@ -3386 +3370 @@
 DLLEXPORT libtrace_info_t *trace_get_information(libtrace_t * libtrace);
 
+/**
+ * Tuning the parallel sizes
+ */
+struct user_configuration {
+        // Packet memory cache settings (ocache_init) total
+        /**
+         * See diagrams, this sets the maximum size of freelist used to
+         * maintain packets and their memory buffers.
+         * NOTE setting this to less than recommend could cause deadlock a
+         * trace that manages its own packets.
+         * A unblockable error message will be printed.
+         */
+        size_t packet_global_cache_size;
+        // Per thread
+        size_t packet_thread_cache_size;
+        // Packet count limited
+        bool fixed_packet_count;
+        // Bursts/Batches of packets this size are combined, used in single thread mode
+        size_t burst_size;
+        // Each perpkt thread has a queue leading into the reporter
+        //size_t reporter_queue_size;
+        /** The tick interval - in milliseconds (0) */
+        size_t tick_interval;
+        // The tick interval for file based traces, in number of packets TODO implement this
+        size_t tick_count;
+
+        // The number of per packet threads requested
+        size_t perpkt_threads;
+
+        /**
+         * See diagrams, this sets the maximum size of buffers used between
+         * the single hasher thread and the buffer.
+         * NOTE setting this to less than recommend could cause deadlock a
+         * trace that manages its own packets.
+         * A unblockable warning message will be printed to stderr in this case.
+         */
+        /** The number of packets that can queue per thread from hasher thread */
+        size_t hasher_queue_size;
+
+        // Reporter threashold before results are sent
+        size_t reporter_thold;
+};
+
+#define ZERO_USER_CONFIG(config) memset(&config, 0, sizeof(struct user_configuration));
+
 #ifdef __cplusplus
 } /* extern "C" */

lib/libtrace_int.h

@@ -177 +177 @@
         THREAD_HASHER,
         THREAD_PERPKT,
-        THREAD_REDUCER,
+        THREAD_REPORTER,
         THREAD_KEEPALIVE
 };
@@ -307 +307 @@
         /** For the sliding window hasher implementation */
         pthread_rwlock_t window_lock;
-        /** Set once trace_join has been called */
-        bool joined;
         /** Set to indicate a perpkt's queue is full as such the writing perpkt cannot proceed */
         bool perpkt_queue_full;
         /** Global storage for this trace, shared among all the threads  */
         void* global_blob;
-        /** Requested size of the pkt buffer (currently only used if using dedicated hasher thread) */
-        int packet_freelist_size;
         /** The actual freelist */
         libtrace_ocache_t packet_freelist;
-        /** The number of packets that can queue per thread - XXX consider deadlocks with non malloc()'d packets that need to be released */
-        int perpkt_buffer_size;
-        /** The reducer flags */
-        int reducer_flags;
-        /** The tick interval - in milliseconds (0 or -ve==disabled) */
-        int tick_interval;
+        /** The reporter flags */
+        int reporter_flags;
         /** Used to track the next expected key */
         uint64_t expected_key;
         /** User defined per_pkt function called when a pkt is ready */
         fn_per_pkt per_pkt;
-        /** User defined reducer function entry point XXX not hooked up */
-        fn_reducer reducer;
+        /** User defined reporter function entry point XXX not hooked up */
+        fn_reporter reporter;
         /** The hasher function */
         enum hasher_types hasher_type;
@@ -336 +328 @@
         
         libtrace_thread_t hasher_thread;
-        libtrace_thread_t reducer_thread;
+        libtrace_thread_t reporter_thread;
         libtrace_thread_t keepalive_thread;
         int perpkt_thread_count;
@@ -351 +343 @@
          */
         uint64_t dropped_packets;
-        uint64_t received_packets; 
+        uint64_t received_packets;
+        struct user_configuration config;
 };
 

lib/trace.c

@@ -265 +265 @@
         libtrace->state = STATE_NEW;
         libtrace->perpkt_queue_full = false;
-        libtrace->reducer_flags = 0;
+        libtrace->reporter_flags = 0;
         libtrace->global_blob = NULL;
         libtrace->per_pkt = NULL;
-        libtrace->reducer = NULL;
+        libtrace->reporter = NULL;
         libtrace->hasher = NULL;
-        libtrace->packet_freelist_size = 0;
-        libtrace->perpkt_buffer_size = 0;
         libtrace->expected_key = 0;
         libtrace_zero_ocache(&libtrace->packet_freelist);
         libtrace_zero_thread(&libtrace->hasher_thread);
-        libtrace_zero_thread(&libtrace->reducer_thread);
+        libtrace_zero_thread(&libtrace->reporter_thread);
         libtrace_zero_thread(&libtrace->keepalive_thread);
         libtrace_zero_slidingwindow(&libtrace->sliding_window);
-        libtrace->reducer_thread.type = THREAD_EMPTY;
+        libtrace->reporter_thread.type = THREAD_EMPTY;
         libtrace->perpkt_thread_count = 0;
         libtrace->perpkt_threads = NULL;
         libtrace->tracetime = 0;
-        libtrace->tick_interval = 0;
         libtrace->first_packets.first = 0;
         libtrace->first_packets.count = 0;
         libtrace->first_packets.packets = NULL;
         libtrace->dropped_packets = UINT64_MAX;
+        ZERO_USER_CONFIG(libtrace->config);
 
         /* Parse the URI to determine what sort of trace we are dealing with */
@@ -388 +386 @@
         libtrace->state = STATE_NEW; // TODO MAYBE DEAD
         libtrace->perpkt_queue_full = false;
-        libtrace->reducer_flags = 0;
+        libtrace->reporter_flags = 0;
         libtrace->global_blob = NULL;
         libtrace->per_pkt = NULL;
-        libtrace->reducer = NULL;
+        libtrace->reporter = NULL;
         libtrace->hasher = NULL;
         libtrace->expected_key = 0;
-        libtrace->packet_freelist_size = 0;
-        libtrace->perpkt_buffer_size = 0;
         libtrace_zero_ocache(&libtrace->packet_freelist);
         libtrace_zero_thread(&libtrace->hasher_thread);
-        libtrace_zero_thread(&libtrace->reducer_thread);
+        libtrace_zero_thread(&libtrace->reporter_thread);
         libtrace_zero_thread(&libtrace->keepalive_thread);
         libtrace_zero_slidingwindow(&libtrace->sliding_window);
-        libtrace->reducer_thread.type = THREAD_EMPTY;
+        libtrace->reporter_thread.type = THREAD_EMPTY;
         libtrace->perpkt_thread_count = 0;
         libtrace->perpkt_threads = NULL;
         libtrace->tracetime = 0;
-        libtrace->tick_interval = 0;
+        ZERO_USER_CONFIG(libtrace->config);
         
         for(tmp=formats_list;tmp;tmp=tmp->next) {

lib/trace_parallel.c

@@ -101 +101 @@
 
 
-#define VERBOSE_DEBBUGING 0
+#define VERBOSE_DEBBUGING 1
 
 
@@ -181 +181 @@
 
 /**
+ * True if the trace has dedicated hasher thread otherwise false,
+ * to be used after the trace is running
+ */
+static inline int trace_has_dedicated_reporter(libtrace_t * libtrace)
+{
+        assert(libtrace->state != STATE_NEW);
+        return libtrace->reporter_thread.type == THREAD_REPORTER && libtrace->reporter;
+}
+
+/**
  * Changes a thread's state and broadcasts the condition variable. This
  * should always be done when the lock is held. 
@@ -206 +216 @@
 
 #if VERBOSE_DEBBUGING
-        fprintf(stderr, "Thread %d State changed from %d to %d\n", t->tid,
-                t->state, prev_state);
+        fprintf(stderr, "Thread %d state changed from %d to %d\n", (int) t->tid,
+                prev_state, t->state);
 #endif
         if (need_lock)
@@ -232 +242 @@
 #if VERBOSE_DEBBUGING
         fprintf(stderr, "Trace(%s) state changed from %s to %s\n",
-                trace->uridata, get_trace_state_name(trace->state),
-                get_trace_state_name(prev_state));
+                trace->uridata, get_trace_state_name(prev_state),
+                get_trace_state_name(trace->state));
 #endif
         if (need_lock)
@@ -309 +319 @@
         if (!(ret = get_thread_table(libtrace))) {
                 pthread_t tid = pthread_self();
-                // Check if we are reducer or something else
-                if (pthread_equal(tid, libtrace->reducer_thread.tid))
-                        ret = &libtrace->reducer_thread;
+                // Check if we are reporter or something else
+                if (pthread_equal(tid, libtrace->reporter_thread.tid))
+                        ret = &libtrace->reporter_thread;
                 else if (pthread_equal(tid, libtrace->hasher_thread.tid))
                         ret = &libtrace->hasher_thread;
@@ -324 +334 @@
 {
         libtrace_result_t *res = (libtrace_result_t *)data;
-        if (res->is_packet) {
+        if (res->type == RESULT_PACKET) {
                 // Duplicate the packet in standard malloc'd memory and free the
-                // original
+                // original, This is a 1:1 exchange so is ocache count remains unchanged.
                 libtrace_packet_t *oldpkt, *dup;
                 oldpkt = (libtrace_packet_t *) res->value;
@@ -342 +352 @@
 static void trace_make_results_packets_safe(libtrace_t *trace) {
         libtrace_thread_t *t = get_thread_descriptor(trace);
-        if (trace->reducer_flags & (REDUCE_SEQUENTIAL | REDUCE_ORDERED))
+        if (trace->reporter_flags & (REDUCE_SEQUENTIAL | REDUCE_ORDERED))
                 libtrace_deque_apply_function(&t->deque, &do_copy_result_packet);
         else 
@@ -353 +363 @@
  */
 static void trace_thread_pause(libtrace_t *trace, libtrace_thread_t *t) {
-        trace_make_results_packets_safe(trace);
+        if (t->type == THREAD_PERPKT)
+                trace_make_results_packets_safe(trace);
         ASSERT_RET(pthread_mutex_lock(&trace->libtrace_lock), == 0);
         thread_change_state(trace, t, THREAD_PAUSED, false);
@@ -363 +374 @@
 }
 
-#define PACKETQUEUES 10
 
 /**
@@ -372 +382 @@
         libtrace_thread_t * t;
         libtrace_message_t message = {0};
-        libtrace_packet_t *packets[PACKETQUEUES] = {NULL};
+        libtrace_packet_t *packets[trace->config.burst_size];
         size_t nb_packets;
         size_t i;
 
+        memset(&packets, 0, sizeof(void*) * trace->config.burst_size);
         // Force this thread to wait until trace_pstart has been completed
         ASSERT_RET(pthread_mutex_lock(&trace->libtrace_lock), == 0);
@@ -389 +400 @@
         // Send a message to say we've started
 
-        message.code = MESSAGE_STARTED;
+        // Let the per_packet function know we have started
+        message.code = MESSAGE_STARTING;
         message.sender = t;
-
-        // Let the per_packet function know we have started
+        (*trace->per_pkt)(trace, NULL, &message, t);
+        message.code = MESSAGE_RESUMING;
         (*trace->per_pkt)(trace, NULL, &message, t);
 
@@ -403 +415 @@
                                         // 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 loose data
@@ -409 +422 @@
                                                 // The hasher has stopped by this point, so the queue shouldn't be filling
                                                 while(!libtrace_ringbuffer_is_empty(&t->rbuffer)) {
-                                                        nb_packets = trace_pread_packet(trace, t, packets, 1);
-                                                        if (nb_packets == 1) {
-                                                                if (packets[0]->error > 0)
-                                                                        packets[0] = (*trace->per_pkt)(trace, packets[0], NULL, t);
-                                                        } else {
-                                                                fprintf(stderr, "Psize=%d empty=%d this is probably EOF or message waiting, but if this continues we have a bug!!!\n", packets[0]->error, libtrace_ringbuffer_is_empty(&t->rbuffer));
+                                                        ASSERT_RET(trace_pread_packet(trace, t, packets, 1), == 1);
+                                                        if (packets[0]->error > 0)
+                                                                packets[0] = (*trace->per_pkt)(trace, packets[0], NULL, t);
+                                                        else if (packets[0]->error != -2) {
+                                                                // 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;
                                                         }
                                                 }
                                         }
-                                        // Send a paused message as a final chance to memory copy any packets
-                                        message.code = MESSAGE_PAUSED;
-                                        (*trace->per_pkt)(trace, NULL, &message, t);
                                         // 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
                                         continue;
@@ -440 +457 @@
                         nb_packets = 1;
                 } else {
-                        nb_packets = trace_pread_packet(trace, t, packets, PACKETQUEUES);
+                        nb_packets = trace_pread_packet(trace, t, packets, trace->config.burst_size);
                 }
                 // Loop through the packets we just read
@@ -450 +467 @@
                                 // An error this should be the last packet we read
                                 size_t z;
-                                for (z = i ; z < nb_packets; ++z)
-                                        fprintf(stderr, "i=%d nb_packet=%d err=%d\n", (int) z, (int) nb_packets, packets[i]->error);
-                                assert (i == nb_packets-1);
+                                // We could have an eof or error and a message such as pause
+                                for (z = i ; z < nb_packets; ++z) {
+                                        fprintf(stderr, "i=%d nb_packet=%d err=%d\n", (int) z, (int) nb_packets, packets[z]->error);
+                                        assert (packets[z]->error < 1);
+                                }
                                 goto stop;
                         }
@@ -463 +482 @@
 stop:
         /* ~~~~~~~~~~~~~~ Trace is finished do tear down ~~~~~~~~~~~~~~~~~~~~~ */
+
         // Let the per_packet function know we have stopped
-        message.code = MESSAGE_STOPPED;
-        message.sender = NULL;
+        message.code = MESSAGE_PAUSING;
+        message.sender = t;
+        (*trace->per_pkt)(trace, NULL, &message, t);
+        message.code = MESSAGE_STOPPING;
         message.additional.uint64 = 0;
         (*trace->per_pkt)(trace, NULL, &message, t);
 
         // Free any remaining packets
-        for (i = 0; i < PACKETQUEUES; i++) {
+        for (i = 0; i < trace->config.burst_size; i++) {
                 if (packets[i]) {
                         libtrace_ocache_free(&trace->packet_freelist, (void **) &packets[i], 1, 1);
@@ -483 +505 @@
         message.code = MESSAGE_PERPKT_ENDED;
         message.additional.uint64 = 0;
-        trace_send_message_to_reducer(trace, &message);
+        trace_send_message_to_reporter(trace, &message);
 
         ASSERT_RET(pthread_mutex_lock(&trace->libtrace_lock), == 0);
@@ -566 +588 @@
                         pkt_skipped = 0;
                 } else {
+                        assert(!"Dropping a packet!!");
                         pkt_skipped = 1; // Reuse that packet no one read it
                 }
@@ -573 +596 @@
         for (i = 0; i < trace->perpkt_thread_count; i++) {
                 libtrace_packet_t * bcast;
-                printf("Broadcasting error/EOF now the trace is over\n");
+                fprintf(stderr, "Broadcasting error/EOF now the trace is over\n");
                 if (i == trace->perpkt_thread_count - 1) {
                         bcast = packet;
@@ -582 +605 @@
                 ASSERT_RET(pthread_mutex_lock(&trace->libtrace_lock), == 0);
                 if (trace->perpkt_threads[i].state != THREAD_FINISHED) {
-                        ASSERT_RET(pthread_mutex_unlock(&trace->libtrace_lock), == 0);
                         // Unlock early otherwise we could deadlock
                         libtrace_ringbuffer_write(&trace->perpkt_threads[i].rbuffer, bcast);
+                        ASSERT_RET(pthread_mutex_unlock(&trace->libtrace_lock), == 0);
                 } else {
+                        fprintf(stderr, "SKIPPING THREAD !!!%d!!!/n", (int) i);
                         ASSERT_RET(pthread_mutex_unlock(&trace->libtrace_lock), == 0);
                 }
@@ -596 +620 @@
         message.code = MESSAGE_PERPKT_ENDED;
         message.additional.uint64 = 0;
-        trace_send_message_to_reducer(trace, &message);
+        trace_send_message_to_reporter(trace, &message);
         ASSERT_RET(pthread_mutex_lock(&trace->libtrace_lock), == 0);
         if (trace->format->punregister_thread) {
@@ -1070 +1094 @@
                 libtrace_message_t mesg = {0};
                 mesg.code = MESSAGE_FIRST_PACKET;
-                trace_send_message_to_reducer(libtrace, &mesg);
+                trace_send_message_to_reporter(libtrace, &mesg);
                 t->recorded_first = true;
         }
@@ -1124 +1148 @@
 
 /** Similar to delay_tracetime but send messages to all threads periodically */
+static void* reporter_entry(void *data) {
+        libtrace_message_t message = {0};
+        libtrace_t *trace = (libtrace_t *)data;
+        libtrace_thread_t *t = &trace->reporter_thread;
+        size_t res_size;
+        libtrace_vector_t results;
+        libtrace_vector_init(&results, sizeof(libtrace_result_t));
+        fprintf(stderr, "Reporter thread starting\n");
+        libtrace_result_t result;
+        size_t i;
+
+        message.code = MESSAGE_STARTING;
+        message.sender = t;
+        (*trace->reporter)(trace, NULL, &message);
+        message.code = MESSAGE_RESUMING;
+        (*trace->reporter)(trace, NULL, &message);
+
+        while (!trace_finished(trace)) {
+
+                //while ( != LIBTRACE_MQ_FAILED) { }
+                libtrace_message_queue_get(&t->messages, &message);
+
+                switch (message.code) {
+                        // Check for results
+                        case MESSAGE_POST_REPORTER:
+                                res_size = trace_get_results(trace, &results);
+                                for (i = 0; i < res_size; i++) {
+                                        ASSERT_RET(libtrace_vector_get(&results, i, (void *) &result), == 1);
+                                        (*trace->reporter)(trace, &result, NULL);
+                                }
+                                break;
+                        case MESSAGE_DO_PAUSE:
+                                message.code = MESSAGE_PAUSING;
+                                message.sender = t;
+                                (*trace->reporter)(trace, NULL, &message);
+                                trace_thread_pause(trace, t);
+                                message.code = MESSAGE_RESUMING;
+                                (*trace->reporter)(trace, NULL, &message);
+                                break;
+                        default:
+                                (*trace->reporter)(trace, NULL, &message);
+                }
+        }
+
+        // Flush out whats left now all our threads have finished
+        res_size = trace_get_results(trace, &results);
+        for (i = 0; i < res_size; i++) {
+                ASSERT_RET(libtrace_vector_get(&results, i, (void *) &result), == 1);
+                (*trace->reporter)(trace, &result, NULL);
+        }
+
+        // GOODBYE
+        message.code = MESSAGE_PAUSING;
+        message.sender = t;
+        (*trace->reporter)(trace, NULL, &message);
+        message.code = MESSAGE_STOPPING;
+        (*trace->reporter)(trace, NULL, &message);
+
+        thread_change_state(trace, &trace->reporter_thread, THREAD_FINISHED, true);
+        print_memory_stats();
+        return NULL;
+}
+
+/** Similar to delay_tracetime but send messages to all threads periodically */
 static void* keepalive_entry(void *data) {
         struct timeval prev, next;
@@ -1135 +1223 @@
         while (trace->state != STATE_FINSHED) {
                 fd_set rfds;
-                next_release = tv_to_usec(&prev) + (trace->tick_interval * 1000);
+                next_release = tv_to_usec(&prev) + (trace->config.tick_interval * 1000);
                 gettimeofday(&next, NULL);
                 if (next_release > tv_to_usec(&next)) {
@@ -1329 +1417 @@
  * @returns 0 on success
  */
-DLLEXPORT int trace_pstart(libtrace_t *libtrace, void* global_blob, fn_per_pkt per_pkt, fn_reducer reducer)
+DLLEXPORT int trace_pstart(libtrace_t *libtrace, void* global_blob, fn_per_pkt per_pkt, fn_reporter reporter)
 {
         int i;
@@ -1353 +1441 @@
                 if (per_pkt)
                         libtrace->per_pkt = per_pkt;
+
+                if (reporter)
+                        libtrace->reporter = reporter;
 
                 assert(libtrace_parallel);
@@ -1382 +1473 @@
         libtrace->global_blob = global_blob;
         libtrace->per_pkt = per_pkt;
-        libtrace->reducer = reducer;
+        libtrace->reporter = reporter;
 
         ASSERT_RET(pthread_mutex_init(&libtrace->libtrace_lock, NULL), == 0);
@@ -1391 +1482 @@
 
         // Set default buffer sizes
-        if (libtrace->perpkt_buffer_size <= 0)
-                libtrace->perpkt_buffer_size = 1000;
+        if (libtrace->config.hasher_queue_size <= 0)
+                libtrace->config.hasher_queue_size = 1000;
 
         if (libtrace->perpkt_thread_count <= 0) {
@@ -1402 +1493 @@
         }
 
-        if(libtrace->packet_freelist_size <= 0)
-                libtrace->packet_freelist_size = (libtrace->perpkt_buffer_size + 1) * libtrace->perpkt_thread_count;
-
-        if(libtrace->packet_freelist_size <
-                (libtrace->perpkt_buffer_size + 1) * libtrace->perpkt_thread_count)
+        if (libtrace->config.reporter_thold <= 0)
+                libtrace->config.reporter_thold = 100;
+        if (libtrace->config.burst_size <= 0)
+                libtrace->config.burst_size = 10;
+        if (libtrace->config.packet_thread_cache_size <= 0)
+                libtrace->config.packet_thread_cache_size = 20;
+        if (libtrace->config.packet_global_cache_size <= 0)
+                libtrace->config.packet_global_cache_size = (libtrace->config.hasher_queue_size + 1) * libtrace->perpkt_thread_count;
+
+        if (libtrace->config.packet_global_cache_size <
+                (libtrace->config.hasher_queue_size + 1) * libtrace->perpkt_thread_count)
                 fprintf(stderr, "WARNING deadlocks may occur and extra memory allocating buffer sizes (packet_freelist_size) mismatched\n");
 
@@ -1432 +1529 @@
                 libtrace->hasher_thread.type = THREAD_EMPTY;
         }
-        //libtrace_ocache_init(&libtrace->packet_freelist, trace_create_packet, trace_destroy_packet, 64, libtrace->packet_freelist_size * 4, true);
+
         libtrace_ocache_init(&libtrace->packet_freelist,
                                                  (void* (*)()) trace_create_packet,
                                                  (void (*)(void *))trace_destroy_packet,
-                                                 64,
-                                                 libtrace->packet_freelist_size * 4,
-                                                 true);
+                                                 libtrace->config.packet_thread_cache_size,
+                                                 libtrace->config.packet_global_cache_size * 4,
+                                                 libtrace->config.fixed_packet_count);
+        // Unused slidingwindow code
         //libtrace_slidingwindow_init(&libtrace->sliding_window, libtrace->packet_freelist_size, 0);
-        ASSERT_RET(sem_init(&libtrace->sem, 0, libtrace->packet_freelist_size), == 0);
+        //ASSERT_RET(sem_init(&libtrace->sem, 0, libtrace->packet_freelist_size), == 0);
+
         // This will be applied to every new thread that starts, i.e. they will block all signals
         // Lets start a fixed number of reading threads
-
-        // For now we never have a dedicated thread for the reducer
-        // i.e. This main thread is used as the reducer
-        libtrace->reducer_thread.tid = pthread_self();
-        libtrace->reducer_thread.type = THREAD_REDUCER;
-        libtrace->reducer_thread.state = THREAD_RUNNING;
-        libtrace_message_queue_init(&libtrace->reducer_thread.messages, sizeof(libtrace_message_t));
 
         /* Ready some storages */
@@ -1470 +1562 @@
                 t->perpkt_num = i;
                 if (libtrace->hasher)
-                        libtrace_ringbuffer_init(&t->rbuffer, libtrace->perpkt_buffer_size, LIBTRACE_RINGBUFFER_POLLING);
+                        libtrace_ringbuffer_init(&t->rbuffer, libtrace->config.hasher_queue_size, LIBTRACE_RINGBUFFER_POLLING);
                 // Depending on the mode vector or deque might be chosen
                 libtrace_vector_init(&t->vector, sizeof(libtrace_result_t));
@@ -1492 +1584 @@
                 threads_started = trace_start_perpkt_threads(libtrace);
 
-        if (libtrace->tick_interval > 0) {
+        libtrace->reporter_thread.type = THREAD_REPORTER;
+        libtrace->reporter_thread.state = THREAD_RUNNING;
+        libtrace_message_queue_init(&libtrace->reporter_thread.messages, sizeof(libtrace_message_t));
+        if (reporter) {
+                // Got a real reporter
+                ASSERT_RET(pthread_create(&libtrace->reporter_thread.tid, NULL, reporter_entry, (void *) libtrace), == 0);
+        } else {
+                // Main thread is reporter
+                libtrace->reporter_thread.tid = pthread_self();
+        }
+
+        if (libtrace->config.tick_interval > 0) {
                 libtrace->keepalive_thread.type = THREAD_KEEPALIVE;
                 libtrace->keepalive_thread.state = THREAD_RUNNING;
@@ -1580 +1683 @@
                         fprintf(stderr, "Mapper threads should not be used to pause a trace this could cause any number of problems!!\n");
                 }
+        }
+
+        // Deal with the reporter
+        if (trace_has_dedicated_reporter(libtrace)) {
+                fprintf(stderr, "Reporter thread running we deal with this special!\n");
+                libtrace_message_t message = {0};
+                message.code = MESSAGE_DO_PAUSE;
+                trace_send_message_to_thread(libtrace, &libtrace->reporter_thread, &message);
+                // Wait for it to pause
+                ASSERT_RET(pthread_mutex_lock(&libtrace->libtrace_lock), == 0);
+                while (libtrace->reporter_thread.state == THREAD_RUNNING) {
+                        ASSERT_RET(pthread_cond_wait(&libtrace->perpkt_cond, &libtrace->libtrace_lock), == 0);
+                }
+                ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
         }
 
@@ -1760 +1877 @@
         // buffers so clean them up
         for (i = 0; i < libtrace->perpkt_thread_count; i++) {
-                // Its possible 1 packet got added by the reducer (or 1 per any other thread) since we cleaned up
+                // Its possible 1 packet got added by the reporter (or 1 per any other thread) since we cleaned up
                 // if they lost timeslice before-during a write
                 libtrace_packet_t * packet;
@@ -1773 +1890 @@
         // TODO consider perpkt threads marking trace as finished before join is called
         libtrace_change_state(libtrace, STATE_FINSHED, true);
+
+        if (trace_has_dedicated_reporter(libtrace)) {
+                fprintf(stderr, "Waiting to join with the reporter\n");
+                pthread_join(libtrace->reporter_thread.tid, NULL);
+                fprintf(stderr, "Joined with the reporter\n");
+                assert(libtrace->reporter_thread.state == THREAD_FINISHED);
+        }
         
         // Wait for the tick (keepalive) thread if it has been started
@@ -1812 +1936 @@
  * Return backlog indicator
  */
-DLLEXPORT int trace_post_reduce(libtrace_t *libtrace)
+DLLEXPORT int trace_post_reporter(libtrace_t *libtrace)
 {
         libtrace_message_t message = {0};
-        message.code = MESSAGE_POST_REDUCE;
+        message.code = MESSAGE_POST_REPORTER;
         message.sender = get_thread_descriptor(libtrace);
-        return libtrace_message_queue_put(&libtrace->reducer_thread.messages, (void *) &message);
+        return libtrace_message_queue_put(&libtrace->reporter_thread.messages, (void *) &message);
 }
 
@@ -1823 +1947 @@
  * Return backlog indicator
  */
-DLLEXPORT int trace_send_message_to_reducer(libtrace_t * libtrace, libtrace_message_t * message)
-{
-        //printf("Sending message code=%d to reducer\n", message->code);
+DLLEXPORT int trace_send_message_to_reporter(libtrace_t * libtrace, libtrace_message_t * message)
+{
+        //printf("Sending message code=%d to reporter\n", message->code);
         message->sender = get_thread_descriptor(libtrace);
-        return libtrace_message_queue_put(&libtrace->reducer_thread.messages, message);
+        return libtrace_message_queue_put(&libtrace->reporter_thread.messages, message);
 }
 
@@ -1835 +1959 @@
 DLLEXPORT int trace_send_message_to_thread(libtrace_t * libtrace, libtrace_thread_t *t, libtrace_message_t * message)
 {
-        //printf("Sending message code=%d to reducer\n", message->code);
+        //printf("Sending message code=%d to reporter\n", message->code);
         message->sender = get_thread_descriptor(libtrace);
         return libtrace_message_queue_put(&t->messages, message);
@@ -1847 +1971 @@
                 libtrace_message_queue_put(&libtrace->perpkt_threads[i].messages, message);
         }
-        //printf("Sending message code=%d to reducer\n", message->code);
+        //printf("Sending message code=%d to reporter\n", message->code);
         return 0;
 }
@@ -1909 +2033 @@
 /**
  * Publish to the reduce queue, return
- */
-DLLEXPORT void trace_publish_result(libtrace_t *libtrace, uint64_t key, void * value) {
+ * Should only be called by a perpkt thread, i.e. from a perpkt handler
+ */
+DLLEXPORT void trace_publish_result(libtrace_t *libtrace, libtrace_thread_t *t, uint64_t key, void * value, int type) {
         libtrace_result_t res;
-        res.is_packet = 0;
-        // Who am I???
-        int this_thread = get_thread_table_num(libtrace); // Could be worth caching ... ?
-        libtrace_thread_t * t = &libtrace->perpkt_threads[this_thread];
-        // Now put it into my table
         UNUSED static __thread int count = 0;
-
+        res.type = type;
 
         libtrace_result_set_key_value(&res, key, value);
@@ -1930 +2050 @@
                 printf("My vector size is %d\n", libtrace_deque_get_size(&t->deque));
         count = (count+1)%1000;*/
-        if (libtrace->reducer_flags & (REDUCE_SEQUENTIAL | REDUCE_ORDERED)) {
-                if (libtrace_deque_get_size(&t->deque) >= 800) {
-                        trace_post_reduce(libtrace);
+        if (libtrace->reporter_flags & (REDUCE_SEQUENTIAL | REDUCE_ORDERED)) {
+                if (libtrace_deque_get_size(&t->deque) >= libtrace->config.reporter_thold) {
+                        trace_post_reporter(libtrace);
                 }
                 //while (libtrace_deque_get_size(&t->deque) >= 1000)
@@ -1941 +2061 @@
                 //      sched_yield();
 
-                if (libtrace_vector_get_size(&t->vector) >= 800) {
-                        trace_post_reduce(libtrace);
+                if (libtrace_vector_get_size(&t->vector) >= libtrace->config.reporter_thold) {
+                        trace_post_reporter(libtrace);
                 }
                 libtrace_vector_push_back(&t->vector, &res); // Automatically locking for us :)
         }
 }
-
-DLLEXPORT void trace_publish_packet(libtrace_t *libtrace, libtrace_packet_t *packet) {
-        libtrace_result_t res;
-        // Who am I???
-        int this_thread = get_thread_table_num(libtrace); // Could be worth caching ... ?
-        libtrace_thread_t * t = &libtrace->perpkt_threads[this_thread];
-        // Now put it into my table
-        UNUSED static __thread int count = 0;
-
-        res.is_packet = 1;
-        libtrace_result_set_key_value(&res, trace_packet_get_order(packet), packet);
-        /*
-        if (count == 1)
-                printf("My vector size is %d\n", libtrace_vector_get_size(&t->vector));
-        count = (count+1) %1000;
-        libtrace_vector_push_back(&t->vector, &res); // Automatically locking for us :)
-        */
-        /*if (count == 1)
-                printf("My vector size is %d\n", libtrace_deque_get_size(&t->deque));
-        count = (count+1)%1000;*/
-        if (libtrace->reducer_flags & (REDUCE_SEQUENTIAL | REDUCE_ORDERED)) {
-                if (libtrace_deque_get_size(&t->deque) >= 800) {
-                        trace_post_reduce(libtrace);
-                }
-                //while (libtrace_deque_get_size(&t->deque) >= 1000)
-                //      sched_yield();
-                libtrace_deque_push_back(&t->deque, &res); // Automatically locking for us :)
-        } else {
-                //while (libtrace_vector_get_size(&t->vector) >= 1000)
-                //      sched_yield();
-
-                if (libtrace_vector_get_size(&t->vector) >= 800) {
-                        trace_post_reduce(libtrace);
-                }
-                libtrace_vector_push_back(&t->vector, &res); // Automatically locking for us :)
-        }
-}
-
 
 static int compareres(const void* p1, const void* p2)
@@ -1998 +2080 @@
 DLLEXPORT int trace_get_results(libtrace_t *libtrace, libtrace_vector_t * results) {
         int i;
-        int flags = libtrace->reducer_flags; // Hint these aren't a changing
+        int flags = libtrace->reporter_flags; // Hint these aren't a changing
 
         libtrace_vector_empty(results);
@@ -2105 +2187 @@
 DLLEXPORT int trace_finished(libtrace_t * libtrace) {
         // TODO I don't like using this so much, we could use state!!!
-        return !(libtrace->perpkt_thread_states[THREAD_RUNNING] || libtrace->perpkt_thread_states[THREAD_FINISHING]);
+        return libtrace->perpkt_thread_states[THREAD_FINISHED] == libtrace->perpkt_thread_count;
 }
 
@@ -2113 +2195 @@
         switch (option) {
                 case TRACE_OPTION_TICK_INTERVAL:
-                        libtrace->tick_interval = *((int *) value);
+                        libtrace->config.tick_interval = *((int *) value);
                         return 1;
                 case TRACE_OPTION_SET_HASHER:
                         return trace_set_hasher(libtrace, (enum hasher_types) *((int *) value), NULL, NULL);
-                case TRACE_OPTION_SET_PERPKT_BUFFER_SIZE:
-                        libtrace->perpkt_buffer_size = *((int *) value);
-                        return 1;
-                case TRACE_OPTION_SET_PACKET_FREELIST_SIZE:
-                        libtrace->packet_freelist_size = *((int *) value);
-                        return 1;
                 case TRACE_OPTION_SET_PERPKT_THREAD_COUNT:
                         libtrace->perpkt_thread_count = *((int *) value);
@@ -2128 +2204 @@
                 case TRACE_DROP_OUT_OF_ORDER:
                         if (*((int *) value))
-                                libtrace->reducer_flags |= REDUCE_DROP_OOO;
+                                libtrace->reporter_flags |= REDUCE_DROP_OOO;
                         else
-                                libtrace->reducer_flags &= ~REDUCE_DROP_OOO;
+                                libtrace->reporter_flags &= ~REDUCE_DROP_OOO;
                         return 1;
                 case TRACE_OPTION_SEQUENTIAL:
                         if (*((int *) value))
-                                libtrace->reducer_flags |= REDUCE_SEQUENTIAL;
+                                libtrace->reporter_flags |= REDUCE_SEQUENTIAL;
                         else
-                                libtrace->reducer_flags &= ~REDUCE_SEQUENTIAL;
+                                libtrace->reporter_flags &= ~REDUCE_SEQUENTIAL;
                         return 1;
                 case TRACE_OPTION_ORDERED:
                         if (*((int *) value))
-                                libtrace->reducer_flags |= REDUCE_ORDERED;
+                                libtrace->reporter_flags |= REDUCE_ORDERED;
                         else
-                                libtrace->reducer_flags &= ~REDUCE_ORDERED;
+                                libtrace->reporter_flags &= ~REDUCE_ORDERED;
                         return 1;
                 case TRACE_OPTION_TRACETIME:

test/Makefile

@@ -15 +15 @@
 BINS_PARALLEL = test-format-parallel test-format-parallel-hasher \
         test-format-parallel-singlethreaded test-format-parallel-stressthreads \
-        test-format-parallel-singlethreaded-hasher
+        test-format-parallel-singlethreaded-hasher test-format-parallel-reporter
 
 BINS = test-pcap-bpf test-event test-time test-dir test-wireless test-errors \

test/do-tests-parallel.sh

@@ -57 +57 @@
 do_test ./test-format-parallel-stressthreads erf
 
+echo \* Read stress testing with 100 threads
+do_test ./test-format-parallel-reporter erf
+
 echo
 echo "Tests passed: $OK"

test/test-format-parallel-hasher.c

@@ -95 +95 @@
         bool seen_start_message;
         bool seen_stop_message;
-        bool seen_paused_message;
+        bool seen_resumed_message;
         bool seen_pausing_message;
         int count;
@@ -125 +125 @@
         }
         else switch (mesg->code) {
-                case MESSAGE_STARTED:
+                case MESSAGE_STARTING:
                         assert(tls == NULL);
                         tls = calloc(sizeof(struct TLS), 1);
@@ -132 +132 @@
                         tls->seen_start_message = true;
                         break;
-                case MESSAGE_STOPPED:
+                case MESSAGE_STOPPING:
                         assert(tls->seen_start_message);
                         assert(tls != NULL);
@@ -139 +139 @@
 
                         // All threads publish to verify the thread count
-                        trace_publish_result(trace, (uint64_t) 0, (void *) tls->count);
-                        trace_post_reduce(trace);
+                        trace_publish_result(trace, t, (uint64_t) 0, (void *) tls->count, RESULT_NORMAL);
+                        trace_post_reporter(trace);
                         free(tls);
                         break;
@@ -152 +152 @@
                         tls->seen_pausing_message = true;
                         break;
-                case MESSAGE_PAUSED:
-                        assert(tls->seen_pausing_message);
-                        tls->seen_paused_message = true;
+                case MESSAGE_RESUMING:
+                        assert(tls->seen_pausing_message  || tls->seen_start_message);
+                        tls->seen_resumed_message = true;
                         break;
         }
@@ -197 +197 @@
         trace_pstart(trace, NULL, per_packet, NULL);
         iferr(trace,tracename);
-
         /* Make sure traces survive a pause and restart */
         trace_ppause(trace);

test/test-format-parallel-singlethreaded-hasher.c

@@ -95 +95 @@
         bool seen_start_message;
         bool seen_stop_message;
-        bool seen_paused_message;
+        bool seen_resuming_message;
         bool seen_pausing_message;
         int count;
@@ -125 +125 @@
         }
         else switch (mesg->code) {
-                case MESSAGE_STARTED:
+                case MESSAGE_STARTING:
                         assert(tls == NULL);
                         tls = calloc(sizeof(struct TLS), 1);
@@ -132 +132 @@
                         tls->seen_start_message = true;
                         break;
-                case MESSAGE_STOPPED:
+                case MESSAGE_STOPPING:
                         assert(tls->seen_start_message);
                         assert(tls != NULL);
@@ -139 +139 @@
 
                         // All threads publish to verify the thread count
-                        trace_publish_result(trace, (uint64_t) 0, (void *) tls->count);
-                        trace_post_reduce(trace);
+                        trace_publish_result(trace, t, (uint64_t) 0, (void *) tls->count, RESULT_NORMAL);
+                        trace_post_reporter(trace);
                         free(tls);
                         break;
@@ -152 +152 @@
                         tls->seen_pausing_message = true;
                         break;
-                case MESSAGE_PAUSED:
-                        assert(tls->seen_pausing_message);
-                        tls->seen_paused_message = true;
+                case MESSAGE_RESUMING:
+                        assert(tls->seen_pausing_message || tls->seen_start_message);
+                        tls->seen_resuming_message = true;
                         break;
         }

test/test-format-parallel-singlethreaded.c

@@ -95 +95 @@
         bool seen_start_message;
         bool seen_stop_message;
-        bool seen_paused_message;
+        bool seen_resuming_message;
         bool seen_pausing_message;
         int count;
@@ -125 +125 @@
         }
         else switch (mesg->code) {
-                case MESSAGE_STARTED:
+                case MESSAGE_STARTING:
                         assert(tls == NULL);
                         tls = calloc(sizeof(struct TLS), 1);
@@ -132 +132 @@
                         tls->seen_start_message = true;
                         break;
-                case MESSAGE_STOPPED:
+                case MESSAGE_STOPPING:
                         assert(tls->seen_start_message);
                         assert(tls != NULL);
@@ -139 +139 @@
 
                         // All threads publish to verify the thread count
-                        trace_publish_result(trace, (uint64_t) 0, (void *) tls->count);
-                        trace_post_reduce(trace);
+                        trace_publish_result(trace, t, (uint64_t) 0, (void *) tls->count, RESULT_NORMAL);
+                        trace_post_reporter(trace);
                         free(tls);
                         break;
@@ -152 +152 @@
                         tls->seen_pausing_message = true;
                         break;
-                case MESSAGE_PAUSED:
-                        assert(tls->seen_pausing_message);
-                        tls->seen_paused_message = true;
+                case MESSAGE_RESUMING:
+                        assert(tls->seen_pausing_message || tls->seen_start_message);
+                        tls->seen_resuming_message = true;
                         break;
         }

test/test-format-parallel-stressthreads.c

@@ -95 +95 @@
         bool seen_start_message;
         bool seen_stop_message;
-        bool seen_paused_message;
+        bool seen_resuming_message;
         bool seen_pausing_message;
         int count;
@@ -125 +125 @@
         }
         else switch (mesg->code) {
-                case MESSAGE_STARTED:
+                case MESSAGE_STARTING:
                         assert(tls == NULL);
                         tls = calloc(sizeof(struct TLS), 1);
@@ -132 +132 @@
                         tls->seen_start_message = true;
                         break;
-                case MESSAGE_STOPPED:
+                case MESSAGE_STOPPING:
                         assert(tls->seen_start_message);
                         assert(tls != NULL);
@@ -139 +139 @@
 
                         // All threads publish to verify the thread count
-                        trace_publish_result(trace, (uint64_t) 0, (void *) tls->count);
-                        trace_post_reduce(trace);
+                        trace_publish_result(trace, t, (uint64_t) 0, (void *) tls->count, RESULT_NORMAL);
+                        trace_post_reporter(trace);
                         free(tls);
                         break;
@@ -152 +152 @@
                         tls->seen_pausing_message = true;
                         break;
-                case MESSAGE_PAUSED:
-                        assert(tls->seen_pausing_message);
-                        tls->seen_paused_message = true;
+                case MESSAGE_RESUMING:
+                        assert(tls->seen_pausing_message || tls->seen_start_message);
+                        tls->seen_resuming_message = true;
                         break;
         }

test/test-format-parallel.c

@@ -95 +95 @@
         bool seen_start_message;
         bool seen_stop_message;
-        bool seen_paused_message;
+        bool seen_resuming_message;
         bool seen_pausing_message;
         int count;
@@ -132 +132 @@
         }
         else switch (mesg->code) {
-                case MESSAGE_STARTED:
+                case MESSAGE_STARTING:
                         assert(!seen_start_message || seen_paused_message);
                         assert(tls == NULL);
@@ -141 +141 @@
                         tls->seen_start_message = true;
                         break;
-                case MESSAGE_STOPPED:
+                case MESSAGE_STOPPING:
                         assert(seen_start_message);
                         assert(tls != NULL);
@@ -152 +152 @@
 
                         // All threads publish to verify the thread count
-                        trace_publish_result(trace, (uint64_t) 0, (void *) count);
-                        trace_post_reduce(trace);
+                        trace_publish_result(trace, t, (uint64_t) 0, (void *) count, RESULT_NORMAL);
+                        trace_post_reporter(trace);
                         break;
                 case MESSAGE_TICK:
@@ -165 +165 @@
                         tls->seen_pausing_message = true;
                         break;
-                case MESSAGE_PAUSED:
-                        assert(seen_pausing_message);
+                case MESSAGE_RESUMING:
+                        assert(tls->seen_pausing_message  || tls->seen_start_message );
                         seen_paused_message = true;
-                        tls->seen_paused_message = true;
+                        tls->seen_resuming_message = true;
                         break;
         }

tools/traceanon/traceanon_parallel.c

@@ -27 +27 @@
         static int s = 0;
         (void)signal;
-        // trace_interrupt();
+    //trace_interrupt();
         // trace_pstop isn't really signal safe because its got lots of locks in it
-        // trace_pstop(trace);
-        if (s == 0) {
+    trace_pstop(trace);
+    /*if (s == 0) {
                 if (trace_ppause(trace) == -1)
                         trace_perror(trace, "Pause failed");
@@ -37 +37 @@
                 if (trace_pstart(trace, NULL, NULL, NULL) == -1)
                         trace_perror(trace, "Start failed");
-        }
+    }*/
         s = !s;
 }
@@ -141 +141 @@
 
 
-static uint64_t bad_hash(libtrace_packet_t * pkt)
+UNUSED static uint64_t bad_hash(UNUSED libtrace_packet_t * pkt)
 {
         return 0;
@@ -147 +147 @@
 
 
-static uint64_t rand_hash(libtrace_packet_t * pkt)
+UNUSED static uint64_t rand_hash(UNUSED libtrace_packet_t * pkt)
 {
         return rand();
@@ -153 +153 @@
 
 
-static void* per_packet(libtrace_t *trace, libtrace_packet_t *pkt, libtrace_message_t *mesg, libtrace_thread_t *t)
-{
-        int i;
+static void* per_packet(libtrace_t *trace, libtrace_packet_t *pkt, libtrace_message_t *mesg, UNUSED libtrace_thread_t *t)
+{
         
         if (pkt) {
@@ -191 +190 @@
                 //libtrace_packet_t * packet_copy = trace_result_packet(trace, pkt);
                 //trace_publish_result(trace, trace_packet_get_order(pkt), pkt);
-                trace_publish_packet(trace, pkt);
+                trace_publish_result(trace, t, trace_packet_get_order(pkt), pkt, RESULT_PACKET);
                 //return ;
         }
@@ -197 +196 @@
                 // printf ("%d.%06d READ #%"PRIu64"\n", tv.tv_sec, tv.tv_usec, trace_packet_get(packet));
                 switch (mesg->code) {
-                        case MESSAGE_STARTED:
+                        case MESSAGE_STARTING:
                                 enc_init(enc_type,key);
                 }
@@ -204 +203 @@
 }
 
+struct libtrace_out_t *writer = 0;
+
+static void* write_out(libtrace_t *trace, libtrace_result_t *result, UNUSED libtrace_message_t *mesg) {
+        static uint64_t packet_count = 0; // TESTING PURPOSES, this is not going to work with a live format
+        if (result) {
+                libtrace_packet_t *packet = (libtrace_packet_t*) libtrace_result_get_value(result);
+                assert(libtrace_result_get_key(result) == packet_count++);
+                if (trace_write_packet(writer,packet)==-1) {
+                        trace_perror_output(writer,"writer");
+                        trace_interrupt();
+                }
+                trace_free_result_packet(trace, packet);
+        }
+        return NULL;
+}
+
+
 int main(int argc, char *argv[]) 
 {
         //struct libtrace_t *trace = 0;
-        struct libtrace_packet_t *packet/* = trace_create_packet()*/;
-        struct libtrace_out_t *writer = 0;
         struct sigaction sigact;
         char *output = 0;
@@ -380 +394 @@
         //trace_parallel_config(trace, TRACE_OPTION_SET_PERPKT_BUFFER_SIZE, &i);
         i = 2;
-        trace_parallel_config(trace, TRACE_OPTION_SET_PERPKT_THREAD_COUNT, &i);
-        
-        if (trace_pstart(trace, NULL, &per_packet, NULL)==-1) {
+    trace_parallel_config(trace, TRACE_OPTION_SET_PERPKT_THREAD_COUNT, &i);
+        
+        if (trace_pstart(trace, NULL, &per_packet, &write_out)==-1) {
                 trace_perror(trace,"trace_start");
                 trace_destroy_output(writer);
@@ -402 +416 @@
         sigaction(SIGINT, &sigact, NULL);
         sigaction(SIGTERM, &sigact, NULL);
-        
-        // Read in the resulting packets and then free them when done
-        libtrace_vector_t res;
-        int res_size = 0;
-        libtrace_vector_init(&res, sizeof(libtrace_result_t));
-        uint64_t packet_count = 0;
-        while (!trace_finished(trace)) {
-                // Read messages
-                libtrace_message_t message;
-                
-                // We just release and do work currently, maybe if something
-                // interesting comes through we'd deal with that
-                libtrace_thread_get_message(trace, &message);
-                
-                while (libtrace_thread_try_get_message(trace, &message) != LIBTRACE_MQ_FAILED) { }
-                
-                if ((res_size = trace_get_results(trace, &res)) == 0)
-                        ;/*sched_yield();*/
-                
-                for (i = 0 ; i < res_size ; i++) {
-                        libtrace_result_t result;
-                        assert(libtrace_vector_get(&res, i, (void *) &result) == 1);
-                        packet = libtrace_result_get_value(&result);
-                        assert(libtrace_result_get_key(&result) == packet_count);
-                        packet_count++;
-                        if (trace_write_packet(writer,packet)==-1) {
-                                trace_perror_output(writer,"writer");
-                                trace_interrupt();
-                                break;
-                        }
-                        //trace_destroy_packet(packet);
-                        trace_free_result_packet(trace, packet);
-                }
-        }
+
+        // Wait for the trace to finish
         trace_join(trace);
-        
-        // Grab everything that's left here
-        res_size = trace_get_results(trace, &res);
-        
-        for (i = 0 ; i < res_size ; i++) {
-                libtrace_result_t result;
-                assert(libtrace_vector_get(&res, i, (void *) &result) == 1);
-                packet = libtrace_result_get_value(&result);
-                if (libtrace_result_get_key(&result) != packet_count)
-                        printf ("Got a %"PRIu64" but expected a %"PRIu64" %d\n", libtrace_result_get_key(&result), packet_count, res_size);
-                assert(libtrace_result_get_key(&result) == packet_count);
-                
-                packet_count++;
-                if (trace_write_packet(writer,packet)==-1) {
-                        trace_perror_output(writer,"writer");
-                        trace_interrupt();
-                        break;
-                }
-                trace_destroy_packet(packet);
-        }
-        libtrace_vector_destroy(&res);
         
         //trace_destroy_packet(packet);

tools/tracertstats/tracertstats_parallel.c

@@ -209 +209 @@
                         // Publish and make a new one new
                         //fprintf(stderr, "Publishing result %"PRIu64"\n", last_ts);
-                        trace_publish_result(trace, (uint64_t) last_ts, results);
-                        trace_post_reduce(trace);
+                        trace_publish_result(trace, t, (uint64_t) last_ts, results, RESULT_NORMAL);
+                        trace_post_reporter(trace);
                         results = calloc(1, sizeof(result_t) + sizeof(statistic_t) * filter_count);
                         last_ts++;
@@ -234 +234 @@
                 // printf ("%d.%06d READ #%"PRIu64"\n", tv.tv_sec, tv.tv_usec, trace_packet_get(packet));
                 switch (mesg->code) {
-                        case MESSAGE_STARTED:
+                        case MESSAGE_STARTING:
                                 results = calloc(1, sizeof(result_t) + sizeof(statistic_t) * filter_count);
                                 break;
-                        case MESSAGE_STOPPED:
+                        case MESSAGE_STOPPING:
                                 // Should we always post this?
                                 if (results->total.count) {
-                                        trace_publish_result(trace, (uint64_t) last_ts, results);
-                                        trace_post_reduce(trace);
+                                        trace_publish_result(trace, t, (uint64_t) last_ts, results, RESULT_NORMAL);
+                                        trace_post_reporter(trace);
                                         results = NULL;
                                 }
@@ -260 +260 @@
                                         if (next_update_time <= mesg->additional.uint64) {
                                                 //fprintf(stderr, "Got a tick and publishing early!!\n");
-                                                trace_publish_result(trace, (uint64_t) last_ts, results);
-                                                trace_post_reduce(trace);
+                                                trace_publish_result(trace, t, (uint64_t) last_ts, results, RESULT_NORMAL);
+                                                trace_post_reporter(trace);
                                                 results = calloc(1, sizeof(result_t) + sizeof(statistic_t) * filter_count);
                                                 last_ts++;

tools/tracestats/tracestats_parallel.c

@@ -67 +67 @@
         // trace_interrupt();
         // trace_pstop isn't really signal safe because its got lots of locks in it
-        // trace_pstop(trace);
-        if (s == 0) {
+    trace_pstop(trace);
+    /*if (s == 0) {
                 if (trace_ppause(trace) == -1)
                         trace_perror(trace, "Pause failed");
@@ -75 +75 @@
                 if (trace_pstart(trace, NULL, NULL, NULL) == -1)
                         trace_perror(trace, "Start failed");
-        }
+    }*/
         s = !s;
 }
@@ -129 +129 @@
                 // printf ("%d.%06d READ #%"PRIu64"\n", tv.tv_sec, tv.tv_usec, trace_packet_get(packet));
                 switch (mesg->code) {
-                        case MESSAGE_STOPPED:
-                                trace_publish_result(trace, 0, results); // Only ever using a single key 0
+                        case MESSAGE_STOPPING:
+                                trace_publish_result(trace, t, 0, results, RESULT_NORMAL); // Only ever using a single key 0
                                 fprintf(stderr, "Thread published resuslts WOWW\n");
                                 break;
-                        case MESSAGE_STARTED:
+                        case MESSAGE_STARTING:
                                 results = calloc(1, sizeof(statistics_t) * (filter_count + 1));
                                 break;
@@ -142 +142 @@
                                 fprintf(stderr, "Thread is pausing\n");
                                 break;
-                        case MESSAGE_PAUSED:
+                        case MESSAGE_RESUMING:
                                 fprintf(stderr, "Thread has paused\n");
                                 break;
@@ -228 +228 @@
         int option = 2;
         //option = 10000;
-        trace_set_hasher(trace, HASHER_CUSTOM, &rand_hash, NULL);
+    //trace_set_hasher(trace, HASHER_CUSTOM, &rand_hash, NULL);
         option = 2;
         trace_parallel_config(trace, TRACE_OPTION_SET_PERPKT_THREAD_COUNT, &option);