Changeset 526d9d0

Timestamp:

02/24/15 17:18:21 (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:

116f970

Parents:

10c47a0

Message:

Move the accepted packet count to dispatching packets for threads.
Accounting for the doubled count when using a single threaded format.

Rework statistics logic slightly to remove duplicated code.

Location:

lib

Files:

• format_dag25.c (modified) (2 diffs)
• format_linux_common.c (modified) (1 diff)
• libtrace.h.in (modified) (1 diff)
• libtrace_int.h (modified) (2 diffs)
• trace.c (modified) (4 diffs)
• trace_parallel.c (modified) (7 diffs)

lib/format_dag25.c

@@ -1447 +1447 @@
         tmp = FORMAT_DATA_HEAD;
 
-        /* We don't filter packets on the card itself */
-        stat->filtered_valid = 1;
-        stat->filtered = 0;
-
-        /* Dropped, filtered the  */
+        /* Dropped packets */
         stat->dropped_valid = 1;
         stat->dropped = 0;
@@ -1512 +1508 @@
         libtrace_list_node_t *node;
 
-        if (reader) {
-                if (t->type == THREAD_PERPKT) {
-
-                        node = libtrace_list_get_index(FORMAT_DATA->per_stream,
-                                                        t->perpkt_num);
-                        if (node == NULL) {
-                                trace_set_err(libtrace, TRACE_ERR_INIT_FAILED,
-                                              "Too few streams supplied for the"
-                                              " number of threads lanuched");
-                                return -1;
-                        }
-                        stream_data = node->data;
-
-                        /* Pass the per thread data to the thread */
-                        t->format_data = stream_data;
-
-                        /* Attach and start the DAG stream */
-                        printf("t%u: starting and attaching stream #%u\n",
-                               t->perpkt_num, stream_data->dagstream);
-                        if (dag_start_input_stream(libtrace, stream_data) < 0)
-                                return -1;
-                } else {
-                        /* TODO: Figure out why t->type != THREAD_PERPKT in
-                         * order to figure out what this line does */
-                        t->format_data = FORMAT_DATA_FIRST;
-                }
+        if (reader && t->type == THREAD_PERPKT) {
+                fprintf(stderr, "t%u: registered reader thread", t->perpkt_num);
+                node = libtrace_list_get_index(FORMAT_DATA->per_stream,
+                                                t->perpkt_num);
+                if (node == NULL) {
+                        trace_set_err(libtrace, TRACE_ERR_INIT_FAILED,
+                                      "Too few streams supplied for the"
+                                      " number of threads lanuched");
+                        return -1;
+                }
+                stream_data = node->data;
+
+                /* Pass the per thread data to the thread */
+                t->format_data = stream_data;
+
+                /* Attach and start the DAG stream */
+                printf("t%u: starting and attaching stream #%u\n",
+                       t->perpkt_num, stream_data->dagstream);
+                if (dag_start_input_stream(libtrace, stream_data) < 0)
+                        return -1;
         }
 

lib/format_linux_common.c

@@ -596 +596 @@
 
         /* filtered count == dev received - socket received */
-        if (FORMAT_DATA->filter == NULL) {
-                stat->filtered_valid = 1;
-                stat->filtered = 0;
-        } else if (FORMAT_DATA->stats_valid && dev_stats.if_name[0]) {
-                stat->filtered_valid = 1;
-                stat->filtered = DEV_DIFF(rx_packets) -
-                                 FORMAT_DATA->stats.tp_packets;
-                if (stat->filtered > UINT64_MAX - 100000) {
-                        stat->filtered = 0;
+        if (FORMAT_DATA->filter != NULL &&
+            FORMAT_DATA->stats_valid &&
+            dev_stats.if_name[0]) {
+                uint64_t filtered = DEV_DIFF(rx_packets) -
+                                    FORMAT_DATA->stats.tp_packets;
+                /* Check the value is sane, due to timing it could be below 0 */
+                if (filtered < UINT64_MAX - 100000) {
+                        stat->filtered += filtered;
                 }
         }

lib/libtrace.h.in

@@ -249 +249 @@
 typedef struct libtrace_filter_t libtrace_filter_t;
 
+/** Opaque structure holding information about libtrace thread */
 typedef struct libtrace_thread_t libtrace_thread_t;
 

lib/libtrace_int.h

@@ -836 +836 @@
         uint64_t (*get_dropped_packets)(libtrace_t *trace);
 
-        /** Returns statistics about a trace. Flags are all set to 0 when
-         * invoked.
+        /** Returns statistics about a trace.
          *
          * @param trace The libtrace object
-         * @param stat A zeroed structure ready to be filled.
+         * @param stat [in,out] A statistics structure ready to be filled
+         *
+         * The filtered and accepted statistics will be set to the values
+         * stored in the library. All other statistics are not set.
+         *
+         * @note If filtering of packets is performed by a trace and the number
+         * of filtered packets is unknown this should be marked as invalid by
+         * the format.
          */
         void (*get_statistics)(libtrace_t *trace, libtrace_stat_t *stat);
@@ -955 +961 @@
         void (*punregister_thread)(libtrace_t *libtrace, libtrace_thread_t *t);
 
-        /**
-         * Return statistics for a single thread.
+        /** Returns statistics for a single thread.
+         *
+         * @param trace The libtrace object
+         * @param t The thread to return statistics for
+         * @param stat [in,out] A statistics structure ready to be filled
+         *
+         * The filtered and accepted statistics will be set to the values
+         * stored in the library. All other statistics are not set.
+         *
+         * @note If filtering of packets is performed by a trace and the number
+         * of filtered packets is unknown this should be marked as invalid by
+         * the format.
          */
         void (*get_thread_statistics)(libtrace_t *libtrace,

lib/trace.c

@@ -2027 +2027 @@
         assert(trace);
         int i = 0;
-        uint64_t ret = trace->accepted_packets;
+        uint64_t ret = 0;
+        /* We always add to a thread's accepted count before dispatching the
+         * packet to the user. However if the underlying trace is single
+         * threaded it will also be increasing the global count. So if we
+         * find perpkt ignore the global count.
+         */
         for (i = 0; i < trace->perpkt_thread_count; i++) {
                 ret += trace->perpkt_threads[i].accepted_packets;
         }
-        return ret;
+        return ret ? ret : trace->accepted_packets;
 }
 
@@ -2061 +2066 @@
         LIBTRACE_STAT_FIELDS;
 #undef X
-        if (trace->format->get_statistics) {
-                trace->format->get_statistics(trace, stat);
-                ret = trace_get_accepted_packets(trace);
-                if (ret != UINT64_MAX) {
-                        stat->accepted_valid = 1;
-                        stat->accepted = ret;
-                }
-                /* Now add on any library filtered packets */
-                if (stat->filtered_valid) {
-                        stat->filtered += trace->filtered_packets;
-                        for (i = 0; i < trace->perpkt_thread_count; i++) {
-                                stat->filtered += trace->perpkt_threads[i].filtered_packets;
-                        }
-                }
-                return stat;
-        }
+        /* Both accepted and filtered are stored against in the library */
         ret = trace_get_accepted_packets(trace);
         if (ret != UINT64_MAX) {
@@ -2082 +2072 @@
                 stat->accepted = ret;
         }
-        ret = trace_get_received_packets(trace);
-        if (ret != UINT64_MAX) {
-                stat->received_valid = 1;
-                stat->received = ret;
-        }
-        /* Fallback to the old way */
-        ret = trace_get_dropped_packets(trace);
-        if (ret != UINT64_MAX) {
-                stat->dropped_valid = 1;
-                stat->dropped = ret;
-        }
-        ret = trace_get_filtered_packets(trace);
-        if (ret != UINT64_MAX) {
-                stat->filtered_valid = 1;
-                stat->filtered = ret;
+
+        stat->filtered_valid = 1;
+        stat->filtered = trace->filtered_packets;
+        for (i = 0; i < trace->perpkt_thread_count; i++) {
+                stat->filtered += trace->perpkt_threads[i].filtered_packets;
+        }
+
+        if (trace->format->get_statistics) {
+                trace->format->get_statistics(trace, stat);
+        } else {
+                /* Fallback to the old way */
+                ret = trace_get_received_packets(trace);
+                if (ret != UINT64_MAX) {
+                        stat->received_valid = 1;
+                        stat->received = ret;
+                }
+                ret = trace_get_dropped_packets(trace);
+                if (ret != UINT64_MAX) {
+                        stat->dropped_valid = 1;
+                        stat->dropped = ret;
+                }
         }
         return stat;
@@ -2112 +2108 @@
         LIBTRACE_STAT_FIELDS;
 #undef X
-        if (trace->format->get_thread_statistics) {
+        stat->accepted_valid = 1;
+        stat->accepted = t->accepted_packets;
+        stat->filtered_valid = 1;
+        stat->filtered = t->filtered_packets;
+        if (!trace_has_dedicated_hasher(trace) && trace->format->get_thread_statistics) {
                 trace->format->get_thread_statistics(trace, t, stat);
-        }
-        if (t->accepted_packets != UINT64_MAX) {
-                stat->accepted_valid = 1;
-                stat->accepted = t->accepted_packets;
-        }
-        /* Now add on any library filtered packets */
-        if (stat->filtered_valid) {
-                stat->filtered += t->filtered_packets;
         }
         return;

lib/trace_parallel.c

@@ -173 +173 @@
  * @return true if the trace has dedicated hasher thread otherwise false.
  */
-static inline bool trace_has_dedicated_hasher(libtrace_t * libtrace)
+inline bool trace_has_dedicated_hasher(libtrace_t * libtrace)
 {
         return libtrace->hasher_thread.type == THREAD_HASHER;
@@ -412 +412 @@
                                 return READ_MESSAGE;
                 }
+                t->accepted_packets++;
                 *packet = (*trace->per_pkt)(trace, *packet, NULL, t);
                 trace_fin_packet(*packet);
@@ -732 +733 @@
                 assert(packet);
 
-                if (libtrace_halt) // Signal to die has been sent - TODO
+                if (libtrace_halt) {
+                        packet->error = 0;
                         break;
+                }
 
                 // Check for messages that we expect MESSAGE_DO_PAUSE, (internal messages only)
@@ -750 +753 @@
                                         break;
                                 case MESSAGE_DO_STOP:
-                                        // Stop called after pause
                                         assert(trace->started == false);
                                         assert(trace->state == STATE_FINSHED);
+                                        /* Mark the current packet as EOF */
+                                        packet->error = 0;
                                         break;
                                 default:
@@ -761 +765 @@
                 }
 
-                if ((packet->error = trace_read_packet(trace, packet)) <1 /*&& psize != LIBTRACE_MESSAGE_WAITING*/) {
+                if ((packet->error = trace_read_packet(trace, packet)) <1) {
                         break; /* We are EOF or error'd either way we stop  */
                 }
@@ -1443 +1447 @@
                                 trace_packet_set_order(packets[i], trace_get_erf_timestamp(packets[i]));
                         }
-                        t->accepted_packets += ret;
                 } while(ret == 0);
                 return ret;
@@ -1681 +1684 @@
          */
         if (trace_has_dedicated_hasher(libtrace)) {
+                libtrace->hasher_thread.type = THREAD_EMPTY;
                 ret = trace_start_thread(libtrace, &libtrace->hasher_thread,
                                    THREAD_HASHER, hasher_entry, -1,