Changeset 17a3dff for lib

Timestamp:

04/26/14 23:23:17 (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:

5ce14a5

Parents:

60e8e86

Message:

Rename from google map/reduce framework names to something more meaningful.
Rename mapper to perpkt since this is what it actually is in libtrace.

Location:

lib

Files:

• format_dpdk.c (modified) (4 diffs)
• format_linux.c (modified) (3 diffs)
• libtrace.h.in (modified) (5 diffs)
• libtrace_int.h (modified) (10 diffs)
• trace.c (modified) (8 diffs)
• trace_parallel.c (modified) (66 diffs)

lib/format_dpdk.c

@@ -827 +827 @@
     return 0;
 }
-int mapper_start(void *data); // This actually a void*
 
 /* Attach memory to the port and start the port or restart the ports.
@@ -967 +966 @@
     // Can use remote launch for all
     /*RTE_LCORE_FOREACH_SLAVE(i) {
-                rte_eal_remote_launch(mapper_start, (void *)libtrace, i);
+                rte_eal_remote_launch(perpkt_threads_entry, (void *)libtrace, i);
         }*/
     
@@ -996 +995 @@
     char err[500];
     int enabled_lcore_count = 0, i=0;
-    int tot = libtrace->mapper_thread_count;
+    int tot = libtrace->perpkt_thread_count;
     err[0] = 0;
         
-        libtrace->mapper_thread_count;
+        libtrace->perpkt_thread_count;
         
         for (i = 0; i < RTE_MAX_LCORE; i++)
@@ -1007 +1006 @@
         }
         
-        tot = MIN(libtrace->mapper_thread_count, enabled_lcore_count);
+        tot = MIN(libtrace->perpkt_thread_count, enabled_lcore_count);
         tot = MIN(tot, 8);
-        printf("Running pstart DPDK %d %d %d %d\n", tot, libtrace->mapper_thread_count, enabled_lcore_count, rte_lcore_count());
+        printf("Running pstart DPDK %d %d %d %d\n", tot, libtrace->perpkt_thread_count, enabled_lcore_count, rte_lcore_count());
         
     if (dpdk_start_port_queues(libtrace, FORMAT(libtrace), err, sizeof(err), tot) != 0) {

lib/format_linux.c

@@ -649 +649 @@
 {
         int i;
-        int tot = libtrace->mapper_thread_count;
+        int tot = libtrace->perpkt_thread_count;
         printf("CAlling native pause packet\n");
         
@@ -690 +690 @@
 static int linuxnative_pstart_input(libtrace_t *libtrace) {
         int i = 0;
-        int tot = libtrace->mapper_thread_count;
+        int tot = libtrace->perpkt_thread_count;
         int iserror = 0;
         // We store this here otherwise it will be leaked if the memory doesn't know
@@ -701 +701 @@
                 // Whats going on this might not work 100%
                 // We assume all sockets have been closed ;)
-                printf("Pause and then start called again lets hope that mapper_thread_count hasn't changed\n");
+                printf("Pause and then start called again lets hope that perpkt_thread_count hasn't changed\n");
         }
         

lib/libtrace.h.in

@@ -3130 +3130 @@
 DLLEXPORT void trace_destroy_result(libtrace_result_t ** result);
 
-// Ways to access Global and TLS  storage that we provide the user
+// Ways to access Global and TLS storage that we provide the user
 DLLEXPORT void * trace_get_global(libtrace_t *trace);
 DLLEXPORT void * trace_set_global(libtrace_t *trace, void * data);
@@ -3151 +3151 @@
 DLLEXPORT void trace_packet_set_hash(libtrace_packet_t * packet, uint64_t hash);
 
+DLLEXPORT int retrive_first_packet(libtrace_t *libtrace, libtrace_packet_t **packet, struct timeval **tv);
 
 typedef enum {
@@ -3175 +3176 @@
          * A unblockable warning message will be printed to stderr in this case.
          */
-        TRACE_OPTION_SET_MAPPER_BUFFER_SIZE,
+        TRACE_OPTION_SET_PERPKT_BUFFER_SIZE,
         
         /**
-         * Libtrace set mapper thread count
+         * Libtrace set perpkt thread count
          */
-        TRACE_OPTION_SET_MAPPER_THREAD_COUNT,
+        TRACE_OPTION_SET_PERPKT_THREAD_COUNT,
         
         /**
@@ -3220 +3221 @@
         MESSAGE_STOPPED,
         MESSAGE_FIRST_PACKET,
-        MESSAGE_MAPPER_ENDED,
-        MESSAGE_MAPPER_RESUMED,
-        MESSAGE_MAPPER_PAUSED,
-        MESSAGE_MAPPER_EOF,
+        MESSAGE_PERPKT_ENDED,
+        MESSAGE_PERPKT_RESUMED,
+        MESSAGE_PERPKT_PAUSED,
+        MESSAGE_PERPKT_EOF,
         MESSAGE_POST_REDUCE,
         MESSAGE_POST_RANGE,
@@ -3229 +3230 @@
 };
 
+enum hasher_types {
+        HASHER_BALANCE, /** Balance load across CPUs best as possible this is basically to say don't care about hash, but this might still might be implemented using a hash or round robin etc.. */
+        HASHER_BIDIRECTIONAL, /** Use a hash which is uni-directional for TCP flows (IP src dest,TCP port src dest), non TCP
+                                                        Will be sent to the same place, with the exception of UDP which may or may not be sent to separate cores */
+        HASHER_UNIDIRECTIONAL, /** Use a hash which is uni-directional across TCP flow */
+        HASHER_CUSTOM, /** Always use the user supplied hasher */
+        HASHER_HARDWARE, /** Set by the format if the hashing is going to be done in hardware */
+};
+
 DLLEXPORT int trace_parallel_config(libtrace_t *libtrace, trace_parallel_option_t option, void *value);
 DLLEXPORT libtrace_packet_t* trace_result_packet(libtrace_t * libtrace, libtrace_packet_t * packet);
 DLLEXPORT void trace_free_result_packet(libtrace_t * libtrace, libtrace_packet_t * packet);
+enum hasher_types;
+DLLEXPORT int trace_set_hasher(libtrace_t *trace, enum hasher_types type, fn_hasher hasher, void *data);
 
 #ifdef __cplusplus

lib/libtrace_int.h

@@ -175 +175 @@
         THREAD_EMPTY,
         THREAD_HASHER,
-        THREAD_MAPPER,
+        THREAD_PERPKT,
         THREAD_REDUCER
 };
@@ -184 +184 @@
         THREAD_FINISHED,
         THREAD_PAUSED
-};
-
-enum hasher_types {
-        HASHER_BALANCE, /** Balance load across CPUs best as possible this is basically to say don't care about hash, but this might still might be implemented using a hash or round robin etc.. */
-        HASHER_BIDIRECTIONAL, /** Use a hash which is uni-directional for TCP flows (IP src dest,TCP port src dest), non TCP
-                                                        Will be sent to the same place, with the exception of UDP which may or may not be sent to separate cores */
-        HASHER_UNIDIRECTIONAL, /** Use a hash which is uni-directional across TCP flow */
-        HASHER_CUSTOM, /** Always use the user supplied hasher */
-        HASHER_HARDWARE, /** Set by the format if the hashing is going to be done in hardware */
 };
 
@@ -206 +197 @@
 #define REDUCE_STEPPING 0x10
 
-#define MAPPER_USE_SLIDING_WINDOW 0x20
+#define PERPKT_USE_SLIDING_WINDOW 0x20
 
 /**
@@ -218 +209 @@
         void* user_data; // TLS for the user to use
         pthread_t tid;
-        int map_num; // A number from 0-X that represents this mapper number
+        int perpkt_num; // A number from 0-X that represents this perpkt threads number
                                 // in the table, intended to quickly identify this thread
-                                // -1 represents NA (such as in the case this is not a mapper thread)
+                                // -1 represents NA (such as the case this is not a perpkt thread)
         libtrace_ringbuffer_t rbuffer; // Input
         libtrace_vector_t vector; // Output
@@ -243 +234 @@
 struct first_packets {
         pthread_spinlock_t lock;
-        size_t count; // If == mappers we have all
+        size_t count; // If == perpkt_thread_count threads we have all
         size_t first; // Valid if count != 0
         struct __packet_storage_magic_type {
@@ -285 +276 @@
         /** Use to control pausing threads and finishing threads etc always used with libtrace_lock */
         pthread_cond_t perpkt_cond;
-        /** Set to the number of mapper threads that are finishing (or have finished), or to -1 once all have been joined, 0 implies all are running */
-        int mappers_finishing;
-        /** A count of mappers that are pausing */
-        int perpkt_pausing;
+        /** Set to the number of perpkt threads that are finishing (or have finished), or to -1 once all have been joined, 0 implies all are running */
+        int perpkts_finishing;
+        /** A count of perpkt threads that are pausing */
+        int perpkts_pausing;
         
         /** For the sliding window hasher implementation */
@@ -294 +285 @@
         /** Set once trace_join has been called */
         bool joined;
-        /** Set to indicate a mappers queue is full and such the writing mapper cannot proceed */
-        bool mapper_queue_full;
+        /** 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;
@@ -302 +293 @@
         /** The actual freelist */
         libtrace_ringbuffer_t packet_freelist;
-        /** The number of packets that can queue per mapper thread - XXX consider deadlocks with non malloc()'d packets that need to be released */
-        int mapper_buffer_size;
+        /** 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;
@@ -320 +311 @@
         libtrace_thread_t hasher_thread;
         libtrace_thread_t reducer_thread;
-        int mapper_thread_count;
-        libtrace_thread_t * mapper_threads; // All our mapper threads
+        int perpkt_thread_count;
+        libtrace_thread_t * perpkt_threads; // All our perpkt threads
         libtrace_slidingwindow_t sliding_window;
         sem_t sem;
@@ -331 +322 @@
 inline void libtrace_zero_thread(libtrace_thread_t * t);
 inline void store_first_packet(libtrace_t *libtrace, libtrace_packet_t *packet, libtrace_thread_t *t);
+libtrace_thread_t * get_thread_table(libtrace_t *libtrace);
+int get_thread_table_num(libtrace_t *libtrace);
+
 
 /** A libtrace output trace

lib/trace.c

@@ -111 +111 @@
 
 /* Set once pstart is called used for backwards compatibility reasons */
-extern int libtrace_parallel = 0;
+int libtrace_parallel = 0;
 
 /* strncpy is not assured to copy the final \0, so we
@@ -263 +263 @@
         // libtrace->libtrace_lock
         // libtrace->perpkt_cond;
-        libtrace->perpkt_pausing = 0;
-        libtrace->mapper_queue_full = false;
-        libtrace->mappers_finishing = -1;
+        libtrace->perpkts_pausing = 0;
+        libtrace->perpkt_queue_full = false;
+        libtrace->perpkts_finishing = -1;
         libtrace->reducer_flags = 0;
         libtrace->joined = false;
@@ -273 +273 @@
         libtrace->hasher = NULL;
         libtrace->packet_freelist_size = 0;
-        libtrace->mapper_buffer_size = 0;
+        libtrace->perpkt_buffer_size = 0;
         libtrace->expected_key = 0;
         libtrace_zero_ringbuffer(&libtrace->packet_freelist);
@@ -280 +280 @@
         libtrace_zero_slidingwindow(&libtrace->sliding_window);
         libtrace->reducer_thread.type = THREAD_EMPTY;
-        libtrace->mapper_thread_count = 0;
-        libtrace->mapper_threads = NULL;
+        libtrace->perpkt_thread_count = 0;
+        libtrace->perpkt_threads = NULL;
 
         /* Parse the URI to determine what sort of trace we are dealing with */
@@ -381 +381 @@
         // libtrace->libtrace_lock
         // libtrace->perpkt_cond;
-        libtrace->perpkt_pausing = 0;
-        libtrace->mapper_queue_full = false;
-        libtrace->mappers_finishing = -1;
+        libtrace->perpkts_pausing = 0;
+        libtrace->perpkt_queue_full = false;
+        libtrace->perpkts_finishing = -1;
         libtrace->reducer_flags = 0;
         libtrace->joined = false;
@@ -392 +392 @@
         libtrace->expected_key = 0;
         libtrace->packet_freelist_size = 0;
-        libtrace->mapper_buffer_size = 0;
+        libtrace->perpkt_buffer_size = 0;
         libtrace_zero_ringbuffer(&libtrace->packet_freelist);
         libtrace_zero_thread(&libtrace->hasher_thread);
@@ -398 +398 @@
         libtrace_zero_slidingwindow(&libtrace->sliding_window);
         libtrace->reducer_thread.type = THREAD_EMPTY;
-        libtrace->mapper_thread_count = 0;
-        libtrace->mapper_threads = NULL;
+        libtrace->perpkt_thread_count = 0;
+        libtrace->perpkt_threads = NULL;
         
         for(tmp=formats_list;tmp;tmp=tmp->next) {
@@ -655 +655 @@
         libtrace_ringbuffer_destroy(&libtrace->packet_freelist);
         
-        for (i = 0; i < libtrace->mapper_thread_count; ++i) {
-                        assert (libtrace_vector_get_size(&libtrace->mapper_threads[i].vector) == 0);
-                        libtrace_vector_destroy(&libtrace->mapper_threads[i].vector);
-        }
-        free(libtrace->mapper_threads);
-        libtrace->mapper_threads = NULL;
-        libtrace->mapper_thread_count = 0;
+        for (i = 0; i < libtrace->perpkt_thread_count; ++i) {
+                assert (libtrace_vector_get_size(&libtrace->perpkt_threads[i].vector) == 0);
+                libtrace_vector_destroy(&libtrace->perpkt_threads[i].vector);
+        }
+        free(libtrace->perpkt_threads);
+        libtrace->perpkt_threads = NULL;
+        libtrace->perpkt_thread_count = 0;
         
         if (libtrace->event.packet) {

lib/trace_parallel.c

@@ -125 +125 @@
         int i;
         struct multithreading_stats totals = {0};
-        for (i = 0; i < libtrace->mapper_thread_count ; i++) {
-                printf("\nStats for mapper thread#%d\n", i);
+        for (i = 0; i < libtrace->perpkt_thread_count ; i++) {
+                printf("\nStats for perpkt thread#%d\n", i);
                 printf("\tfull_queue_hits: %"PRIu64"\n", contention_stats[i].full_queue_hits);
                 totals.full_queue_hits += contention_stats[i].full_queue_hits;
@@ -132 +132 @@
                 totals.wait_for_fill_complete_hits += contention_stats[i].wait_for_fill_complete_hits;
         }
-        printf("\nTotals for mapper threads\n");
+        printf("\nTotals for perpkt threads\n");
         printf("\tfull_queue_hits: %"PRIu64"\n", totals.full_queue_hits);
         printf("\twait_for_fill_complete_hits: %"PRIu64"\n", totals.wait_for_fill_complete_hits);
@@ -147 +147 @@
         libtrace_zero_deque(&t->deque);
         t->recorded_first = false;
-        t->map_num = -1;
+        t->perpkt_num = -1;
 }
 
@@ -156 +156 @@
         pthread_t tid = pthread_self();
 
-        for (;i<libtrace->mapper_thread_count ;++i) {
-                if (pthread_equal(tid, libtrace->mapper_threads[i].tid))
-                        return &libtrace->mapper_threads[i];
+        for (;i<libtrace->perpkt_thread_count ;++i) {
+                if (pthread_equal(tid, libtrace->perpkt_threads[i].tid))
+                        return &libtrace->perpkt_threads[i];
         }
         return NULL;
 }
 
-int get_thread_table_num(libtrace_t *libtrace);
-DLLEXPORT int get_thread_table_num(libtrace_t *libtrace) {
+int get_thread_table_num(libtrace_t *libtrace) {
         int i = 0;
         pthread_t tid = pthread_self();
-        for (;i<libtrace->mapper_thread_count; ++i) {
-                if (pthread_equal(tid, libtrace->mapper_threads[i].tid))
+        for (;i<libtrace->perpkt_thread_count; ++i) {
+                if (pthread_equal(tid, libtrace->perpkt_threads[i].tid))
                         return i;
         }
@@ -196 +195 @@
         printf("Pausing thread #%d\n", get_thread_table_num(trace));
         assert(pthread_mutex_lock(&trace->libtrace_lock) == 0);
-        trace->perpkt_pausing++;
+        trace->perpkts_pausing++;
         pthread_cond_broadcast(&trace->perpkt_cond);
         while (!trace->started) {
                 assert(pthread_cond_wait(&trace->perpkt_cond, &trace->libtrace_lock) == 0);
         }
-        trace->perpkt_pausing--;
+        trace->perpkts_pausing--;
         pthread_cond_broadcast(&trace->perpkt_cond);
         assert(pthread_mutex_unlock(&trace->libtrace_lock) == 0);
@@ -207 +206 @@
 }
 
-void* mapper_start(void *data) {
+/**
+ * The is the entry point for our packet processing threads.
+ */
+static void* perpkt_threads_entry(void *data) {
         libtrace_t *trace = (libtrace_t *)data;
         libtrace_thread_t * t;
@@ -216 +218 @@
         t = get_thread_table(trace);
         assert(t);
-        //printf("Yay Started Mapper thread #%d\n", (int) get_thread_table_num(trace));
+        //printf("Yay Started perpkt thread #%d\n", (int) get_thread_table_num(trace));
         assert(pthread_mutex_unlock(&trace->libtrace_lock) == 0);
 
@@ -245 +247 @@
                 }
 
-                if (trace->mapper_thread_count == 1) {
+                if (trace->perpkt_thread_count == 1) {
                         if (!packet) {
                                 if (!libtrace_ringbuffer_try_sread_bl(&trace->packet_freelist, (void **) &packet))
@@ -286 +288 @@
 
         // Notify only after we've defiantly set the state to finished
-        message.code = MESSAGE_MAPPER_ENDED;
+        message.code = MESSAGE_PERPKT_ENDED;
         message.additional = NULL;
         trace_send_message_to_reducer(trace, &message);
@@ -335 +337 @@
                 /* We are guaranteed to have a hash function i.e. != NULL */
                 trace_packet_set_hash(packet, (*trace->hasher)(packet, trace->hasher_data));
-                thread = trace_packet_get_hash(packet) % trace->mapper_thread_count;
+                thread = trace_packet_get_hash(packet) % trace->perpkt_thread_count;
                 /* Blocking write to the correct queue - I'm the only writer */
-                if (trace->mapper_threads[thread].state != THREAD_FINISHED) {
-                        libtrace_ringbuffer_write(&trace->mapper_threads[thread].rbuffer, packet);
+                if (trace->perpkt_threads[thread].state != THREAD_FINISHED) {
+                        libtrace_ringbuffer_write(&trace->perpkt_threads[thread].rbuffer, packet);
                         pkt_skipped = 0;
                 } else {
@@ -346 +348 @@
 
         /* Broadcast our last failed read to all threads */
-        for (i = 0; i < trace->mapper_thread_count; i++) {
+        for (i = 0; i < trace->perpkt_thread_count; i++) {
                 libtrace_packet_t * bcast;
                 printf("Broadcasting error/EOF now the trace is over\n");
-                if (i == trace->mapper_thread_count - 1) {
+                if (i == trace->perpkt_thread_count - 1) {
                         bcast = packet;
                 } else {
@@ -356 +358 @@
                 }
                 assert(pthread_mutex_lock(&trace->libtrace_lock) == 0);
-                if (trace->mapper_threads[i].state != THREAD_FINISHED) {
+                if (trace->perpkt_threads[i].state != THREAD_FINISHED) {
                         assert(pthread_mutex_unlock(&trace->libtrace_lock) == 0);
                         // Unlock early otherwise we could deadlock
-                        libtrace_ringbuffer_write(&trace->mapper_threads[i].rbuffer, NULL);
+                        libtrace_ringbuffer_write(&trace->perpkt_threads[i].rbuffer, NULL);
                 } else {
                         assert(pthread_mutex_unlock(&trace->libtrace_lock) == 0);
@@ -371 +373 @@
         // Notify only after we've defiantly set the state to finished
         libtrace_message_t message;
-        message.code = MESSAGE_MAPPER_ENDED;
+        message.code = MESSAGE_PERPKT_ENDED;
         message.additional = NULL;
         trace_send_message_to_reducer(trace, &message);
@@ -427 +429 @@
 {
         int this_thread = get_thread_table_num(libtrace); // Could be worth caching ... ?
-        libtrace_thread_t* t = &libtrace->mapper_threads[this_thread];
+        libtrace_thread_t* t = &libtrace->perpkt_threads[this_thread];
 
         if (*packet) // Recycle the old get the new
@@ -468 +470 @@
  * before returning EOF/error.
  */
-inline static int trace_handle_finishing_mapper(libtrace_t *libtrace, libtrace_packet_t **packet, libtrace_thread_t * t)
+inline static int trace_handle_finishing_perpkt(libtrace_t *libtrace, libtrace_packet_t **packet, libtrace_thread_t * t)
 {
         /* We are waiting for the condition that another thread ends to check
@@ -480 +482 @@
 
                 // Check before
-                if (libtrace->mappers_finishing == libtrace->mapper_thread_count) {
+                if (libtrace->perpkts_finishing == libtrace->perpkt_thread_count) {
                         complete = true;
                         assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
@@ -489 +491 @@
 
                 // Check after
-                if (libtrace->mappers_finishing == libtrace->mapper_thread_count) {
+                if (libtrace->perpkts_finishing == libtrace->perpkt_thread_count) {
                         complete = true;
                         assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
@@ -512 +514 @@
  * Expects the libtrace_lock to not be held
  */
-inline static int trace_finish_mapper(libtrace_t *libtrace, libtrace_packet_t **packet, libtrace_thread_t * t)
+inline static int trace_finish_perpkt(libtrace_t *libtrace, libtrace_packet_t **packet, libtrace_thread_t * t)
 {
         assert(pthread_mutex_lock(&libtrace->libtrace_lock) == 0);
         t->state = THREAD_FINISHING;
-        libtrace->mappers_finishing++;
+        libtrace->perpkts_finishing++;
         pthread_cond_broadcast(&libtrace->perpkt_cond);
         assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
-        return trace_handle_finishing_mapper(libtrace, packet, t);
+        return trace_handle_finishing_perpkt(libtrace, packet, t);
 }
 
@@ -537 +539 @@
 {
         int this_thread = get_thread_table_num(libtrace); // Could be worth caching ... ?
-        libtrace_thread_t * t = &libtrace->mapper_threads[this_thread];
-        int thread, ret, psize;
+        libtrace_thread_t * t = &libtrace->perpkt_threads[this_thread];
+        int thread, ret/*, psize*/;
 
         while (1) {
@@ -554 +556 @@
 
                 // Another thread cannot write a packet because a queue has filled up. Is it ours?
-                if (libtrace->mapper_queue_full) {
+                if (libtrace->perpkt_queue_full) {
                         contention_stats[this_thread].wait_for_fill_complete_hits++;
                         assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
@@ -576 +578 @@
 
                 trace_packet_set_hash(*packet, (*libtrace->hasher)(*packet, libtrace->hasher_data));
-                thread = trace_packet_get_hash(*packet) % libtrace->mapper_thread_count;
+                thread = trace_packet_get_hash(*packet) % libtrace->perpkt_thread_count;
                 if (thread == this_thread) {
                         // If it's this thread we must be in order because we checked the buffer once we got the lock
@@ -583 +585 @@
                 }
 
-                if (libtrace->mapper_threads[thread].state != THREAD_FINISHED) {
-                        while (!libtrace_ringbuffer_try_swrite_bl(&libtrace->mapper_threads[thread].rbuffer, *packet)) {
-                                libtrace->mapper_queue_full = true;
+                if (libtrace->perpkt_threads[thread].state != THREAD_FINISHED) {
+                        while (!libtrace_ringbuffer_try_swrite_bl(&libtrace->perpkt_threads[thread].rbuffer, *packet)) {
+                                libtrace->perpkt_queue_full = true;
                                 assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
                                 contention_stats[this_thread].full_queue_hits++;
@@ -591 +593 @@
                         }
                         *packet = NULL;
-                        libtrace->mapper_queue_full = false;
+                        libtrace->perpkt_queue_full = false;
                 } else {
                         /* We can get here if the user closes the thread before natural completion/or error */
@@ -616 +618 @@
 {
         int this_thread = get_thread_table_num(libtrace); // Could be worth caching ... ?
-        libtrace_thread_t * t = &libtrace->mapper_threads[this_thread];
-        int ret, i, thread, psize;
+        libtrace_thread_t * t = &libtrace->perpkt_threads[this_thread];
+        int ret, i, thread/*, psize*/;
 
         if (t->state == THREAD_FINISHING)
-                return trace_handle_finishing_mapper(libtrace, packet, t);
+                return trace_handle_finishing_perpkt(libtrace, packet, t);
 
         while (1) {
@@ -641 +643 @@
 
                 // TODO put on *proper* condition variable
-                if (libtrace->mapper_queue_full) {
+                if (libtrace->perpkt_queue_full) {
                         assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
                         assert(sem_post(&libtrace->sem) == 0);
@@ -661 +663 @@
                                 return 0;
                         else
-                                return trace_finish_mapper(libtrace, packet, t);
+                                return trace_finish_perpkt(libtrace, packet, t);
                 }
                 assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
@@ -678 +680 @@
                 while (libtrace_slidingwindow_read_ready(&libtrace->sliding_window)) {
                         assert(pthread_rwlock_wrlock(&libtrace->window_lock) == 0);
-                        if (libtrace->mapper_queue_full) {
+                        if (libtrace->perpkt_queue_full) {
                                 // I might be the holdup in which case if I can read my queue I should do that and return
                                 if(try_waiting_queue(libtrace, t, packet, &ret)) {
@@ -689 +691 @@
                         // Read greedily as many as we can
                         while (libtrace_slidingwindow_try_read(&libtrace->sliding_window, (void **) packet, NULL)) {
-                                thread = trace_packet_get_hash(*packet) % libtrace->mapper_thread_count;
-                                if (libtrace->mapper_threads[thread].state != THREAD_FINISHED) {
-                                        while (!libtrace_ringbuffer_try_swrite_bl(&libtrace->mapper_threads[thread].rbuffer, *packet)) {
+                                thread = trace_packet_get_hash(*packet) % libtrace->perpkt_thread_count;
+                                if (libtrace->perpkt_threads[thread].state != THREAD_FINISHED) {
+                                        while (!libtrace_ringbuffer_try_swrite_bl(&libtrace->perpkt_threads[thread].rbuffer, *packet)) {
                                                 if (this_thread == thread)
                                                 {
@@ -700 +702 @@
                                                         if(try_waiting_queue(libtrace, t, packet, &ret)) {
                                                                 // We must be able to write this now 100% without fail
-                                                                libtrace_ringbuffer_write(&libtrace->mapper_threads[thread].rbuffer, *packet);
+                                                                libtrace_ringbuffer_write(&libtrace->perpkt_threads[thread].rbuffer, *packet);
                                                                 assert(sem_post(&libtrace->sem) == 0);
                                                                 assert(pthread_rwlock_unlock(&libtrace->window_lock) == 0);
@@ -709 +711 @@
                                                 }
                                                 // Not us we have to give the other threads a chance to write there packets then
-                                                libtrace->mapper_queue_full = true;
+                                                libtrace->perpkt_queue_full = true;
                                                 assert(pthread_rwlock_unlock(&libtrace->window_lock) == 0);
-                                                for (i = 0; i < libtrace->mapper_thread_count-1; i++) // Release all other threads to read there packets
+                                                for (i = 0; i < libtrace->perpkt_thread_count-1; i++) // Release all other threads to read there packets
                                                         assert(sem_post(&libtrace->sem) == 0);
 
@@ -717 +719 @@
                                                 assert(pthread_rwlock_wrlock(&libtrace->window_lock) == 0);
                                                 // Grab these back
-                                                for (i = 0; i < libtrace->mapper_thread_count-1; i++) // Release all other threads to read there packets
+                                                for (i = 0; i < libtrace->perpkt_thread_count-1; i++) // Release all other threads to read there packets
                                                         assert(sem_wait(&libtrace->sem) == 0);
-                                                libtrace->mapper_queue_full = false;
+                                                libtrace->perpkt_queue_full = false;
                                         }
                                         assert(sem_post(&libtrace->sem) == 0);
@@ -753 +755 @@
                 dup = trace_copy_packet(packet);
                 assert(pthread_spin_lock(&libtrace->first_packets.lock) == 0);
-                libtrace->first_packets.packets[t->map_num].packet = dup;
+                libtrace->first_packets.packets[t->perpkt_num].packet = dup;
                 //printf("Stored first packet time=%f\n", trace_get_seconds(dup));
-                memcpy(&libtrace->first_packets.packets[t->map_num].tv, &tv, sizeof(tv));
+                memcpy(&libtrace->first_packets.packets[t->perpkt_num].tv, &tv, sizeof(tv));
                 // Now update the first
                 libtrace->first_packets.count++;
                 if (libtrace->first_packets.count == 1) {
                         // We the first entry hence also the first known packet
-                        libtrace->first_packets.first = t->map_num;
+                        libtrace->first_packets.first = t->perpkt_num;
                 } else {
                         // Check if we are newer than the previous 'first' packet
@@ -766 +768 @@
                         if (trace_get_seconds(dup) <
                                 trace_get_seconds(libtrace->first_packets.packets[first].packet))
-                                libtrace->first_packets.first = t->map_num;
+                                libtrace->first_packets.first = t->perpkt_num;
                 }
                 assert(pthread_spin_unlock(&libtrace->first_packets.lock) == 0);
@@ -778 +780 @@
 
 /**
- * Returns 1 if its certain that the first packet is truly the first packet
+ * Returns 1 if it's certain that the first packet is truly the first packet
  * rather than a best guess based upon threads that have published so far.
  * Otherwise 0 is returned.
@@ -791 +793 @@
                 *packet = libtrace->first_packets.packets[libtrace->first_packets.first].packet;
                 *tv = &libtrace->first_packets.packets[libtrace->first_packets.first].tv;
-                if (libtrace->first_packets.count == libtrace->mapper_thread_count) {
+                if (libtrace->first_packets.count == libtrace->perpkt_thread_count) {
                         ret = 1;
                 } else {
@@ -944 +946 @@
         } else if (trace_has_dedicated_hasher(libtrace)) {
                 ret = trace_pread_packet_hasher_thread(libtrace, packet);
-        } else if (libtrace->reducer_flags & MAPPER_USE_SLIDING_WINDOW) {
+        } else if (libtrace->reducer_flags & PERPKT_USE_SLIDING_WINDOW) {
                 ret = trace_pread_packet_sliding_window(libtrace, packet);
         } else {
@@ -967 +969 @@
         int i;
 
-        for (i = 0; i < libtrace->mapper_thread_count; i++) {
-                libtrace_thread_t *t = &libtrace->mapper_threads[i];
-                assert(pthread_create(&t->tid, NULL, mapper_start, (void *) libtrace) == 0);
-        }
-        return libtrace->mapper_thread_count;
+        for (i = 0; i < libtrace->perpkt_thread_count; i++) {
+                libtrace_thread_t *t = &libtrace->perpkt_threads[i];
+                assert(pthread_create(&t->tid, NULL, perpkt_threads_entry, (void *) libtrace) == 0);
+        }
+        return libtrace->perpkt_thread_count;
 }
 
@@ -988 +990 @@
         if (trace_is_err(libtrace))
                 return -1;;
-        if (libtrace->perpkt_pausing != 0) {
+        if (libtrace->perpkts_pausing != 0) {
                 printf("Restarting trace\n");
                 libtrace->format->pstart_input(libtrace);
@@ -1005 +1007 @@
         libtrace->per_pkt = per_pkt;
         libtrace->reducer = reducer;
-        libtrace->mappers_finishing = 0;
+        libtrace->perpkts_finishing = 0;
         // libtrace->hasher = &rand_hash; /* Hasher now set via option */
 
@@ -1014 +1016 @@
 
         // Set default buffer sizes
-        if (libtrace->mapper_buffer_size <= 0)
-                libtrace->mapper_buffer_size = 1000;
-
-        if (libtrace->mapper_thread_count <= 0)
-                libtrace->mapper_thread_count = 2; // XXX scale to system
+        if (libtrace->perpkt_buffer_size <= 0)
+                libtrace->perpkt_buffer_size = 1000;
+
+        if (libtrace->perpkt_thread_count <= 0)
+                libtrace->perpkt_thread_count = 2; // XXX scale to system
 
         if(libtrace->packet_freelist_size <= 0)
-                libtrace->packet_freelist_size = (libtrace->mapper_buffer_size + 1) * libtrace->mapper_thread_count;
+                libtrace->packet_freelist_size = (libtrace->perpkt_buffer_size + 1) * libtrace->perpkt_thread_count;
 
         if(libtrace->packet_freelist_size <
-                (libtrace->mapper_buffer_size + 1) * libtrace->mapper_thread_count)
+                (libtrace->perpkt_buffer_size + 1) * libtrace->perpkt_thread_count)
                 fprintf(stderr, "WARNING deadlocks may occur and extra memory allocating buffer sizes (packet_freelist_size) mismatched\n");
 
@@ -1062 +1064 @@
         libtrace->first_packets.count = 0;
         assert(pthread_spin_init(&libtrace->first_packets.lock, 0) == 0);
-        libtrace->first_packets.packets = calloc(libtrace->mapper_thread_count, sizeof(struct  __packet_storage_magic_type));
-
-
-        /* Start all of our mapper threads */
-        libtrace->mapper_threads = calloc(sizeof(libtrace_thread_t), libtrace->mapper_thread_count);
-        for (i = 0; i < libtrace->mapper_thread_count; i++) {
-                libtrace_thread_t *t = &libtrace->mapper_threads[i];
+        libtrace->first_packets.packets = calloc(libtrace->perpkt_thread_count, sizeof(struct  __packet_storage_magic_type));
+
+
+        /* Start all of our perpkt threads */
+        libtrace->perpkt_threads = calloc(sizeof(libtrace_thread_t), libtrace->perpkt_thread_count);
+        for (i = 0; i < libtrace->perpkt_thread_count; i++) {
+                libtrace_thread_t *t = &libtrace->perpkt_threads[i];
                 t->trace = libtrace;
                 t->ret = NULL;
-                t->type = THREAD_MAPPER;
+                t->type = THREAD_PERPKT;
                 t->state = THREAD_RUNNING;
                 t->user_data = NULL;
                 // t->tid DONE on create
-                t->map_num = i;
+                t->perpkt_num = i;
                 if (libtrace->hasher)
-                        libtrace_ringbuffer_init(&t->rbuffer, libtrace->mapper_buffer_size, LIBTRACE_RINGBUFFER_POLLING);
+                        libtrace_ringbuffer_init(&t->rbuffer, libtrace->perpkt_buffer_size, LIBTRACE_RINGBUFFER_POLLING);
                 // Depending on the mode vector or deque might be chosen
                 libtrace_vector_init(&t->vector, sizeof(libtrace_result_t));
@@ -1091 +1093 @@
         int threads_started = 0;
         /* Setup the trace and start our threads */
-        if (libtrace->mapper_thread_count > 1 && libtrace->format->pstart_input) {
+        if (libtrace->perpkt_thread_count > 1 && libtrace->format->pstart_input) {
                 printf("This format has direct support for p's\n");
                 threads_started = libtrace->format->pstart_input(libtrace);
@@ -1112 +1114 @@
 
         // TODO fix these leaks etc
-        if (libtrace->mapper_thread_count != threads_started)
-                printf("Warning started threads not equal requested s=%d r=%d", threads_started, libtrace->mapper_thread_count);
+        if (libtrace->perpkt_thread_count != threads_started)
+                printf("Warning started threads not equal requested s=%d r=%d", threads_started, libtrace->perpkt_thread_count);
 
 
@@ -1124 +1126 @@
  * 2. All perpkt threads are paused waiting on a condition var
  * 3. Then call ppause on the underlying format if found
- * 4. Return with perpkt_pausing set to mapper_count (Used when restarting so we reuse the threads)
+ * 4. Return with perpkts_pausing set to perpkt_count (Used when restarting so we reuse the threads)
  * 
  * Once done you should be a able to modify the trace setup and call pstart again
@@ -1148 +1150 @@
 
         printf("Sending messages \n");
-        // Stop threads, skip this one if its a mapper
-        for (i = 0; i < libtrace->mapper_thread_count; i++) {
-                if (&libtrace->mapper_threads[i] != t) {
+        // Stop threads, skip this one if its a perpkt
+        for (i = 0; i < libtrace->perpkt_thread_count; i++) {
+                if (&libtrace->perpkt_threads[i] != t) {
                         libtrace_message_t message;
                         message.code = MESSAGE_PAUSE;
                         message.additional = NULL;
-                        trace_send_message_to_thread(libtrace, &libtrace->mapper_threads[i], &message);
+                        trace_send_message_to_thread(libtrace, &libtrace->perpkt_threads[i], &message);
                 }
         }
@@ -1163 +1165 @@
 
         if (t) {
-                // A mapper is doing the pausing interesting fake a extra thread paused
+                // A perpkt is doing the pausing interesting fake a extra thread paused
                 // We rely on the user to not return before starting the trace again
                 assert(pthread_mutex_lock(&libtrace->libtrace_lock) == 0);
-                libtrace->perpkt_pausing++;
+                libtrace->perpkts_pausing++;
                 assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
         }
@@ -1178 +1180 @@
         // Wait for all threads to pause
         assert(pthread_mutex_lock(&libtrace->libtrace_lock) == 0);
-        while (libtrace->mapper_thread_count != libtrace->perpkt_pausing) {
+        while (libtrace->perpkt_thread_count != libtrace->perpkts_pausing) {
                 assert(pthread_cond_wait(&libtrace->perpkt_cond, &libtrace->libtrace_lock) == 0);
         }
@@ -1222 +1224 @@
         // Now send a message asking the threads to stop
         // This will be retrieved before trying to read another packet
-        for (i = 0; i < libtrace->mapper_thread_count; i++) {
+        for (i = 0; i < libtrace->perpkt_thread_count; i++) {
                 libtrace_message_t message;
                 message.code = MESSAGE_STOP;
                 message.additional = NULL;
-                trace_send_message_to_thread(libtrace, &libtrace->mapper_threads[i], &message);
+                trace_send_message_to_thread(libtrace, &libtrace->perpkt_threads[i], &message);
         }
 
@@ -1303 +1305 @@
         int i;
 
-        /* Firstly wait for the mapper threads to finish, since these are
+        /* Firstly wait for the perpkt threads to finish, since these are
          * user controlled */
-        for (i=0; i< libtrace->mapper_thread_count; i++) {
-                //printf("Waiting to join with mapper #%d\n", i);
-                assert(pthread_join(libtrace->mapper_threads[i].tid, NULL) == 0);
-                //printf("Joined with mapper #%d\n", i);
+        for (i=0; i< libtrace->perpkt_thread_count; i++) {
+                //printf("Waiting to join with perpkt #%d\n", i);
+                assert(pthread_join(libtrace->perpkt_threads[i].tid, NULL) == 0);
+                //printf("Joined with perpkt #%d\n", i);
                 // So we must do our best effort to empty the queue - so
                 // the producer (or any other threads) don't block.
@@ -1314 +1316 @@
                 // Mark that we are no longer accepting packets
                 assert(pthread_mutex_lock(&libtrace->libtrace_lock) == 0);
-                libtrace->mapper_threads[i].state = THREAD_FINISHED; // Important we are finished before we empty the buffer
+                libtrace->perpkt_threads[i].state = THREAD_FINISHED; // Important we are finished before we empty the buffer
                 assert(pthread_mutex_unlock(&libtrace->libtrace_lock) == 0);
-                while(libtrace_ringbuffer_try_read(&libtrace->mapper_threads[i].rbuffer, (void **) &packet))
-                        if (packet) // This could be NULL iff the mapper finishes early
+                while(libtrace_ringbuffer_try_read(&libtrace->perpkt_threads[i].rbuffer, (void **) &packet))
+                        if (packet) // This could be NULL iff the perpkt finishes early
                                 trace_destroy_packet(packet);
         }
@@ -1332 +1334 @@
         // Now that everything is finished nothing can be touching our
         // buffers so clean them up
-        for (i = 0; i < libtrace->mapper_thread_count; i++) {
+        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
                 // if they lost timeslice before-during a write
                 libtrace_packet_t * packet;
-                while(libtrace_ringbuffer_try_read(&libtrace->mapper_threads[i].rbuffer, (void **) &packet))
+                while(libtrace_ringbuffer_try_read(&libtrace->perpkt_threads[i].rbuffer, (void **) &packet))
                         trace_destroy_packet(packet);
                 if (libtrace->hasher) {
-                        assert(libtrace_ringbuffer_is_empty(&libtrace->mapper_threads[i].rbuffer));
-                        libtrace_ringbuffer_destroy(&libtrace->mapper_threads[i].rbuffer);
+                        assert(libtrace_ringbuffer_is_empty(&libtrace->perpkt_threads[i].rbuffer));
+                        libtrace_ringbuffer_destroy(&libtrace->perpkt_threads[i].rbuffer);
                 }
                 // Cannot destroy vector yet, this happens with trace_destroy
@@ -1486 +1488 @@
 {
         int this_thread = get_thread_table_num(libtrace); // Could be worth caching ... ?
-        libtrace_thread_t * t = &libtrace->mapper_threads[this_thread];
+        libtrace_thread_t * t = &libtrace->perpkt_threads[this_thread];
 
         assert (pthread_spin_lock(&t->tmp_spinlock) == 0);
@@ -1506 +1508 @@
 {
         int this_thread = get_thread_table_num(libtrace); // Could be worth caching ... ?
-        libtrace_thread_t * t = &libtrace->mapper_threads[this_thread];
+        libtrace_thread_t * t = &libtrace->perpkt_threads[this_thread];
         if (t->tmp_key != key) {
                 if (t->tmp_data) {
@@ -1525 +1527 @@
         // Who am I???
         int this_thread = get_thread_table_num(libtrace); // Could be worth caching ... ?
-        libtrace_thread_t * t = &libtrace->mapper_threads[this_thread];
+        libtrace_thread_t * t = &libtrace->perpkt_threads[this_thread];
         // Now put it into my table
         static __thread int count = 0;
@@ -1578 +1580 @@
         libtrace_vector_empty(results);
         // Check all of the temp queues
-        for (i = 0; i < libtrace->mapper_thread_count; ++i) {
+        for (i = 0; i < libtrace->perpkt_thread_count; ++i) {
                 libtrace_result_t r = {0,0};
-                assert (pthread_spin_lock(&libtrace->mapper_threads[i].tmp_spinlock) == 0);
-                if (libtrace->mapper_threads[i].tmp_key == key) {
-                        libtrace_result_set_key_value(&r, key, libtrace->mapper_threads[i].tmp_data);
-                        libtrace->mapper_threads[i].tmp_data = NULL;
+                assert (pthread_spin_lock(&libtrace->perpkt_threads[i].tmp_spinlock) == 0);
+                if (libtrace->perpkt_threads[i].tmp_key == key) {
+                        libtrace_result_set_key_value(&r, key, libtrace->perpkt_threads[i].tmp_data);
+                        libtrace->perpkt_threads[i].tmp_data = NULL;
                         printf("Found in temp queue\n");
                 }
-                assert (pthread_spin_unlock(&libtrace->mapper_threads[i].tmp_spinlock) == 0);
+                assert (pthread_spin_unlock(&libtrace->perpkt_threads[i].tmp_spinlock) == 0);
                 if (libtrace_result_get_value(&r)) {
                         // Got a result still in temporary
@@ -1593 +1595 @@
                 } else {
                         // This might be waiting on the actual queue
-                        libtrace_queue_t *v = &libtrace->mapper_threads[i].deque;
+                        libtrace_queue_t *v = &libtrace->perpkt_threads[i].deque;
                         if (libtrace_deque_peek_front(v, (void *) &r) &&
                                         libtrace_result_get_value(&r)) {
-                                assert (libtrace_deque_pop_front(&libtrace->mapper_threads[i].deque, (void *) &r) == 1);
+                                assert (libtrace_deque_pop_front(&libtrace->perpkt_threads[i].deque, (void *) &r) == 1);
                                 printf("Found in real queue\n");
                                 libtrace_vector_push_back(results, &r);
@@ -1621 +1623 @@
         if (flags & (REDUCE_SEQUENTIAL | REDUCE_ORDERED)) {
                 int live_count = 0;
-                bool live[libtrace->mapper_thread_count]; // Set if a trace is alive
-                uint64_t key[libtrace->mapper_thread_count]; // Cached keys
+                bool live[libtrace->perpkt_thread_count]; // Set if a trace is alive
+                uint64_t key[libtrace->perpkt_thread_count]; // Cached keys
                 uint64_t min_key = UINT64_MAX; // XXX use max int here stdlimit.h?
                 int min_queue = -1;
 
                 /* Loop through check all are alive (have data) and find the smallest */
-                for (i = 0; i < libtrace->mapper_thread_count; ++i) {
-                        libtrace_queue_t *v = &libtrace->mapper_threads[i].deque;
+                for (i = 0; i < libtrace->perpkt_thread_count; ++i) {
+                        libtrace_queue_t *v = &libtrace->perpkt_threads[i].deque;
                         if (libtrace_deque_get_size(v) != 0) {
                                 libtrace_result_t r;
@@ -1645 +1647 @@
 
                 /* Now remove the smallest and loop - special case if all threads have joined we always flush whats left */
-                while ((live_count == libtrace->mapper_thread_count) || (live_count &&
+                while ((live_count == libtrace->perpkt_thread_count) || (live_count &&
                                 ((flags & REDUCE_SEQUENTIAL && min_key == libtrace->expected_key) ||
                                 libtrace->joined))) {
@@ -1651 +1653 @@
                         libtrace_result_t r;
 
-                        assert (libtrace_deque_pop_front(&libtrace->mapper_threads[min_queue].deque, (void *) &r) == 1);
+                        assert (libtrace_deque_pop_front(&libtrace->perpkt_threads[min_queue].deque, (void *) &r) == 1);
                         libtrace_vector_push_back(results, &r);
 
@@ -1658 +1660 @@
 
                         // Now update the one we just removed
-                        if (libtrace_deque_get_size(&libtrace->mapper_threads[min_queue].deque) )
+                        if (libtrace_deque_get_size(&libtrace->perpkt_threads[min_queue].deque) )
                         {
-                                libtrace_deque_peek_front(&libtrace->mapper_threads[min_queue].deque, (void *) &r);
+                                libtrace_deque_peek_front(&libtrace->perpkt_threads[min_queue].deque, (void *) &r);
                                 key[min_queue] = libtrace_result_get_key(&r);
                                 if (key[min_queue] <= min_key) {
@@ -1668 +1670 @@
                                         min_key = key[min_queue]; // Update our minimum
                                         // Check all find the smallest again - all are alive
-                                        for (i = 0; i < libtrace->mapper_thread_count; ++i) {
+                                        for (i = 0; i < libtrace->perpkt_thread_count; ++i) {
                                                 if (live[i] && min_key > key[i]) {
                                                         min_key = key[i];
@@ -1680 +1682 @@
                                 min_key = UINT64_MAX; // Update our minimum
                                 // Check all find the smallest again - all are alive
-                                for (i = 0; i < libtrace->mapper_thread_count; ++i) {
+                                for (i = 0; i < libtrace->perpkt_thread_count; ++i) {
                                         // Still not 100% TODO (what if order is wrong or not increasing)
                                         if (live[i] && min_key >= key[i]) {
@@ -1690 +1692 @@
                 }
         } else { // Queues are not in order - return all results in the queue
-                for (i = 0; i < libtrace->mapper_thread_count; i++) {
-                        libtrace_vector_append(results, &libtrace->mapper_threads[i].vector);
+                for (i = 0; i < libtrace->perpkt_thread_count; i++) {
+                        libtrace_vector_append(results, &libtrace->perpkt_threads[i].vector);
                 }
                 if (flags & REDUCE_SORT) {
@@ -1721 +1723 @@
         // TODO I don't like using this so much
         //assert(pthread_mutex_lock(&libtrace->libtrace_lock) == 0);
-        for (i = 0; i < libtrace->mapper_thread_count; i++) {
-                if (libtrace->mapper_threads[i].state == THREAD_RUNNING)
+        for (i = 0; i < libtrace->perpkt_thread_count; i++) {
+                if (libtrace->perpkt_threads[i].state == THREAD_RUNNING)
                         b++;
         }
@@ -1735 +1737 @@
                 case TRACE_OPTION_SET_HASHER:
                         return trace_set_hasher(libtrace, (enum hasher_types) *((int *) value), NULL, NULL);
-                case TRACE_OPTION_SET_MAPPER_BUFFER_SIZE:
-                        libtrace->mapper_buffer_size = *((int *) value);
+                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_MAPPER_THREAD_COUNT:
-                        libtrace->mapper_thread_count = *((int *) value);
+                case TRACE_OPTION_SET_PERPKT_THREAD_COUNT:
+                        libtrace->perpkt_thread_count = *((int *) value);
                         return 1;
                 case TRACE_DROP_OUT_OF_ORDER:
@@ -1770 +1772 @@
                 case TRACE_OPTION_USE_SLIDING_WINDOW_BUFFER:
                         if (*((int *) value))
-                                libtrace->reducer_flags |= MAPPER_USE_SLIDING_WINDOW;
+                                libtrace->reducer_flags |= PERPKT_USE_SLIDING_WINDOW;
                         else
-                                libtrace->reducer_flags &= ~MAPPER_USE_SLIDING_WINDOW;
+                                libtrace->reducer_flags &= ~PERPKT_USE_SLIDING_WINDOW;
                         return 1;
                 case TRACE_OPTION_TRACETIME: