Changeset 9d5f464

Timestamp:

11/08/18 17:15:19 (4 years ago)

Author:

Jacob Van Walraven <jcv9@…>

Branches:

develop

Children:

6984c96

Parents:

93f4c64

Message:

Improvments to tickrate calculations

File:

• examples/tutorial/ipdist-parallel.c (modified) (18 diffs)

examples/tutorial/ipdist-parallel.c

@@ -15 +15 @@
         uint64_t src[256];
         uint64_t dst[256];
+        uint64_t lastkey;
+        uint64_t packets;
 };
 /* Structure to hold the result from a processing thread */
@@ -20 +22 @@
         uint64_t src[256];
         uint64_t dst[256];
+        uint64_t packets;
 };
 /* Structure to hold counters the report has one of these, it combines
@@ -27 +30 @@
         uint64_t dst[256];
         uint64_t lastkey;
+        uint64_t packets;
 };
 
@@ -40 +44 @@
 };
 
+uint64_t tickrate;
+
 /* Start callback function - This is run for each thread when it starts */
 static void *start_callback(libtrace_t *trace, libtrace_thread_t *thread, void *global) {
@@ -50 +56 @@
                 local->dst[i] = 0;
         }
+        local->lastkey = 0;
+        local->packets = 0;
 
         /* return the local storage so it is available for all other callbacks for the thread*/
@@ -114 +122 @@
         struct addr_local *local = (struct addr_local *)tls;
 
+        /* Store the timestamp of the last packet in erf format
+         * We use the timestamp in the packet for processing non live traces */
+        local->lastkey = trace_get_erf_timestamp(packet);
+        /* Increment the packet count */
+        local->packets += 1;
+
         /* Regain access to excluded networks pointer */
         struct exclude_networks *exclude = (struct exclude_networks *)global;
@@ -138 +152 @@
 }
 
-/* Stopping callback function */
+/* Stopping callback function - When a thread closes */
 static void stop_processing(libtrace_t *trace, libtrace_thread_t *thread, void *global, void *tls) {
 
-        /* create a structure to hold the result from the processing thread */
-        struct addr_result *result = (struct addr_result *)malloc(sizeof(struct addr_result));
         /* cast the local storage structure */
         struct addr_local *local = (struct addr_local *)tls;
-
-        /* Populate the result structure */
+        /* Create structure to store the result */
+        struct addr_result *result = (struct addr_result *)malloc(sizeof(struct addr_result));
+
+        /* Populate the result */
         int i;
         for(i=0;i<256;i++) {
                 result->src[i] = local->src[i];
-                result->dst[i] = local->dst[i];
-        }
-
-        /* Publish the result to the reporter thread */
+                result->src[i] = local->src[i];
+        }
+        result->packets = local->packets;
+
+        /* This will not cause the result to be printed but will atleast end up going into our tally
+         * The reporter thread can then deal with it when it closes */
         trace_publish_result(trace, thread, 0, (libtrace_generic_t){.ptr=result}, RESULT_USER);
 
@@ -159 +175 @@
         free(local);
 }
-
-
-
 
 /* Starting callback for reporter thread */
@@ -175 +188 @@
         }
         tally->lastkey = 0;
+        tally->packets = 0;
 
         return tally;
 }
 
-static void plot_results(struct addr_tally *tally) {
+static void plot_results(struct addr_tally *tally, uint64_t tick) {
 
         /* Get the current time */
         time_t current_time = time(NULL);
-        char* time_string = ctime(&current_time);
-
-        /* Push all the data into a tmp file for gnuplot */
+
         char outputfile[255];
         char outputplot[255];
-        snprintf(outputfile, sizeof(outputfile), "ipdist-%u.data", current_time);
-        snprintf(outputplot, sizeof(outputplot), "ipdist-%u.png", current_time);
-        FILE *tmp = fopen(outputfile, "w");
+        snprintf(outputfile, sizeof(outputfile), "ipdist-%u.data", tick);
+        snprintf(outputplot, sizeof(outputplot), "ipdist-%u.png", tick);
+
+        /* Push all data into data file */
+        FILE *tmp = fopen(outputfile, "w");
         int i;
         for(i=0;i<255;i++) {
@@ -196 +210 @@
         }
         fclose(tmp);
+        printf("wrote out to file %s\n", outputfile);
 
         /* Open pipe to gnuplot */
@@ -204 +219 @@
         fprintf(gnuplot, "set xrange[0:255]\n");
         fprintf(gnuplot, "set xlabel 'Prefix'\n");
-        fprintf(gnuplot, "set xlabel 'Hits'\n");
+        fprintf(gnuplot, "set ylabel 'Hits'\n");
         fprintf(gnuplot, "set xtics 0,10,255\n");
         fprintf(gnuplot, "set output '%s'\n", outputplot);
         fprintf(gnuplot, "plot '%s' using 1:2 title 'Source address' with boxes, '%s' using 1:3 title 'Destination address' with boxes\n", outputfile, outputfile);
+        fprintf(gnuplot, "replot");
         pclose(gnuplot);
 }
@@ -217 +233 @@
 
         struct addr_result *results;
-        uint64_t key;
         struct addr_tally *tally;
+        uint64_t key;
 
         /* We only want to handle results containing our user-defined structure  */
@@ -233 +249 @@
         /* Grab our tally out of thread local storage */
         tally = (struct addr_tally *)tls;
-        /* increment tally with the new results */
-        int i;
-        for(i=0;i<256;i++) {
-                tally->src[i] += results->src[i];
-                tally->dst[i] += results->dst[i];
+
+        /* Add all the results to the tally */
+        int i;
+        for(i=0;i<256;i++) {
+                tally->src[i] += results->src[i];
+                tally->dst[i] += results->dst[i];
+        }
+        tally->packets += results->packets;
+
+        /* If the current timestamp is greater than the last printed plus the interval, output a result */
+        if((key >> 32) >= (tally->lastkey >> 32) + (tickrate/1000)) {
+
+                /* update last key */
                 tally->lastkey = key;
-        }
-
-        /* Plot the result */
-        plot_results(tally);
+
+                /* Plot the result with the key in epoch seconds*/
+                plot_results(tally, key >> 32);
+
+                /* clear the tally */
+                for(i=0;i<256;i++) {
+                        tally->src[i] = 0;
+                        tally->dst[i] = 0;
+                }
+                tally->packets = 0;
+        }
 
         /* Cleanup the thread results */
@@ -254 +285 @@
         struct addr_tally *tally = (struct addr_tally *)tls;
 
-        //OUTPUT DATA
-        //plot_results(tally);
-
+        /* If there is any remaining data in the tally plot it */
+        if(tally->packets > 0) {
+                plot_results(tally, (tally->lastkey >> 32) + tickrate);
+        }
         /* Cleanup tally results*/
         free(tally);
@@ -264 +296 @@
 
         struct addr_result *result = (struct addr_result *)malloc(sizeof(struct addr_result));
+        /* Proccessing thread local storage */
         struct addr_local *local = (struct addr_local *)tls;
 
@@ -276 +309 @@
                 local->dst[i] = 0;
         }
-
-        printf("tick: %u\n", tick);
-
-        /* Push the result to the combiner */
-        trace_publish_result(trace, thread, tick, (libtrace_generic_t){.ptr=result}, RESULT_USER);
+        result->packets = local->packets;
+
+        /* only use the tick timestamp if running against a live capture */
+        uint64_t key;
+        if(trace_get_information(trace)->live) {
+                key = tick;
+        } else {
+                key = local->lastkey;
+        }
+
+        /* Push result to the combiner */
+        trace_publish_result(trace, thread, key, (libtrace_generic_t){.ptr=result}, RESULT_USER);
 }
 
@@ -296 +336 @@
         }
         /* Convert tick into correct format */
-        uint64_t tickrate = atoi(argv[2]);
+        tickrate = atoi(argv[2]);
 
         /* Create the trace */
@@ -324 +364 @@
         /* Try to balance the load across all processing threads */
         trace_set_hasher(trace, HASHER_BALANCE, NULL, NULL);
+
         /* Set the tick interval */
         trace_set_tick_interval(trace, tickrate);
-
+        /* Do not buffer the reports */
+        trace_set_reporter_thold(trace, 1);