Changeset 82facc5 for tools

Timestamp:

06/04/14 02:28:58 (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:

b13b939

Parents:

fac8c46

Message:

Adds a thread keepalive that sends a messages to the perpkt threads every second(still todo make this configurable)
Updated tracertstats to use this rather than the temporary result system which has been removed
Also fixes a handful of compile warnings

File:

• tools/tracertstats/tracertstats_parallel.c (modified) (7 diffs)

tools/tracertstats/tracertstats_parallel.c

@@ -65 +65 @@
 
 #define DEFAULT_OUTPUT_FMT "txt"
-#define TRACE_TIME 1
 
 struct libtrace_t *trace;
@@ -190 +189 @@
 } timestamp_sync_t;
 
-
-static int reduce_tracetime(libtrace_t* trace, void* global_blob, uint64_t *last_ts)
-{
-        int i,j;
-        //uint64_t count=0, bytes=0;
-        static uint64_t ts = 0;
-        libtrace_vector_t results;
-        libtrace_vector_init(&results, sizeof(libtrace_result_t));
-        trace_get_results_check_temp(trace, &results, *last_ts);
-        //trace_get_results(trace, &results);
-        //uint64_t packets;
-        
-        /* Get the results from each core and sum 'em up */
-        for (i = 0 ; i < libtrace_vector_get_size(&results) ; i++) {
-                libtrace_result_t result;
-                
-                assert(libtrace_vector_get(&results, i, (void *) &result) == 1);
-                ts = libtrace_result_get_key(&result);
-                if (*last_ts == 0)
-                        *last_ts = ts;
-                
-                result_t * res = libtrace_result_get_value(&result);
-                static result_t *  last_res = NULL;
-                if (res == last_res) {
-                        printf("Hmm could be asserting but I'm not ;)\n");
-                }
-                //assert(res != last_res);
-                last_res = res;
-                //printf("Perpkt published %"PRIu64" - c=%"PRIu64"\n", ts, res->total.count);
-                /*while (*last_ts < ts) {
-                        report_results((double) *last_ts * (double) packet_interval, count, bytes);
-                        count = 0;
-                        bytes = 0;
-                        for (j = 0; j < filter_count; j++)
-                                filters[j].count = filters[j].bytes = 0;
-                        (*last_ts)++;
-                }*/
-                
-                count += res->total.count;
-                bytes += res->total.bytes;
-                for (j = 0; j < filter_count; j++) {
-                        filters[j].count += res->filters[j].count;
-                        filters[j].bytes += res->filters[j].bytes;
-                }
-                free(res);
-        }
-        report_results((double) *last_ts * (double) packet_interval, count, bytes);
-        count = 0;
-        bytes = 0;
-        for (j = 0; j < filter_count; j++)
-                filters[j].count = filters[j].bytes = 0;
-        (*last_ts)++;
-        
-        // Done with these results - Free internally and externally
-        libtrace_vector_destroy(&results);
-        
-        return 0;
-}
-
 static void* per_packet(libtrace_t *trace, libtrace_packet_t *pkt, 
                                                 libtrace_message_t *mesg,
@@ -259 +199 @@
         // Unsure when we would hit this case but the old code had it, I 
         // guess we should keep it
-        if (pkt && trace_get_packet_buffer(pkt,NULL,NULL) == NULL) {
-                
+        if (pkt && trace_get_packet_buffer(pkt,NULL,NULL) != NULL) {
+                //fprintf(stderr, "Got packet t=%x\n", t);
                 ts = trace_get_seconds(pkt) / packet_interval;
                 if (last_ts == 0)
                         last_ts = ts;
-                
+
                 while (packet_interval != UINT64_MAX && last_ts<ts) {
                         // 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);
@@ -296 +237 @@
                                 break;
                         case MESSAGE_STOPPED:
+                                // Should we always post this?
                                 if (results->total.count) {
                                         trace_publish_result(trace, (uint64_t) last_ts, results);
                                         trace_post_reduce(trace);
+                                        results = NULL;
                                 }
+                                break;
+                        case MESSAGE_TICK:
+                        {
+                                int64_t offset;
+                                struct timeval *tv, tv_real;
+                                libtrace_packet_t *first_packet = NULL;
+                                retrive_first_packet(trace, &first_packet, &tv);
+                                if (first_packet != NULL) {
+                                        // So figure out our running offset
+                                        tv_real = trace_get_timeval(first_packet);
+                                        offset = tv_to_usec(tv) - tv_to_usec(&tv_real);
+                                        // Get time of day and do this stuff
+                                        uint64_t next_update_time;
+                                        next_update_time = (last_ts*packet_interval + packet_interval) * 1000000 + offset;
+                                        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);
+                                                results = calloc(1, sizeof(result_t) + sizeof(statistic_t) * filter_count);
+                                                last_ts++;
+                                        } else {
+                                                fprintf(stderr, "Got a tick but no publish ...\n");
+                                        }
+                                } else {
+                                        fprintf(stderr, "Got a tick but no packets seen yet!!!\n");
+                                }
+                        }
                 }
         }
         return pkt;
 }
-void * trace_retrive_inprogress_result(libtrace_t *libtrace, uint64_t key);
-/**
- * A trace time version of map which will attempt to keep upto date 
- * with the incoming data and detect cases where results are missing and
- * recover correctly.
- */
-static void* per_packet_tracetime(libtrace_t *trace, libtrace_packet_t *pkt, 
-                                                libtrace_message_t *mesg,
-                                                libtrace_thread_t *t)
-{
-        // Using first entry as total and those after for filter counts
-        int i;
-        static __thread uint64_t last_ts = 0, ts = 0;
-        static __thread double debug_last = 0.0;
-        static __thread result_t * tmp_result = NULL;
-        
-        if (pkt && trace_get_packet_buffer(pkt,NULL,NULL) != NULL) {
-                ts = trace_get_seconds(pkt) / packet_interval;
-                
-                if (debug_last != 0.0 && debug_last > trace_get_seconds(pkt))
-                        printf("packets out of order bitch :(\n");
-                debug_last = trace_get_seconds(pkt);
-                if (last_ts == 0)
-                        last_ts = ts;
-                
-                /*
-                while (packet_interval != UINT64_MAX && last_ts<ts) {
-                        // Publish and make new
-                        trace_publish_result(trace, (uint64_t) last_ts, results);
-                        results = malloc(sizeof(result_t) + sizeof(statistic_t) * filter_count);
-                        memset(results, 0, sizeof(result_t) + sizeof(statistic_t) * filter_count);
-                        last_ts++;
-                }*/
-                
-                /* Calculate count for filters */
-                for(i=0;i<filter_count;++i) {
-                        if(trace_apply_filter(filters[i].filter, pkt)) {
-                                tmp_result->filters[i].count = 1;
-                                tmp_result->filters[i].bytes = trace_get_wire_length(pkt);
-                        } else {
-                                tmp_result->filters[i].count = 0;
-                                tmp_result->filters[i].bytes = 0;
-                        }
-                }
-                
-                /* Now Update the currently stored result */
-                result_t * results = (result_t *) trace_retrive_inprogress_result(trace, ts);
-                
-                if (!results) {
-                        results = malloc(sizeof(result_t) + sizeof(statistic_t) * filter_count);
-                        memset(results, 0, sizeof(result_t) + sizeof(statistic_t) * filter_count);
-                }
-                assert(results);
-                /* Now add to the current results */
-                results->total.count++;
-                results->total.bytes +=trace_get_wire_length(pkt);
-                /* Now add on filters */
-                for(i=0;i<filter_count;++i) {
-                        results->filters[i].count += tmp_result->filters[i].count;
-                        results->filters[i].bytes += tmp_result->filters[i].bytes;
-                }
-                /* Now release the lock and send it away place that back into the buffer */
-                trace_update_inprogress_result(trace, ts, (void *) results);
-                /*if (count >= packet_count) {
-                        report_results(ts,count,bytes);
-                        count=0;
-                        bytes=0;
-                }*/ // Hmm what was happening here doesn't match up with any of the documentations!!!
-        }
-        if (mesg) {
-                // printf ("%d.%06d READ #%"PRIu64"\n", tv.tv_sec, tv.tv_usec, trace_packet_get(packet));
-                switch (mesg->code) {
-                        case MESSAGE_STARTED:
-                                tmp_result = calloc(1, sizeof(result_t) + sizeof(statistic_t) * filter_count);
-                                break;
-                        case MESSAGE_STOPPED:
-                                trace_retrive_inprogress_result(trace, 0);
-                                trace_update_inprogress_result(trace, 1, NULL);
-                }
-        }
-        // Done push the final results
-        /*if (results->total.count)
-                trace_publish_result(trace, (uint64_t) last_ts, results);*/
-        
-        return pkt;
+
+static uint64_t bad_hash(const libtrace_packet_t * pkt, void *data) {
+        return 0;
 }
 
@@ -422 +310 @@
         trace_parallel_config(trace, TRACE_OPTION_ORDERED, &i);
         trace_parallel_config(trace, TRACE_OPTION_TRACETIME, &i);
+        trace_set_hasher(trace, HASHER_CUSTOM, &bad_hash, NULL);
 #if TRACE_TIME
         if (trace_pstart(trace, NULL, &per_packet_tracetime, NULL)==-1) {
@@ -434 +323 @@
         }
 
-#if TRACE_TIME
-        // First we wait for a message telling us that a timestamp has been 
-        // published this allows us to approximately synchronize with the time
-        libtrace_message_t message;
-        int64_t offset;
-        libtrace_packet_t *packet;
-        struct timeval *tv, tv_real;
-        
-        
-        do {
-                // TODO Put a timeout here also
-                libtrace_thread_get_message(trace, &message);
-        } while (retrive_first_packet(trace, &packet, &tv) == 0);
-        tv_real = trace_get_timeval(packet);
-        offset = tv_to_usec(&tv_real) - tv_to_usec(tv);
-        last_ts = trace_get_seconds(packet) / packet_interval;
-        printf("Got first yay offset=%"PRId64" first_interval=%"PRIu64"\n", offset, last_ts);
-        /*
-        while (!got_first) {
-                // Wait for a message indicating we've got our 'first' packet, note not a 100% guarantee its our first but pretty likely 
-                
-                
-                
-                assert(pthread_mutex_lock(&lock_more) == 0);
-                
-                for (i=0; i < 2; ++i) {
-                        if (initial_stamps[i].difference_usecs) { // Hmm certainly this cannot possibly lineup 100%??
-                                got_first=1;
-                                last_ts = initial_stamps[i].first_interval_number;
-                                offset = initial_stamps[i].difference_usecs;
-                                printf("Got first yay offset=%"PRId64" first_interval=%"PRIu64"\n", offset, last_ts);
-                        }
-                }
-                assert(pthread_mutex_unlock(&lock_more) == 0);
-        }*/
-        while (!trace_finished(trace)) {
-                struct timeval tv;
-                // Now try our best to read that one out
-                
-                // Read messages
-                //libtrace_thread_get_message(trace, &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) { }
-                
-                /* Now wait for a second after we should see the results */
-                uint64_t next_update_time, t_usec;
-                next_update_time = (last_ts*packet_interval + packet_interval + 1) * 1000000 + offset;
-                gettimeofday(&tv, NULL);
-                t_usec = tv.tv_sec;
-                t_usec *= 1000000;
-                t_usec += tv.tv_usec;
-                
-                //printf("Current time=%"PRIu64" Next result ready=%"PRIu64" =%f\n", t_usec, next_update_time, ((double) next_update_time - (double) t_usec) / 1000000.0);
-                if (next_update_time > t_usec) {
-                        tv.tv_sec = (next_update_time - t_usec) / 1000000;
-                        tv.tv_usec = (next_update_time - t_usec) % 1000000;
-                        select(0, NULL, NULL, NULL, &tv);
-                }
-                reduce_tracetime(trace, NULL, &last_ts);
-        }
-#else
+
         // reduce
         while (!trace_finished(trace)) {
@@ -511 +336 @@
                 reduce(trace, NULL, &last_ts);
         }
-#endif
 
         // Wait for all threads to stop