Changeset 2498008 for tools/tracertstats/tracertstats_parallel.c

Timestamp:

09/16/14 02:35:10 (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:

d994324

Parents:

50b1bee

Message:

Refactor the combining step to allow user defined functions here.

Remove the old trace_get_results, now instead simply provide a reporter function which gets called as soon as results are ready.
The combiner function used determines the order of these results and when they are released etc.
The combiner function can be selected from those built-in or a custom version can be defined results are provided when ready.
Quickly hacked the parallel tests to work with this update, these are still a bit messy.

Also some fixes some compile warnings.

File:

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

tools/tracertstats/tracertstats_parallel.c

@@ -136 +136 @@
 } result_t;
 
-
-static int reduce(libtrace_t* trace, void* global_blob, uint64_t *last_ts)
-{       
-        int i,j;
-        //uint64_t count=0, bytes=0;
+static uint64_t last_ts = 0;
+static void process_result(libtrace_t *trace UNUSED, libtrace_result_t *result, libtrace_message_t *mesg UNUSED)  {
         static uint64_t ts = 0;
-        libtrace_vector_t results;
-        libtrace_vector_init(&results, sizeof(libtrace_result_t));
-        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;
-                // Memory manager might falsely trigger this
-                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);
+
+        if (result) {
+                int j;
+                result_t *res;
+                ts = libtrace_result_get_key(result);
+                res = libtrace_result_get_value(result);
+                if (last_ts == 0)
+                        last_ts = ts;
+                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)++;
-                }
-                
+                        last_ts++;
+                }
                 count += res->total.count;
                 bytes += res->total.bytes;
@@ -179 +163 @@
                 free(res);
         }
-        // Done with these results - Free internally and externally
-        libtrace_vector_destroy(&results);
-        
-        return 0;
 }
 
@@ -276 +256 @@
 }
 
-static uint64_t bad_hash(const libtrace_packet_t * pkt, void *data) {
+static uint64_t bad_hash(const libtrace_packet_t * pkt UNUSED, void *data UNUSED) {
         return 0;
 }
@@ -284 +264 @@
 {
         int j;
-        uint64_t last_ts = 0;
 
         if (!merge_inputs) 
@@ -318 +297 @@
         }
 
-        if (trace_pstart(trace, NULL, &per_packet, NULL)==-1) {
+        if (trace_pstart(trace, NULL, &per_packet, process_result)==-1) {
                 trace_perror(trace,"Failed to start trace");
                 trace_destroy(trace);
@@ -327 +306 @@
 
 
-        // reduce
-        while (!trace_finished(trace)) {
-                // Read messages
-                libtrace_message_t message;
-                
-                // We just release and do work currently, maybe if something
-                // interesting comes through we'd deal with that
-                libtrace_thread_get_message(trace, &message);
-                
-                while (libtrace_thread_try_get_message(trace, &message) != LIBTRACE_MQ_FAILED) { }
-                reduce(trace, NULL, &last_ts);
-        }
-
         // Wait for all threads to stop
         trace_join(trace);
         
-        reduce(trace, NULL, &last_ts);
         // Flush the last one out
         report_results((double) last_ts * (double) packet_interval, count, bytes);