Changeset be3f75b

Timestamp:

07/30/14 19:17:14 (6 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:

f9a70ca

Parents:

a49a9eb

Message:

Fixs statistic counters for parallel traces

Location:

lib

Files:

• libtrace_int.h (modified) (2 diffs)
• trace.c (modified) (4 diffs)
• trace_parallel.c (modified) (4 diffs)

lib/libtrace_int.h

@@ -204 +204 @@
  */
 struct libtrace_thread_t {
+        int accepted_packets; // The number of packets accepted only used if pread
+        // is retreving packets
+        // Set to true once the first packet has been stored
+        bool recorded_first;
+        // For thread safety reason we actually must store this here
+        int64_t tracetime_offset_usec;
+        void* user_data; // TLS for the user to use
+        void* format_data; // TLS for the format to use
+        libtrace_message_queue_t messages; // Message handling
+        libtrace_ringbuffer_t rbuffer; // Input
+        libtrace_vector_t vector; // Output
+        libtrace_queue_t deque; // Real Output type makes more sense
         libtrace_t * trace;
         void* ret;
         enum thread_types type;
         enum thread_states state;
-        void* user_data; // TLS for the user to use
-        void* format_data; // TLS for the format to use
         pthread_t tid;
         int perpkt_num; // A number from 0-X that represents this perpkt threads number
                                 // in the table, intended to quickly identify this thread
                                 // -1 represents NA (such as the case this is not a perpkt thread)
-        libtrace_ringbuffer_t rbuffer; // Input
-        libtrace_vector_t vector; // Output
-        libtrace_queue_t deque; // Real Output type makes more sense
-        libtrace_message_queue_t messages; // Message handling
-        // Temp storage for time sensitive results
-        uint64_t tmp_key;
-        void *tmp_data;
-        pthread_spinlock_t tmp_spinlock;
-        // Set to true once the first packet has been stored
-        bool recorded_first;
-        // For thread safety reason we actually must store this here
-        int64_t tracetime_offset_usec;
 };
 
@@ -347 +345 @@
         struct first_packets first_packets;
         int tracetime;
+
+        /*
+         * Caches statistic counters in the case that our trace is
+         * paused or stopped before this counter is taken
+         */
+        uint64_t dropped_packets;
+        uint64_t received_packets; 
 };
 

lib/trace.c

@@ -286 +286 @@
         libtrace->first_packets.count = 0;
         libtrace->first_packets.packets = NULL;
+        libtrace->dropped_packets = UINT64_MAX;
+        libtrace->accepted_packets = UINT64_MAX;
 
         /* Parse the URI to determine what sort of trace we are dealing with */
@@ -1940 +1942 @@
 {
         assert(trace);
+        uint64_t ret;
+
         if (trace->format->get_received_packets) {
-                return trace->format->get_received_packets(trace);
-        }
-        return (uint64_t)-1;
+                if ((ret = trace->format->get_received_packets(trace)) != UINT64_MAX)
+                        return ret;
+        }
+        // Read this cached value taken before the trace was closed
+        return trace->received_packets;
 }
 
@@ -1967 +1973 @@
 {
         assert(trace);
+        uint64_t ret;
+
         if (trace->format->get_dropped_packets) {
-                return trace->format->get_dropped_packets(trace);
-        }
-        return (uint64_t)-1;
+                if ((ret = trace->format->get_dropped_packets(trace)) != UINT64_MAX)
+                        return ret;
+        }
+        // Read this cached value taken before the trace was closed
+        return trace->dropped_packets;
 }
 
@@ -1976 +1986 @@
 {
         assert(trace);
-        return trace->accepted_packets;
+        int i = 0;
+        uint64_t ret = trace->accepted_packets;
+        for (i = 0; i < trace->perpkt_thread_count; i++) {
+                ret += trace->perpkt_threads[i].accepted_packets;
+        }
+        return ret;
 }
 

lib/trace_parallel.c

@@ -279 +279 @@
         t->recorded_first = false;
         t->perpkt_num = -1;
+        t->accepted_packets = 0;
 }
 
@@ -1248 +1249 @@
                                                 libtrace->snaplen);
                         }
-                        trace_packet_set_order(packet, libtrace->accepted_packets);
-                        ++libtrace->accepted_packets;
+                        
+                        ++t->accepted_packets;
+                        // TODO look into this better
+                        trace_packet_set_order(packet, trace_get_erf_timestamp(packet));
+                        //trace_packet_set_order(packet, libtrace->accepted_packets);
+                        //++libtrace->accepted_packets;
                         return ret;
                 } while(1);
@@ -1470 +1475 @@
                 libtrace_deque_init(&t->deque, sizeof(libtrace_result_t));
                 libtrace_message_queue_init(&t->messages, sizeof(libtrace_message_t));
-                t->tmp_key = 0;
-                t->tmp_data = NULL;
                 t->recorded_first = false;
-                ASSERT_RET(pthread_spin_init(&t->tmp_spinlock, 0), == 0);
                 t->tracetime_offset_usec = 0;;
         }
@@ -1602 +1604 @@
 
         if (trace_supports_parallel(libtrace) && !trace_has_dedicated_hasher(libtrace)) {
+                uint64_t tmp_stats;
+                libtrace->dropped_packets = trace_get_dropped_packets(libtrace);
+                libtrace->received_packets = trace_get_received_packets(libtrace);
+                if (libtrace->format->get_filtered_packets) {
+                        if ((tmp_stats = libtrace->format->get_filtered_packets(libtrace)) != UINT64_MAX) {
+                                libtrace->filtered_packets += tmp_stats;
+                        }
+                }
                 libtrace->started = false;
                 if (libtrace->format->ppause_input)