Changeset 322c516 for tools

Timestamp:

09/11/15 17:15:22 (5 years ago)

Author:

Shane Alcock <salcock@…>

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:

b3dca4d

Parents:

f625817

Message:

Port tracestats to new parallel API

Old single-threaded tracestats has been moved to tracestats_single.c, which
won't be built or distributed.

Location:

tools/tracestats

Files:

• tracestats.c (modified) (8 diffs)
• tracestats_parallel.c (deleted)
• tracestats_single.c (added)

tools/tracestats/tracestats.c

@@ -53 +53 @@
 #include <signal.h>
 
-#include "libtrace.h"
+#include "libtrace_parallel.h"
 #include "lt_inttypes.h"
-
-struct libtrace_t *trace;
-
-static volatile int done=0;
-
-static void cleanup_signal(int signal)
+#include <pthread.h>
+
+struct libtrace_t *trace = NULL;
+
+static void cleanup_signal(int signal UNUSED)
 {
-        (void)signal;
-        done=1;
-        trace_interrupt();
+        if (trace)
+                trace_pstop(trace);
 }
 
@@ -70 +68 @@
         char *expr;
         struct libtrace_filter_t *filter;
+} *filters = NULL;
+
+int filter_count=0;
+
+
+typedef struct statistics {
         uint64_t count;
         uint64_t bytes;
-} *filters = NULL;
-int filter_count=0;
-uint64_t totcount;
-uint64_t totbytes;
+} statistics_t;
+
+volatile uint64_t totcount = 0;
+volatile uint64_t totbytes = 0;
+
+
+static void fn_result(libtrace_t *trace UNUSED,
+                libtrace_thread_t *sender UNUSED,
+                void *global UNUSED, void *tls,
+                libtrace_result_t *result) {
+        statistics_t *counters = (statistics_t *)tls;
+        int i;
+
+        assert(result->key == 0);
+        statistics_t * res = result->value.ptr;
+        counters[0].count += res[0].count;
+        counters[0].bytes += res[0].bytes;
+
+        for (i = 0; i < filter_count; i++) {
+                counters[i+1].count += res[i+1].count;
+                counters[i+1].bytes += res[i+1].bytes;
+        }
+        free(res);
+}
+
+static void fn_print_results(libtrace_t *trace, 
+                libtrace_thread_t *sender UNUSED, void *global UNUSED,
+                void *tls) {
+
+        statistics_t *counters = (statistics_t *)tls;
+        libtrace_stat_t *stats = NULL;
+        int i;
+
+        stats = trace_get_statistics(trace, NULL);
+        printf("%-30s\t%12s\t%12s\t%7s\n","filter","count","bytes","%");
+        for(i=0;i<filter_count;++i) {
+                printf("%30s:\t%12"PRIu64"\t%12"PRIu64"\t%7.03f\n",filters[i].expr,counters[i+1].count,counters[i+1].bytes,counters[i+1].count*100.0/counters[0].count);
+        }
+        if (stats->received_valid)
+                fprintf(stderr,"%30s:\t%12" PRIu64"\n",
+                                "Input packets", stats->received);
+        if (stats->filtered_valid)
+                fprintf(stderr,"%30s:\t%12" PRIu64"\n",
+                                "Filtered packets", stats->filtered);
+        if (stats->dropped_valid)
+                fprintf(stderr,"%30s:\t%12" PRIu64"\n",
+                                "Dropped packets",stats->dropped);
+        if (stats->accepted_valid)
+                fprintf(stderr,"%30s:\t%12" PRIu64 "\n",
+                                "Accepted packets", stats->accepted);
+        if (stats->errors_valid)
+                fprintf(stderr,"%30s:\t%12" PRIu64 "\n",
+                                "Erred packets", stats->errors);
+        printf("%30s:\t%12"PRIu64"\t%12" PRIu64 "\n","Total",counters[0].count,counters[0].bytes);
+        totcount+=counters[0].count;
+        totbytes+=counters[0].bytes;
+
+}
+
+
+static void* fn_starting(libtrace_t *trace UNUSED, libtrace_thread_t *t UNUSED, void *global UNUSED) {
+        /* Allocate space to hold a total count and one for each filter */
+        return calloc(1, sizeof(statistics_t) * (filter_count + 1));
+}
+
+
+static void fn_stopping(libtrace_t *trace, libtrace_thread_t *t UNUSED,
+                        void *global UNUSED, void*tls) {
+        statistics_t *results = (statistics_t *)tls;
+        libtrace_generic_t gen;
+        /* We only output one result per thread with the key 0 when the
+         * trace is over. */
+        gen.ptr = results;
+        trace_publish_result(trace, t, 0, gen, RESULT_USER);
+}
+
+static libtrace_packet_t* fn_packet(libtrace_t *trace,
+                libtrace_thread_t *t UNUSED,
+                void *global UNUSED, void*tls, libtrace_packet_t *pkt) {
+        statistics_t *results = (statistics_t *)tls;
+        int i, wlen;
+
+        /* Apply filters to every packet note the result */
+        wlen = trace_get_wire_length(pkt);
+        for(i=0;i<filter_count;++i) {
+                if (filters[i].filter == NULL)
+                        continue;
+                if(trace_apply_filter(filters[i].filter,pkt) > 0) {
+                        results[i+1].count++;
+                        results[i+1].bytes+=wlen;
+                }
+                if (trace_is_err(trace)) {
+                        trace_perror(trace, "trace_apply_filter");
+                        fprintf(stderr, "Removing filter from filterlist\n");
+                        /* This is a race, but will be atomic */
+                        filters[i].filter = NULL;
+                }
+        }
+        results[0].count++;
+        results[0].bytes +=wlen;
+        return pkt;
+}
 
 /* Process a trace, counting packets that match filter(s) */
-static void run_trace(char *uri) 
+static void run_trace(char *uri, int threadcount)
 {
-        struct libtrace_packet_t *packet = trace_create_packet();
-        int i;
-        uint64_t count = 0;
-        uint64_t bytes = 0;
-        libtrace_stat_t *stats;
 
         fprintf(stderr,"%s:\n",uri);
-
-        trace = trace_create(uri);
+        libtrace_callback_set_t *pktcbs, *rescbs;
+
+        trace = trace_create(uri);
 
         if (trace_is_err(trace)) {
@@ -95 +193 @@
         }
 
-        if (trace_start(trace)==-1) {
+        pktcbs = trace_create_callback_set();
+        rescbs = trace_create_callback_set();
+
+        trace_set_packet_cb(pktcbs, fn_packet);
+        trace_set_starting_cb(pktcbs, fn_starting);
+        trace_set_stopping_cb(pktcbs, fn_stopping);
+        trace_set_starting_cb(rescbs, fn_starting);
+        trace_set_result_cb(rescbs, fn_result);
+        trace_set_stopping_cb(rescbs, fn_print_results);
+
+        if (threadcount != 0)
+                trace_set_perpkt_threads(trace, threadcount);
+
+        /* Start the trace as a parallel trace */
+        if (trace_pstart(trace, NULL, pktcbs, rescbs)==-1) {
                 trace_perror(trace,"Failed to start trace");
                 return;
         }
 
-
-        for (;;) {
-                int psize;
-                int wlen;
-                if ((psize = trace_read_packet(trace, packet)) <1) {
-                        break;
-                }
-                
-                if (done)
-                        break;
-                wlen = trace_get_wire_length(packet);
-
-                for(i=0;i<filter_count;++i) {
-                        if (filters[i].filter == NULL)
-                                continue;
-                        if(trace_apply_filter(filters[i].filter,packet) > 0) {
-                                ++filters[i].count;
-                                filters[i].bytes+=wlen;
-                        }
-                        if (trace_is_err(trace)) {
-                                trace_perror(trace, "trace_apply_filter");
-                                fprintf(stderr, "Removing filter from filterlist\n");
-                                filters[i].filter = NULL;
-                        }
-                }
-
-                ++count;
-                bytes+=wlen;
-        }
-        trace_destroy_packet(packet);
-
-        stats = trace_get_statistics(trace, NULL);
-
-        printf("%-30s\t%12s\t%12s\t%7s\n","filter","count","bytes","%");
-        for(i=0;i<filter_count;++i) {
-                printf("%30s:\t%12"PRIu64"\t%12"PRIu64"\t%7.03f\n",filters[i].expr,filters[i].count,filters[i].bytes,filters[i].count*100.0/count);
-                filters[i].bytes=0;
-                filters[i].count=0;
-        }
-        if (stats->received_valid)
-                fprintf(stderr,"%30s:\t%12" PRIu64"\n", 
-                                "Input packets", stats->received);
-        if (stats->filtered_valid)
-                fprintf(stderr,"%30s:\t%12" PRIu64"\n", 
-                                "Filtered packets", stats->filtered);
-        if (stats->dropped_valid)
-                fprintf(stderr,"%30s:\t%12" PRIu64"\n",
-                                "Dropped packets",stats->dropped);
-        if (stats->accepted_valid)
-                fprintf(stderr,"%30s:\t%12" PRIu64 "\n",
-                                "Accepted packets", stats->accepted);
-        if (stats->errors_valid)
-                fprintf(stderr,"%30s:\t%12" PRIu64 "\n",
-                                "Erred packets", stats->errors);
-        printf("%30s:\t%12"PRIu64"\t%12" PRIu64 "\n","Total",count,bytes);
-        totcount+=count;
-        totbytes+=bytes;
+        /* Wait for all threads to stop */
+        trace_join(trace);
 
         if (trace_is_err(trace))
-                trace_perror(trace,"Processing trace");
-
-        trace_destroy(trace);
+                trace_perror(trace,"%s",uri);
+
+        trace_destroy(trace);
+        trace_destroy_callback_set(pktcbs);
+        trace_destroy_callback_set(rescbs);
 }
 
 static void usage(char *argv0)
 {
-        fprintf(stderr,"Usage: %s [-H|--libtrace-help] [--filter|-f bpf ]... libtraceuri...\n",argv0);
+        fprintf(stderr,"Usage: %s [-h|--help] [--threads|-t threads] [--filter|-f bpf ]... libtraceuri...\n",argv0);
 }
 
@@ -173 +232 @@
         int i;
         struct sigaction sigact;
+        int threadcount = 4;
 
         while(1) {
@@ -178 +238 @@
                 struct option long_options[] = {
                         { "filter",        1, 0, 'f' },
-                        { "libtrace-help", 0, 0, 'H' },
+                        { "help", 0, 0, 'h' },
+                        { "threads",            1, 0, 't' },
                         { NULL,            0, 0, 0   },
                 };
 
-                int c=getopt_long(argc, argv, "f:H",
+                int c=getopt_long(argc, argv, "f:ht:",
                                 long_options, &option_index);
 
@@ -189 +250 @@
 
                 switch (c) {
-                        case 'f': 
+                        case 'f':
                                 ++filter_count;
                                 filters=realloc(filters,filter_count*sizeof(struct filter_t));
                                 filters[filter_count-1].expr=strdup(optarg);
                                 filters[filter_count-1].filter=trace_create_filter(optarg);
-                                filters[filter_count-1].count=0;
-                                filters[filter_count-1].bytes=0;
                                 break;
-                        case 'H':
-                                trace_help();
-                                exit(1);
-                                break;
-                        default:
+                        case 'h':
+                                usage(argv[0]);
+                                return 1;
+                        case 't':
+                                threadcount = atoi(optarg);
+                                if (threadcount <= 0)
+                                        threadcount = 1;
+                                break;
+                        default:
                                 fprintf(stderr,"Unknown option: %c\n",c);
                                 usage(argv[0]);
@@ -216 +279 @@
 
         for(i=optind;i<argc;++i) {
-                run_trace(argv[i]);
+                run_trace(argv[i], threadcount);
         }
         if (optind+1<argc) {
@@ -222 +285 @@
                 printf("%30s:\t%12"PRIu64"\t%12" PRIu64 "\n","Total",totcount,totbytes);
         }
-
-
-        return 0;
-}
+        
+        return 0;
+}