Changeset 5b4d121

Timestamp:

08/19/14 13:59:33 (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:

f0e8bd6

Parents:

957a72a

Message:

Adds a configuration parser to make it easy to change the parallel configuration.
Adds more configuration options (Tidies some verbose debugging output).
Implements tick packets for the hasher thread case.
Some other minor bug fixes

Files:

• lib/libtrace.h.in (modified) (3 diffs)
• lib/trace.c (modified) (1 diff)
• lib/trace_parallel.c (modified) (31 diffs)
• tools/traceanon/traceanon_parallel.c (modified) (6 diffs)

lib/libtrace.h.in

@@ -3288 +3288 @@
            * or less this is ignored.
            */
-          TRACE_OPTION_TICK_INTERVAL
+          TRACE_OPTION_TICK_INTERVAL,
+        TRACE_OPTION_GET_CONFIG,
+        TRACE_OPTION_SET_CONFIG
 } trace_parallel_option_t;
 
@@ -3382 +3384 @@
          * A unblockable error message will be printed.
          */
-        size_t packet_global_cache_size;
-        // Per thread
+        size_t packet_cache_size;
+        /**
+         * Per thread local cache size for the packet freelist
+         */
         size_t packet_thread_cache_size;
-        // Packet count limited
+        /**
+         * If true the total number of packets that can be created by a trace is limited
+         * to the packet_cache_size, otherwise once packet_cache_size is exceeded alloc
+         * and free will be used to create and free packets, this will be slower than
+         * using the freelist and could run a machine out of memory.
+         *
+         * However this does make it easier to ensure that deadlocks will not occur
+         * due to running out of packets
+         */
         bool fixed_packet_count;
-        // Bursts/Batches of packets this size are combined, used in single thread mode
+        /**
+         * When reading from a single threaded input source to reduce
+         * lock contention a 'burst' of packets is read per pkt thread
+         * this determines the bursts size.
+         */
         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) */
+
+        /**
+         * The tick interval - in milliseconds
+         * When a live trace is used messages are sent at the tick
+         * interval to ensure that all perpkt threads receive data
+         * this allows results to be printed in cases flows are
+         * not being directed to a certian thread, while still
+         * maintaining order.
+         */
         size_t tick_interval;
-        // The tick interval for file based traces, in number of packets TODO implement this
+
+        /**
+         * Like the tick interval but used in the case of file format
+         * This specifies the number of packets before inserting a tick to
+         * every thread.
+         */
         size_t tick_count;
 
-        // The number of per packet threads requested
+        /**
+         * The number of per packet threads requested, 0 means use default.
+         * Default typically be the number of processor threads detected less one or two.
+         */
         size_t perpkt_threads;
 
@@ -3409 +3441 @@
         size_t hasher_queue_size;
 
-        // Reporter threashold before results are sent
+        /**
+         * If true use a polling hasher queue, that means that we will spin/or yeild
+         * when rather than blocking on a lock. This applies to both the hasher thread
+         * and perpkts reading the queues.
+         */
+        bool hasher_polling;
+
+        /**
+         * If true the reporter thread will continuously poll waiting for results
+         * if false they are only checked when a message is received, this message
+         * is controlled by reporter_thold.
+         */
+        bool reporter_polling;
+
+        /**
+         * Perpkt thread result queue size before triggering the reporter step to read results
+         */
         size_t reporter_thold;
+
+        /**
+         * Prints a line to standard error for every state change
+         * for both the trace as a whole and for each thread.
+         */
+        bool debug_state;
 };
+#include <stdio.h>
+DLLEXPORT void parse_user_config(struct user_configuration* uc, char * str);
+DLLEXPORT void parse_user_config_file(struct user_configuration* uc, FILE *file);
+
+#define READ_EOF 0
+#define READ_ERROR -1
+#define READ_MESSAGE -2
+// Used for inband tick message
+#define READ_TICK -3
 
 #define ZERO_USER_CONFIG(config) memset(&config, 0, sizeof(struct user_configuration));

lib/trace.c

@@ -919 +919 @@
 DLLEXPORT int trace_write_packet(libtrace_out_t *libtrace, libtrace_packet_t *packet) {
         assert(libtrace);
-        assert(packet); 
+        assert(packet);
         /* Verify the packet is valid */
         if (!libtrace->started) {

lib/trace_parallel.c

@@ -101 +101 @@
 
 
-#define VERBOSE_DEBBUGING 1
-
-
 static size_t trace_pread_packet(libtrace_t *libtrace, libtrace_thread_t *t, libtrace_packet_t *packets[], size_t nb_packets);
 
@@ -215 +212 @@
         }
 
-#if VERBOSE_DEBBUGING
-        fprintf(stderr, "Thread %d state changed from %d to %d\n", (int) t->tid,
-                prev_state, t->state);
-#endif
+        if (trace->config.debug_state)
+                fprintf(stderr, "Thread %d state changed from %d to %d\n", (int) t->tid,
+                        prev_state, t->state);
+
         if (need_lock)
                 pthread_mutex_unlock(&trace->libtrace_lock);
@@ -240 +237 @@
         prev_state = trace->state;
         trace->state = new_state;
-#if VERBOSE_DEBBUGING
-        fprintf(stderr, "Trace(%s) state changed from %s to %s\n",
-                trace->uridata, get_trace_state_name(prev_state),
-                get_trace_state_name(trace->state));
-#endif
+
+        if (trace->config.debug_state)
+                fprintf(stderr, "Trace(%s) state changed from %s to %s\n",
+                        trace->uridata, get_trace_state_name(prev_state),
+                        get_trace_state_name(trace->state));
+
         if (need_lock)
                 pthread_mutex_unlock(&trace->libtrace_lock);
@@ -342 +340 @@
                 res->value = (void *)dup;
                 trace_destroy_packet(oldpkt);
-                fprintf(stderr, "Made a packet safe!!\n");
         }
 }
@@ -375 +372 @@
 
 
+
+/**
+ * Dispatches packets to their correct place and applies any translations
+ * as needed
+ * @param trace
+ * @param t
+ * @param packet (in, out) this will be set to NULL if the user doesn't return the packet for reuse
+ * @return -1 if an error or EOF has occured and the trace should end otherwise 0 to continue as normal
+ */
+static inline int dispatch_packets(libtrace_t *trace, libtrace_thread_t *t, libtrace_packet_t **packets,
+                                   size_t nb_packets) {
+        libtrace_message_t message;
+        size_t i;
+        for (i = 0; i < nb_packets; ++i) {
+                if (packets[i]->error > 0) {
+                        packets[i] = (*trace->per_pkt)(trace, packets[i], NULL, t);
+                } else if (packets[i]->error == READ_TICK) {
+                        message.code = MESSAGE_TICK;
+                        message.additional.uint64 = trace_packet_get_order(packets[i]);
+                        message.sender = t;
+                        (*trace->per_pkt)(trace, NULL, &message, t);
+                } else if (packets[i]->error != READ_MESSAGE) {
+                        // An error this should be the last packet we read
+                        size_t z;
+                        // We could have an eof or error and a message such as pause
+                        for (z = i ; 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 <= 0);
+                        }
+                        return -1;
+                }
+                // -2 is a message its not worth checking now just finish this lot and we'll check
+                // when we loop next
+        }
+        return 0;
+}
+
 /**
  * The is the entry point for our packet processing threads.
@@ -417 +451 @@
                                         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
+                                        // If a hasher thread is running empty input queues so we don't lose data
                                         if (trace_has_dedicated_hasher(trace)) {
                                                 fprintf(stderr, "Trace is using a hasher thread emptying queues\n");
@@ -423 +457 @@
                                                 while(!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) {
+                                                        if (dispatch_packets(trace, t, packets, 1) == -1) {
                                                                 // EOF or error, either way we'll stop
                                                                 while (!libtrace_ringbuffer_is_empty(&t->rbuffer)) {
@@ -460 +492 @@
                 }
                 // Loop through the packets we just read
-                for (i = 0; i < nb_packets; ++i) {
-                        
-                        if (packets[i]->error > 0) {
-                                packets[i] = (*trace->per_pkt)(trace, packets[i], NULL, t);
-                        } else if (packets[i]->error != -2) {
-                                // An error this should be the last packet we read
-                                size_t z;
-                                // We could have an eof or error and a message such as pause
-                                for (z = i ; 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;
-                        }
-                        // -2 is a message its not worth checking now just finish this lot and we'll check
-                        // when we loop next
-                }
+                if (dispatch_packets(trace, t, packets, nb_packets) == -1)
+                        break;
         }
 
@@ -523 +540 @@
  * core to process it.
  */
-static void* hasher_start(void *data) {
+static void* hasher_entry(void *data) {
         libtrace_t *trace = (libtrace_t *)data;
         libtrace_thread_t * t;
@@ -569 +586 @@
                                         assert(trace->started == false);
                                         assert(trace->state == STATE_FINSHED);
+                                        break;
                                 default:
                                         fprintf(stderr, "Hasher thread didn't expect message code=%d\n", message.code);
@@ -585 +603 @@
                 /* Blocking write to the correct queue - I'm the only writer */
                 if (trace->perpkt_threads[thread].state != THREAD_FINISHED) {
+                        uint64_t order = trace_packet_get_order(packet);
                         libtrace_ringbuffer_write(&trace->perpkt_threads[thread].rbuffer, packet);
+                        if (trace->config.tick_count && order % trace->config.tick_count == 0) {
+                                // Write ticks to everyone else
+                                libtrace_packet_t * pkts[trace->perpkt_thread_count];
+                                memset(pkts, 0, sizeof(void *) * trace->perpkt_thread_count);
+                                libtrace_ocache_alloc(&trace->packet_freelist, (void **) pkts, trace->perpkt_thread_count, trace->perpkt_thread_count);
+                                for (i = 0; i < trace->perpkt_thread_count; i++) {
+                                        pkts[i]->error = READ_TICK;
+                                        trace_packet_set_order(pkts[i], order);
+                                        libtrace_ringbuffer_write(&trace->perpkt_threads[i].rbuffer, pkts[i]);
+                                }
+                        }
                         pkt_skipped = 0;
                 } else {
@@ -705 +735 @@
 {
         size_t i = 0;
+        bool tick_hit = false;
 
         nb_packets = fill_array_with_empty_packets(libtrace, packets, nb_packets);
@@ -712 +743 @@
         for (i = 0; i < nb_packets; ++i) {
                 packets[i]->error = trace_read_packet(libtrace, packets[i]);
-                // Doing this inside the lock ensures the first packet is always
-                // recorded first
                 if (packets[i]->error <= 0) {
                         ++i;
                         break;
                 }
-        }
+                /*
+                if (libtrace->config.tick_count && trace_packet_get_order(packets[i]) % libtrace->config.tick_count == 0) {
+                        tick_hit = true;
+                }*/
+        }
+        // Doing this inside the lock ensures the first packet is always
+        // recorded first
         if (packets[0]->error > 0) {
                 store_first_packet(libtrace, packets[0], t);
         }
         ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
+        /* XXX TODO this needs to be inband with packets, or we don't bother in this case
+        if (tick_hit) {
+                libtrace_message_t tick;
+                tick.additional.uint64 = trace_packet_get_order(packets[i]);
+                tick.code = MESSAGE_TICK;
+                trace_send_message_to_perpkts(libtrace, &tick);
+        } */
         return i;
 }
@@ -739 +781 @@
                 libtrace_ocache_free(&libtrace->packet_freelist, (void **) packets, 1, 1);
         packets[0] = libtrace_ringbuffer_read(&t->rbuffer);
-
-
-        if (packets[0] == NULL) {
-                libtrace_ocache_alloc(&libtrace->packet_freelist, (void **) packets, 1, 1);
-                packets[0]->error = -2;
-        }
 
         if (packets[0]->error < 0)
@@ -756 +792 @@
                         break;
                 }
-                // Message wating
-                if (packets[i] == NULL) {
-                        libtrace_ocache_alloc(&libtrace->packet_freelist, (void **) &packets[i], 1, 1);
-                        packets[i]->error = -2;
-                        ++i;
+                // These are typically urgent
+                if (packets[i]->error < 0)
                         break;
-                }
         }
         
         return i;
-        /*if (*packet) {
-                return (*packet)->error;
-        } else {
-                // This is how we do a notify, we send a message before hand to note that the trace is over etc.
-                // And this will notify the perpkt thread to read that message, this is easiest
-                // since cases like pause can also be dealt with this way without actually
-                // having to be the end of the stream.
-                fprintf(stderr, "Got a NULL packet from hasher therefore message waiting\n");
-                return -2;
-        }*/
 }
 
@@ -1166 +1188 @@
 
         while (!trace_finished(trace)) {
-
-                //while ( != LIBTRACE_MQ_FAILED) { }
-                libtrace_message_queue_get(&t->messages, &message);
-
+                if (trace->config.reporter_polling) {
+                        if (libtrace_message_queue_try_get(&t->messages, &message) == LIBTRACE_MQ_FAILED)
+                                message.code = MESSAGE_POST_REPORTER;
+                } else {
+                        libtrace_message_queue_get(&t->messages, &message);
+                }
                 switch (message.code) {
                         // Check for results
@@ -1198 +1222 @@
                 (*trace->reporter)(trace, &result, NULL);
         }
+        libtrace_vector_destroy(&results);
 
         // GOODBYE
@@ -1320 +1345 @@
                         packet->trace = libtrace;
                         ret=libtrace->format->pread_packet(libtrace, t, packet);
-                        if (ret==(size_t)-1 || ret==(size_t)-2 || ret==0) {
+                        if (ret <= 0) {
                                 return ret;
                         }
@@ -1499 +1524 @@
         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 <
+        if (libtrace->config.packet_cache_size <= 0)
+                libtrace->config.packet_cache_size = (libtrace->config.hasher_queue_size + 1) * libtrace->perpkt_thread_count;
+
+        if (libtrace->config.packet_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");
@@ -1523 +1548 @@
                 t->state = THREAD_RUNNING;
                 libtrace_message_queue_init(&t->messages, sizeof(libtrace_message_t));
-                ASSERT_RET(pthread_create(&t->tid, NULL, hasher_start, (void *) libtrace), == 0);
+                ASSERT_RET(pthread_create(&t->tid, NULL, hasher_entry, (void *) libtrace), == 0);
                 snprintf(name, sizeof(name), "hasher-thread");
                 pthread_setname_np(t->tid, name);
@@ -1534 +1559 @@
                                                  (void (*)(void *))trace_destroy_packet,
                                                  libtrace->config.packet_thread_cache_size,
-                                                 libtrace->config.packet_global_cache_size * 4,
+                                                 libtrace->config.packet_cache_size * 4,
                                                  libtrace->config.fixed_packet_count);
         // Unused slidingwindow code
@@ -1562 +1587 @@
                 t->perpkt_num = i;
                 if (libtrace->hasher)
-                        libtrace_ringbuffer_init(&t->rbuffer, libtrace->config.hasher_queue_size, LIBTRACE_RINGBUFFER_POLLING);
+                        libtrace_ringbuffer_init(&t->rbuffer, libtrace->config.hasher_queue_size,
+                                                 libtrace->config.hasher_polling?LIBTRACE_RINGBUFFER_POLLING:0);
                 // Depending on the mode vector or deque might be chosen
                 libtrace_vector_init(&t->vector, sizeof(libtrace_result_t));
@@ -1654 +1680 @@
         // Special case handle the hasher thread case
         if (trace_has_dedicated_hasher(libtrace)) {
-                fprintf(stderr, "Hasher thread running we deal with this special!\n");
+                if (libtrace->config.debug_state)
+                        fprintf(stderr, "Hasher thread is running, asking it to pause ...");
                 libtrace_message_t message = {0};
                 message.code = MESSAGE_DO_PAUSE;
@@ -1664 +1691 @@
                 }
                 ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
-        }
-
-        fprintf(stderr, "Sending messages \n");
+                if (libtrace->config.debug_state)
+                        fprintf(stderr, " DONE\n");
+        }
+
+        if (libtrace->config.debug_state)
+                fprintf(stderr, "Asking perpkt threads to pause ...");
         // Stop threads, skip this one if it's a perpkt
         for (i = 0; i < libtrace->perpkt_thread_count; i++) {
@@ -1676 +1706 @@
                                 // The hasher has stopped and other threads have messages waiting therefore
                                 // If the queues are empty the other threads would have no data
-                                // So send some NULL packets to simply ask the threads to check there message queues
+                                // So send some message packets to simply ask the threads to check
                                 // We are the only writer since hasher has paused
-                                libtrace_ringbuffer_write(&libtrace->perpkt_threads[i].rbuffer, NULL);
+                                libtrace_packet_t *pkt;
+                                libtrace_ocache_alloc(&libtrace->packet_freelist, (void **) &pkt, 1, 1);
+                                pkt->error = READ_MESSAGE;
+                                libtrace_ringbuffer_write(&libtrace->perpkt_threads[i].rbuffer, pkt);
                         }
                 } else {
@@ -1685 +1718 @@
         }
 
+        if (t) {
+                // A perpkt is doing the pausing, interesting, fake an extra thread paused
+                // We rely on the user to *not* return before starting the trace again
+                thread_change_state(libtrace, t, THREAD_PAUSED, true);
+        }
+
+        // Wait for all threads to pause
+        ASSERT_RET(pthread_mutex_lock(&libtrace->libtrace_lock), == 0);
+        while(libtrace->perpkt_thread_states[THREAD_RUNNING]) {
+                ASSERT_RET(pthread_cond_wait(&libtrace->perpkt_cond, &libtrace->libtrace_lock), == 0);
+        }
+        ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
+
+        if (libtrace->config.debug_state)
+                fprintf(stderr, " DONE\n");
+
         // Deal with the reporter
         if (trace_has_dedicated_reporter(libtrace)) {
-                fprintf(stderr, "Reporter thread running we deal with this special!\n");
+                if (libtrace->config.debug_state)
+                        fprintf(stderr, "Reporter thread is running, asking it to pause ...");
                 libtrace_message_t message = {0};
                 message.code = MESSAGE_DO_PAUSE;
@@ -1697 +1747 @@
                 }
                 ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
-        }
-
-        // Formats must support native message handling if a message is ready
-        // Approach per Perry's suggestion is a non-blocking read
-        // followed by a blocking read. XXX STRIP THIS OUT
-
-        if (t) {
-                // A perpkt is doing the pausing, interesting, fake an extra thread paused
-                // We rely on the user to *not* return before starting the trace again
-                thread_change_state(libtrace, t, THREAD_PAUSED, true);
-        }
-
-        fprintf(stderr, "Asking threads to pause\n");
-
-        // Wait for all threads to pause
-        ASSERT_RET(pthread_mutex_lock(&libtrace->libtrace_lock), == 0);
-        while(libtrace->perpkt_thread_states[THREAD_RUNNING]) {
-                ASSERT_RET(pthread_cond_wait(&libtrace->perpkt_cond, &libtrace->libtrace_lock), == 0);
-        }
-        ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
-
-        fprintf(stderr, "Threads have paused\n");
+                if (libtrace->config.debug_state)
+                        fprintf(stderr, " DONE\n");
+        }
 
         if (trace_supports_parallel(libtrace) && !trace_has_dedicated_hasher(libtrace)) {
@@ -1866 +1897 @@
 
         /* Now the hasher */
-        // XXX signal it to stop if it hasn't already we should never be in this situation!!
         if (trace_has_dedicated_hasher(libtrace)) {
-                fprintf(stderr, "Waiting to join with the hasher\n");
                 pthread_join(libtrace->hasher_thread.tid, NULL);
-                fprintf(stderr, "Joined with the hasher\n");
                 assert(libtrace->hasher_thread.state == THREAD_FINISHED);
         }
@@ -1892 +1920 @@
 
         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);
         }
@@ -1902 +1928 @@
                 libtrace_message_t msg = {0};
                 msg.code = MESSAGE_DO_STOP;
-                fprintf(stderr, "Waiting to join with the keepalive\n");
                 trace_send_message_to_thread(libtrace, &libtrace->keepalive_thread, &msg);
                 pthread_join(libtrace->keepalive_thread.tid, NULL);
-                fprintf(stderr, "Joined with with the keepalive\n");
         }
         
@@ -2226 +2250 @@
                                 libtrace->tracetime = 0;
                         return 0;
+                case TRACE_OPTION_SET_CONFIG:
+                        libtrace->config = *((struct user_configuration *) value);
+                case TRACE_OPTION_GET_CONFIG:
+                        *((struct user_configuration *) value) = libtrace->config;
         }
         return 0;
+}
+
+static bool config_bool_parse(char *value, size_t nvalue) {
+        if (strncmp(value, "true", nvalue) == 0)
+                return true;
+        else if (strncmp(value, "false", nvalue) == 0)
+                return false;
+        else
+                return strtoll(value, NULL, 10) != 0;
+}
+
+static void config_string(struct user_configuration *uc, char *key, size_t nkey, char *value, size_t nvalue) {
+        assert(key);
+        assert(value);
+        assert(uc);
+        if (strncmp(key, "packet_cache_size", nkey) == 0
+                || strncmp(key, "pcs", nkey) == 0) {
+                uc->packet_cache_size = strtoll(value, NULL, 10);
+        } else if (strncmp(key, "packet_thread_cache_size", nkey) == 0
+                           || strncmp(key, "ptcs", nkey) == 0) {
+                uc->packet_thread_cache_size = strtoll(value, NULL, 10);
+        } else if (strncmp(key, "fixed_packet_count", nkey) == 0
+                  || strncmp(key, "fpc", nkey) == 0) {
+                uc->fixed_packet_count = config_bool_parse(value, nvalue);
+        } else if (strncmp(key, "burst_size", nkey) == 0
+                   || strncmp(key, "bs", nkey) == 0) {
+                uc->burst_size = strtoll(value, NULL, 10);
+        } else if (strncmp(key, "tick_interval", nkey) == 0
+                   || strncmp(key, "ti", nkey) == 0) {
+                uc->tick_interval = strtoll(value, NULL, 10);
+        } else if (strncmp(key, "tick_count", nkey) == 0
+                   || strncmp(key, "tc", nkey) == 0) {
+                uc->tick_count = strtoll(value, NULL, 10);
+        } else if (strncmp(key, "perpkt_threads", nkey) == 0
+                   || strncmp(key, "pt", nkey) == 0) {
+                uc->perpkt_threads = strtoll(value, NULL, 10);
+        } else if (strncmp(key, "hasher_queue_size", nkey) == 0
+                   || strncmp(key, "hqs", nkey) == 0) {
+                uc->hasher_queue_size = strtoll(value, NULL, 10);
+        } else if (strncmp(key, "hasher_polling", nkey) == 0
+                   || strncmp(key, "hp", nkey) == 0) {
+                uc->hasher_polling = config_bool_parse(value, nvalue);
+        } else if (strncmp(key, "reporter_polling", nkey) == 0
+                   || strncmp(key, "rp", nkey) == 0) {
+                uc->reporter_polling = config_bool_parse(value, nvalue);
+        } else if (strncmp(key, "reporter_thold", nkey) == 0
+                   || strncmp(key, "rt", nkey) == 0) {
+                uc->reporter_thold = strtoll(value, NULL, 10);
+        } else if (strncmp(key, "debug_state", nkey) == 0
+                   || strncmp(key, "ds", nkey) == 0) {
+                uc->debug_state = config_bool_parse(value, nvalue);
+        } else {
+                fprintf(stderr, "No matching value %s(=%s)\n", key, value);
+        }
+}
+
+DLLEXPORT void parse_user_config(struct user_configuration* uc, char * str) {
+        char *pch;
+        char key[100];
+        char value[100];
+        assert(str);
+        assert(uc);
+        printf ("Splitting string \"%s\" into tokens:\n",str);
+        pch = strtok (str," ,.-");
+        while (pch != NULL)
+        {
+                if (sscanf(pch, "%99[^=]=%99s", key, value) == 2) {
+                        config_string(uc, key, sizeof(key), value, sizeof(value));
+                } else {
+                        fprintf(stderr, "Error parsing %s\n", pch);
+                }
+                pch = strtok (NULL," ,.-");
+        }
+}
+
+DLLEXPORT void parse_user_config_file(struct user_configuration* uc, FILE *file) {
+        char line[1024];
+        while (fgets(line, sizeof(line), file) != NULL)
+        {
+                parse_user_config(uc, line);
+        }
 }
 

tools/traceanon/traceanon_parallel.c

@@ -198 +198 @@
                         case MESSAGE_STARTING:
                                 enc_init(enc_type,key);
+                        break;
+                        case MESSAGE_TICK:
+                                trace_publish_result(trace, t, mesg->additional.uint64, NULL, RESULT_TICK);
                 }
         }
@@ -207 +210 @@
 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);
+                if (result->type == RESULT_PACKET) {
+                        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);
+
+                } else {
+                        assert(result->type == RESULT_TICK);
+                        // Ignore it
+                }
         }
         return NULL;
@@ -228 +238 @@
         char *compress_type_str=NULL;
         trace_option_compresstype_t compress_type = TRACE_OPTION_COMPRESSTYPE_NONE;
-
+        struct user_configuration uc;
+        ZERO_USER_CONFIG(uc);
 
         if (argc<2)
@@ -244 +255 @@
                         { "compress-type",      1, 0, 'Z' },
                         { "libtrace-help",      0, 0, 'H' },
+                        { "config",             1, 0, 'u' },
+                    { "config-file",            1, 0, 'U' },
                         { NULL,                 0, 0, 0   },
                 };
 
-                int c=getopt_long(argc, argv, "Z:z:sc:f:dp:H",
+                int c=getopt_long(argc, argv, "Z:z:sc:f:dp:Hu:U:",
                                 long_options, &option_index);
 
@@ -297 +310 @@
                                   exit(1); 
                                   break;
+                        case 'u':
+                                  parse_user_config(&uc, optarg);
+                                  break;
+                        case 'U':;
+                                FILE * f = fopen(optarg, "r");
+                                if (f != NULL) {
+                                        parse_user_config_file(&uc, f);
+                                } else {
+                                        perror("Failed to open configuration file\n");
+                                        usage(argv[0]);
+                                }
+                                break;
                         default:
                                 fprintf(stderr,"unknown option: %c\n",c);
@@ -391 +416 @@
          
         int i = 1;
-        trace_parallel_config(trace, TRACE_OPTION_SEQUENTIAL, &i);
+        trace_parallel_config(trace, TRACE_OPTION_ORDERED, &i);
         //trace_parallel_config(trace, TRACE_OPTION_SET_PERPKT_BUFFER_SIZE, &i);
-        i = 2;
+        i = 6;
     trace_parallel_config(trace, TRACE_OPTION_SET_PERPKT_THREAD_COUNT, &i);
+        trace_parallel_config(trace, TRACE_OPTION_SET_CONFIG, &uc);
+        trace_set_hasher(trace, HASHER_CUSTOM, bad_hash, NULL);
         
         if (trace_pstart(trace, NULL, &per_packet, &write_out)==-1) {