Changeset f2066fa

Timestamp:

08/25/15 17:40:12 (5 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:

2fa43fa

Parents:

03aca91

Message:

Fix #5 make trace_pstart fallback to the single threaded format

If starting a parallel format fails we now retry as a single threaded format.
This fixes ring/int on older (pre 3.1 kernels) machines without PACKET_FANOUT.
This behaviour can be detected using trace_is_parallel()

Location:

lib

Files:

• format_linux_common.c (modified) (1 diff)
• format_linux_int.c (modified) (1 diff)
• format_linux_ring.c (modified) (1 diff)
• libtrace_int.h (modified) (1 diff)
• libtrace_parallel.h (modified) (5 diffs)
• trace_parallel.c (modified) (9 diffs)

lib/format_linux_common.c

@@ -661 +661 @@
                         linuxcommon_close_input_stream(libtrace, stream);
                 }
-                libtrace_list_deinit(FORMAT_DATA->per_stream);
-                free(libtrace->format_data);
-                libtrace->format_data = NULL;
                 return -1;
         }

lib/format_linux_int.c

@@ -69 +69 @@
 {
         int ret = linuxcommon_start_input_stream(libtrace, FORMAT_DATA_FIRST);
-        if (ret != 0) {
-                libtrace_list_deinit(FORMAT_DATA->per_stream);
-                free(libtrace->format_data);
-                libtrace->format_data = NULL;
-        }
         return ret;
 }

lib/format_linux_ring.c

@@ -280 +280 @@
 {
         int ret = linuxring_start_input_stream(libtrace, FORMAT_DATA_FIRST);
-        if (ret != 0) {
-                libtrace_list_deinit(FORMAT_DATA->per_stream);
-                free(libtrace->format_data);
-                libtrace->format_data = NULL;
-        }
         return ret;
 }

lib/libtrace_int.h

@@ -332 +332 @@
         /** User defined per_msg function called when a message is ready */
         fn_cb_msg per_msg;
-        /** User defined reporter function entry point XXX not hooked up */
+        /** User defined reporter function entry point */
         fn_reporter reporter;
         /** The hasher function */

lib/libtrace_parallel.h

@@ -537 +537 @@
  *
  * @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.
@@ -627 +626 @@
 /** Store data against a thread.
  *
- * @param The parallel trace.
- * @param data The new value to save against the trace
+ * @param thread The thread
+ * @param data The new value to save against the thread
  * @return The previously stored value
  *
@@ -713 +712 @@
  * as possible (real-time).
  *
- * @param A parallel input trace
+ * @param trace A parallel input trace
  * @param tracetime If true packets are released with time intervals matching
  * the original trace. Otherwise packets are read as fast as possible.
@@ -895 +894 @@
          *
          * @param key (Typically) The packets order, see trace_packet_get_order()
-         * @param
          */
         RESULT_PACKET,
@@ -1051 +1049 @@
  */
 DLLEXPORT bool trace_has_finished(libtrace_t * libtrace);
+
+
+/** Check if libtrace is directly reading from multiple queues
+ * from the format (such as a NICs hardware queues).
+ *
+ * When a parallel trace is running, or if checked after its completion
+ * this returns true if a trace was able to run natively parallel
+ * from the format. Otherwise false is returned, meaning libtrace is
+ * distibuting packets across multiple threads from a single source.
+ *
+ * Factors that may stop this happening despite the format supporting
+ * native parallel reads include: the choice of hasher function,
+ * the number of threads choosen (such as 1 or more than the trace supports)
+ * or another error when trying to start the parallel format.
+ *
+ * If this is called before the trace is started. I.e. before pstart
+ * this returns an indication that the trace has the possiblity to support
+ * native parallel reads. After trace pstart is called this should be
+ * checked again to confirm this has happened.
+ *
+ *
+ * @return true if the trace is parallel or false if the library is splitting
+ * the trace into multiple threads.
+ */
+DLLEXPORT bool trace_is_parallel(libtrace_t * libtrace);
 
 /** Returns either the sequence number or erf timestamp of a packet.

lib/trace_parallel.c

@@ -365 +365 @@
                 pthread_t tid = pthread_self();
                 // Check if we are reporter or something else
-                if (pthread_equal(tid, libtrace->reporter_thread.tid))
+                if (libtrace->hasher_thread.type == THREAD_REPORTER &&
+                                pthread_equal(tid, libtrace->reporter_thread.tid))
                         ret = &libtrace->reporter_thread;
-                else if (pthread_equal(tid, libtrace->hasher_thread.tid))
+                else if (libtrace->hasher_thread.type == THREAD_HASHER &&
+                         pthread_equal(tid, libtrace->hasher_thread.tid))
                         ret = &libtrace->hasher_thread;
                 else
@@ -589 +591 @@
 
         if (trace->format->pregister_thread) {
-                trace->format->pregister_thread(trace, t, !trace_has_dedicated_hasher(trace));
+                trace->format->pregister_thread(trace, t, trace_is_parallel(trace));
         }
 
@@ -1364 +1366 @@
                 libtrace->global_blob = global_blob;
 
-        if (libtrace->perpkt_thread_count > 1 &&
-            trace_supports_parallel(libtrace) &&
-            !trace_has_dedicated_hasher(libtrace)) {
+        if (trace_is_parallel(libtrace)) {
                 err = libtrace->format->pstart_input(libtrace);
         } else {
@@ -1579 +1579 @@
 }
 
+DLLEXPORT bool trace_is_parallel(libtrace_t * libtrace) {
+        if (libtrace->state == STATE_NEW)
+                return trace_supports_parallel(libtrace);
+        return libtrace->pread == trace_pread_packet_wrapper;
+}
+
 DLLEXPORT int trace_pstart(libtrace_t *libtrace, void* global_blob,
                            fn_cb_msg per_msg, fn_reporter reporter) {
@@ -1626 +1632 @@
         verify_configuration(libtrace);
 
+        ret = -1;
         /* Try start the format - we prefer parallel over single threaded, as
          * these formats should support messages better */
@@ -1632 +1639 @@
                 ret = libtrace->format->pstart_input(libtrace);
                 libtrace->pread = trace_pread_packet_wrapper;
-        } else {
+        }
+        if (ret != 0) {
                 if (libtrace->format->start_input) {
                         ret = libtrace->format->start_input(libtrace);
@@ -1770 +1778 @@
         libtrace->perpkt_thread_states[THREAD_FINISHED] = 0;
 cleanup_started:
-        if (trace_supports_parallel(libtrace) &&
-            !trace_has_dedicated_hasher(libtrace)
-            && libtrace->perpkt_thread_count > 1) {
+        if (libtrace->pread == trace_pread_packet_wrapper) {
                 if (libtrace->format->ppause_input)
                         libtrace->format->ppause_input(libtrace);
@@ -1933 +1939 @@
         // Save the statistics against the trace
         trace_get_statistics(libtrace, NULL);
-        if (trace_supports_parallel(libtrace) && !trace_has_dedicated_hasher(libtrace) && libtrace->perpkt_thread_count > 1) {
+        if (trace_is_parallel(libtrace)) {
                 libtrace->started = false;
                 if (libtrace->format->ppause_input)
@@ -2355 +2361 @@
         return 0;
 }
-
-
 
 static bool config_bool_parse(char *value, size_t nvalue) {