Changeset fac8c46 for tools

Timestamp:

05/20/14 03:25:13 (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:

82facc5

Parents:

a5662447

Message:

Tidies up the pausing so that it now works as expected and a trace can easily be paused and restarted.
Ensures that packets will not be lost if pause is called on a file, any queued packets will be read (a message is sent allowing the user to drop these packets if they are unwanted).
Differentiates packets from other results in the queues to the reducer/reporter and makes a copy of the packets in result queues when pausing

• this is needed to ensure that bad memory isn't referenced if a zero-copy trace is paused by closing sockets/associated data like in the case of ring:.

Fixed up the re-starting of traces which hadn't been finished to account for different configurations.
Adds a 'state' to libtrace to handle the state of parallel traces, rather than hacking around the existing 'started' boolean. Also provides two levels of checks for consistency if the trace is using existing that are checking started.

Various other bug fixes and tidy ups.

Location:

tools

Files:

• traceanon/traceanon_parallel.c (modified) (5 diffs)
• tracestats/tracestats_parallel.c (modified) (4 diffs)

tools/traceanon/traceanon_parallel.c

@@ -13 +13 @@
 #include <data-struct/vector.h>
 #include <data-struct/message_queue.h>
+#include <signal.h>
 
 bool enc_source = false;
@@ -18 +19 @@
 enum enc_type_t enc_type = ENC_NONE;
 char *key = NULL;
+
+
+struct libtrace_t *trace = NULL;
+
+static void cleanup_signal(int signal)
+{
+        static int s = 0;
+        (void)signal;
+        // trace_interrupt();
+        // trace_pstop isn't really signal safe because its got lots of locks in it
+        // trace_pstop(trace);
+        if (s == 0) {
+                if (trace_ppause(trace) == -1)
+                        trace_perror(trace, "Pause failed");
+        }
+        else {
+                if (trace_pstart(trace, NULL, NULL, NULL) == -1)
+                        trace_perror(trace, "Start failed");
+        }
+        s = !s;
+}
+
+
 
 static void usage(char *argv0)
@@ -166 +190 @@
                 //libtrace_packet_t * packet_copy = trace_copy_packet(packet);
                 //libtrace_packet_t * packet_copy = trace_result_packet(trace, pkt);
-                trace_publish_result(trace, trace_packet_get_order(pkt), pkt);
+                //trace_publish_result(trace, trace_packet_get_order(pkt), pkt);
+                trace_publish_packet(trace, pkt);
                 //return ;
         }
@@ -181 +206 @@
 int main(int argc, char *argv[]) 
 {
-        struct libtrace_t *trace = 0;
+        //struct libtrace_t *trace = 0;
         struct libtrace_packet_t *packet/* = trace_create_packet()*/;
         struct libtrace_out_t *writer = 0;
+        struct sigaction sigact;
         char *output = 0;
         int level = -1;
@@ -371 +397 @@
                 return 1;
         }
+
+        sigact.sa_handler = cleanup_signal;
+        sigemptyset(&sigact.sa_mask);
+        sigact.sa_flags = SA_RESTART;
+
+        sigaction(SIGINT, &sigact, NULL);
+        sigaction(SIGTERM, &sigact, NULL);
         
         // Read in the resulting packets and then free them when done

tools/tracestats/tracestats_parallel.c

@@ -59 +59 @@
 #include <pthread.h>
 
-struct libtrace_t *trace;
+struct libtrace_t *trace = NULL;
 
 static void cleanup_signal(int signal)
 {
+        static int s = 0;
         (void)signal;
-        //trace_interrupt();
+        // trace_interrupt();
         // trace_pstop isn't really signal safe because its got lots of locks in it
-        trace_pstop(trace);
+        // trace_pstop(trace);
+        if (s == 0) {
+                if (trace_ppause(trace) == -1)
+                        trace_perror(trace, "Pause failed");
+        }
+        else {
+                if (trace_pstart(trace, NULL, NULL, NULL) == -1)
+                        trace_perror(trace, "Start failed");
+        }
+        s = !s;
 }
 
@@ -121 +131 @@
                         case MESSAGE_STOPPED:
                                 trace_publish_result(trace, 0, results); // Only ever using a single key 0
-                                fprintf(stderr, "Thread published resuslts WOWW \n");
+                                fprintf(stderr, "Thread published resuslts WOWW\n");
                                 break;
                         case MESSAGE_STARTED:
                                 results = calloc(1, sizeof(statistics_t) * (filter_count + 1));
                                 break;
-                        case MESSAGE_PAUSE:
+                        case MESSAGE_DO_PAUSE:
                                 fprintf(stderr, "GOT Asked to pause ahh\n");
+                                break;
+                        case MESSAGE_PAUSING:
+                                fprintf(stderr, "Thread is pausing\n");
+                                break;
+                        case MESSAGE_PAUSED:
+                                fprintf(stderr, "Thread has paused\n");
                                 break;
                 }
@@ -189 +205 @@
 }
 
-static uint64_t rand_hash(libtrace_packet_t * pkt) {
+static uint64_t rand_hash(libtrace_packet_t * pkt, void *data) {
         return rand();
 }
 
-static uint64_t bad_hash(libtrace_packet_t * pkt) {
+static uint64_t bad_hash(libtrace_packet_t * pkt, void *data) {
         return 0;
 }
@@ -212 +228 @@
         int option = 2;
         //option = 10000;
-        //trace_parallel_config(trace, TRACE_OPTION_USE_DEDICATED_HASHER, &option);
+        trace_parallel_config(trace, TRACE_OPTION_USE_DEDICATED_HASHER, &option);
+        trace_set_hasher(trace, HASHER_CUSTOM, &rand_hash, NULL);
         //trace_parallel_config(trace, TRACE_OPTION_USE_SLIDING_WINDOW_BUFFER, &option);
         option = 2;