Changeset 76291d1

Timestamp:

03/31/15 16:02:53 (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:

4007dbb

Parents:

58bfabf

Message:

Add a quick mockup of shanes interface using functions for each message

I.e. set a function for MESSAGE_PACKET etc. which only receives packets.

Ported tracestats_parallel to use this interface.

Note it is not yet complete, however should give a feel of how it will work.
And provide a good work around to the thread issues on older machines.

Files:

• lib/libtrace_int.h (modified) (1 diff)
• lib/libtrace_parallel.h (modified) (1 diff)
• lib/trace.c (modified) (2 diffs)
• lib/trace_parallel.c (modified) (4 diffs)
• tools/tracestats/tracestats_parallel.c (modified) (3 diffs)

lib/libtrace_int.h

@@ -358 +358 @@
         struct user_configuration config;
         libtrace_combine_t combiner;
+        struct {
+                fn_handler message_starting;
+                fn_handler message_stopping;
+                fn_handler message_resuming;
+                fn_handler message_pausing;
+                fn_handler message_packet;
+        } callbacks;
 };
 

lib/libtrace_parallel.h

@@ -465 +465 @@
 DLLEXPORT int trace_pstart(libtrace_t *libtrace, void* global_blob,
                            fn_per_pkt per_pkt, fn_reporter reporter);
+
+
+/**
+ * @param libtrace The parallel trace
+ * @param t The thread
+ * @param data The data associated with the message
+ * @param global The global storage
+ * @param tls The thread local storage
+ */
+typedef void* (*fn_handler)(libtrace_t *libtrace,
+                               libtrace_thread_t *t,
+                               libtrace_generic_t data,
+                               void *global,
+                               void *tls);
+
+/** Registers a built-in message with a handler.
+ * Note we do not include the sending thread as an argument to the reporter.
+ * If set to NULL, the message will be sent to default perpkt handler.
+ *
+ * @param libtrace The input trace to start
+ * @param message The message to intercept
+ * @param handler the handler to be called when the message is received
+ * @return 0 if successful otherwise -1.
+ */
+DLLEXPORT int trace_set_handler(libtrace_t *libtrace, enum libtrace_messages message, fn_handler handler);
 
 /** Pauses a trace previously started with trace_pstart()

lib/trace.c

@@ -287 +287 @@
         ZERO_USER_CONFIG(libtrace->config);
         memset(&libtrace->combiner, 0, sizeof(libtrace->combiner));
+        memset(&libtrace->callbacks, 0, sizeof(libtrace->callbacks));
 
         /* Parse the URI to determine what sort of trace we are dealing with */
@@ -405 +406 @@
         ZERO_USER_CONFIG(libtrace->config);
         memset(&libtrace->combiner, 0, sizeof(libtrace->combiner));
+        memset(&libtrace->callbacks, 0, sizeof(libtrace->callbacks));
         
         for(tmp=formats_list;tmp;tmp=tmp->next) {

lib/trace_parallel.c

@@ -377 +377 @@
                 t->accepted_packets++;
                 libtrace_generic_t data = {.pkt = *packet};
-                *packet = (*trace->per_pkt)(trace, t, MESSAGE_PACKET, data, t);
+                if (trace->callbacks.message_packet)
+                        *packet = (*trace->callbacks.message_packet)(trace, t, data, trace->global_blob, t->user_data);
+                else
+                        *packet = (*trace->per_pkt)(trace, t, MESSAGE_PACKET, data, t);
                 trace_fin_packet(*packet);
         } else {
@@ -535 +538 @@
 
         /* Let the per_packet function know we have started */
-        (*trace->per_pkt)(trace, t, MESSAGE_STARTING, (libtrace_generic_t){0}, t);
-        (*trace->per_pkt)(trace, t, MESSAGE_RESUMING, (libtrace_generic_t){0}, t);
+        if (trace->callbacks.message_starting)
+                (*trace->callbacks.message_starting)(trace, t, (libtrace_generic_t){0}, trace->global_blob, t->user_data);
+        else
+                (*trace->per_pkt)(trace, t, MESSAGE_STARTING, (libtrace_generic_t){0}, t);
+
+        if (trace->callbacks.message_resuming)
+                (*trace->callbacks.message_resuming)(trace, t, (libtrace_generic_t){0}, trace->global_blob, t->user_data);
+        else
+                (*trace->per_pkt)(trace, t, MESSAGE_RESUMING, (libtrace_generic_t){0}, t);
 
         for (;;) {
@@ -619 +629 @@
 
         // Let the per_packet function know we have stopped
-        (*trace->per_pkt)(trace, t, MESSAGE_PAUSING, (libtrace_generic_t){0}, t);
-        (*trace->per_pkt)(trace, t, MESSAGE_STOPPING, (libtrace_generic_t){0}, t);
+        if (trace->callbacks.message_pausing)
+                (*trace->callbacks.message_pausing)(trace, t, (libtrace_generic_t){0}, trace->global_blob, t->user_data);
+        else
+                (*trace->per_pkt)(trace, t, MESSAGE_PAUSING, (libtrace_generic_t){0}, t);
+        if (trace->callbacks.message_stopping)
+                (*trace->callbacks.message_stopping)(trace, t, (libtrace_generic_t){0}, trace->global_blob, t->user_data);
+        else
+                (*trace->per_pkt)(trace, t, MESSAGE_STOPPING, (libtrace_generic_t){0}, t);
+
 
         // Free any remaining packets
@@ -1719 +1736 @@
         return ret;
 }
+
+DLLEXPORT int trace_set_handler(libtrace_t *libtrace, enum libtrace_messages message, fn_handler handler) {
+        switch (message) {
+        case MESSAGE_STARTING:
+                libtrace->callbacks.message_starting = handler;
+                return 0;
+        case MESSAGE_STOPPING:
+                libtrace->callbacks.message_stopping = handler;
+                return 0;
+        case MESSAGE_RESUMING:
+                libtrace->callbacks.message_resuming = handler;
+                return 0;
+        case MESSAGE_PAUSING:
+                libtrace->callbacks.message_pausing = handler;
+                return 0;
+        case MESSAGE_PACKET:
+                libtrace->callbacks.message_packet = handler;
+                return 0;
+        default:
+                return -1;
+        }
+        return -1;
+}
+
 
 /*

tools/tracestats/tracestats_parallel.c

@@ -87 +87 @@
 
 
-static void* per_packet(libtrace_t *trace, libtrace_thread_t *t,
-                        int mesg, libtrace_generic_t data,
+static void* per_packet(libtrace_t *trace UNUSED, libtrace_thread_t *t UNUSED,
+                        int mesg UNUSED, libtrace_generic_t data UNUSED,
                         libtrace_thread_t *sender UNUSED)
 {
-        /* Using first entry as total and those after for filter counts */
-        static __thread statistics_t * results = NULL;
-        int i, wlen;
-        libtrace_stat_t *stats;
-        libtrace_generic_t gen;
-
-        switch (mesg) {
-        case MESSAGE_PACKET:
-                /* Apply filters to every packet note the result */
-                wlen = trace_get_wire_length(data.pkt);
-                for(i=0;i<filter_count;++i) {
-                        if (filters[i].filter == NULL)
-                                continue;
-                        if(trace_apply_filter(filters[i].filter,data.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 data.pkt;
-        case MESSAGE_STARTING:
-                /* Allocate space to hold a total count and one for each filter */
-                results = calloc(1, sizeof(statistics_t) * (filter_count + 1));
-                break;
-        case MESSAGE_STOPPING:
-                /* 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);
-                break;
-        default:
-                break;
-        }
         return NULL;
 }
@@ -184 +144 @@
 }
 
+
+static void* fn_starting(libtrace_t *trace UNUSED, libtrace_thread_t *t,
+                     libtrace_generic_t data UNUSED, void *global UNUSED, void*tls UNUSED) {
+        /* Allocate space to hold a total count and one for each filter */
+        statistics_t *results = calloc(1, sizeof(statistics_t) * (filter_count + 1));
+        trace_set_tls(t, results);
+        return NULL;
+}
+
+
+static void* fn_stopping(libtrace_t *trace, libtrace_thread_t *t UNUSED,
+                     libtrace_generic_t data UNUSED, void *global UNUSED, void*tls) {
+        statistics_t *results = 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);
+        return NULL;
+}
+
+static void* fn_packet(libtrace_t *trace, libtrace_thread_t *t UNUSED,
+                   libtrace_generic_t data, void *global UNUSED, void*tls) {
+        statistics_t *results = tls;
+        int i, wlen;
+
+        /* Apply filters to every packet note the result */
+        wlen = trace_get_wire_length(data.pkt);
+        for(i=0;i<filter_count;++i) {
+                if (filters[i].filter == NULL)
+                        continue;
+                if(trace_apply_filter(filters[i].filter,data.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 data.pkt;
+}
+
 /* Process a trace, counting packets that match filter(s) */
 static void run_trace(char *uri, char *config, char *config_file)
@@ -212 +219 @@
                 }
         }
+
+        trace_set_handler(trace, MESSAGE_PACKET, fn_packet);
+        trace_set_handler(trace, MESSAGE_STARTING, fn_starting);
+        trace_set_handler(trace, MESSAGE_STOPPING, fn_stopping);
 
         /* Start the trace as a parallel trace */