Changeset 5e3f16c

Timestamp:

12/05/16 19:04:41 (5 years ago)

Author:

Richard Sanger <rsanger@…>

Branches:

4.0.1-hotfixes, cachetimestamps, develop, dpdk-ndag, etsilive, master, ndag_format, rc-4.0.1, rc-4.0.2, rc-4.0.3, rc-4.0.4, ringdecrementfix, ringperformance, ringtimestampfixes

Children:

adb2c4c, cff1819, f8613e4

Parents:

8ccb8536

Message:

Fix for issue #39 - ring and int pstop() fails on older kernels when using threads

The problem here is that on old kernels without PACKET_FANOUT support
(added in v3.1) will only include the single threaded versions of int
and ring. When used with multiple threads the libtrace API will
fallback to using read rather than pread which does not check message
queues.

To fix this issue, in any format without pread support:

• We check for new messages with each loop around read_packet as we fill the burst
• Within read_packet we update the halt to include the pausing state
• Use a seperate lock to the main lock when reading a burst of packets, otherwise trace_ppause has to wait for a burst to read.

This is not 100% perfect as a single packet might still need to be received
before a generic message can be received.
A proper fix in the future would be to move all format internals purely to the
parallel API.

Location:

lib

Files:

• format_bpf.c (modified) (1 diff)
• format_dag25.c (modified) (1 diff)
• format_dpdk.c (modified) (1 diff)
• format_linux_int.c (modified) (1 diff)
• format_linux_ring.c (modified) (1 diff)
• format_pcap.c (modified) (1 diff)
• libtrace_int.h (modified) (3 diffs)
• trace.c (modified) (5 diffs)
• trace_parallel.c (modified) (4 diffs)

lib/format_bpf.c

@@ -497 +497 @@
 
                 /* Timed out -- check if we should halt */
-                if (libtrace_halt)
-                        return 0;
+                if ((ret=is_halted(libtrace)) != -1)
+                        return ret;
         }
         

lib/format_dag25.c

@@ -1304 +1304 @@
                                 return -2;
 
-                        if (libtrace_halt)
-                                return 0;
+                        if ((numbytes=is_halted(libtrace)) != -1)
+                                return numbytes;
                         /* Block until we see a packet */
                         continue;

lib/format_dpdk.c

@@ -2107 +2107 @@
                 if (mesg && libtrace_message_queue_count(mesg) > 0)
                         return READ_MESSAGE;
-                if (libtrace_halt)
-                        return READ_EOF;
+                if ((nb_rx=is_halted(libtrace)) != -1)
+                        return nb_rx;
                 /* Wait a while, polling on memory degrades performance
                  * This relieves the pressure on memory allowing the NIC to DMA */

lib/format_linux_int.c

@@ -215 +215 @@
                                 return -1;
                         } else {
-                                if (libtrace_halt)
-                                        return READ_EOF;
+                                if ((ret=is_halted(libtrace)) != -1)
+                                        return ret;
                         }
                 }

lib/format_linux_ring.c

@@ -510 +510 @@
                 } else {
                         /* Poll timed out - check if we should exit */
-                        if (libtrace_halt)
-                                return 0;
+                        if ((ret=is_halted(libtrace)) != -1)
+                                return ret;
                         continue;
                 }

lib/format_pcap.c

@@ -447 +447 @@
                         case 1: break; /* no error */
                         case 0: 
-                                if (libtrace_halt)
-                                        return 0;
+                                if ((ret=is_halted(libtrace)) != -1)
+                                        return ret;
                                 continue; /* timeout expired */
                         case -1: 

lib/libtrace_int.h

@@ -327 +327 @@
         /** Synchronise writes/reads across this format object and attached threads etc */
         pthread_mutex_t libtrace_lock;
+        /** Packet read lock, seperate from libtrace_lock as to not block while reading a burst */
+        pthread_mutex_t read_packet_lock;
         /** State */
         enum trace_state state;
@@ -606 +608 @@
          * @return The size of the packet read (in bytes) including the capture
          * framing header, or -1 if an error occurs. 0 is returned in the
-         * event of an EOF. 
+         * event of an EOF or -2 in the case of interrupting the parallel API.
          *
          * If no packets are available for reading, this function should block
@@ -1032 +1034 @@
  */
 extern volatile int libtrace_halt;
+
+/**
+ * Used by a format to check if trace_interrupt or if a trace_pause/stop has
+ * been called. Provides backwards compatibility with traditional read
+ * functions when trace_read_packet() is used by the parallel API.
+ *
+ * Returns -1 if not halting otherwise returns the code that the read
+ * operation should pass on.
+ */
+static inline int is_halted(libtrace_t *trace) {
+        if (!(libtrace_halt || trace->state == STATE_PAUSING)) {
+                return -1;
+        } else if (libtrace_halt) {
+                return READ_EOF;
+        } else {
+                return READ_MESSAGE;
+        }
+}
 
 /** Registers a new capture format module.

lib/trace.c

@@ -261 +261 @@
         /* Parallel inits */
         ASSERT_RET(pthread_mutex_init(&libtrace->libtrace_lock, NULL), == 0);
+        ASSERT_RET(pthread_mutex_init(&libtrace->read_packet_lock, NULL), == 0);
         ASSERT_RET(pthread_cond_init(&libtrace->perpkt_cond, NULL), == 0);
         libtrace->state = STATE_NEW;
@@ -385 +386 @@
         /* Parallel inits */
         ASSERT_RET(pthread_mutex_init(&libtrace->libtrace_lock, NULL), == 0);
+        ASSERT_RET(pthread_mutex_init(&libtrace->read_packet_lock, NULL), == 0);
         ASSERT_RET(pthread_cond_init(&libtrace->perpkt_cond, NULL), == 0);
         libtrace->state = STATE_NEW; // TODO MAYBE DEAD
@@ -680 +682 @@
 
         ASSERT_RET(pthread_mutex_destroy(&libtrace->libtrace_lock), == 0);
+        ASSERT_RET(pthread_mutex_destroy(&libtrace->read_packet_lock), == 0);
         ASSERT_RET(pthread_cond_destroy(&libtrace->perpkt_cond), == 0);
 
@@ -758 +761 @@
 
         ASSERT_RET(pthread_mutex_destroy(&libtrace->libtrace_lock), == 0);
+        ASSERT_RET(pthread_mutex_destroy(&libtrace->read_packet_lock), == 0);
         ASSERT_RET(pthread_cond_destroy(&libtrace->perpkt_cond), == 0);
 
@@ -921 +925 @@
                         size_t ret;
                         int filtret;
-                        if (libtrace_halt)
-                                return 0;
+                        if ((ret=is_halted(libtrace)) != (size_t)-1)
+                                return ret;
                         /* Store the trace we are reading from into the packet opaque 
                          * structure */
                         packet->trace = libtrace;
                         ret=libtrace->format->read_packet(libtrace,packet);
-                        if (ret==(size_t)-1 || ret==0) {
+                        if (ret==(size_t)READ_MESSAGE ||
+                            ret==(size_t)-1 || ret==0) {
                                 packet->trace = NULL;
                                 return ret;

lib/trace_parallel.c

@@ -816 +816 @@
 
                 if ((packet->error = trace_read_packet(trace, packet)) <1) {
-                        break; /* We are EOF or error'd either way we stop  */
+                        if (packet->error == READ_MESSAGE) {
+                                pkt_skipped = 1;
+                                continue;
+                        } else {
+                                break; /* We are EOF or error'd either way we stop  */
+                        }
                 }
 
@@ -884 +889 @@
         //bool tick_hit = false;
 
-        ASSERT_RET(pthread_mutex_lock(&libtrace->libtrace_lock), == 0);
+        ASSERT_RET(pthread_mutex_lock(&libtrace->read_packet_lock), == 0);
         /* Read nb_packets */
         for (i = 0; i < nb_packets; ++i) {
+                if (libtrace_message_queue_count(&t->messages) > 0) {
+                        if ( i==0 ) {
+                                ASSERT_RET(pthread_mutex_unlock(&libtrace->read_packet_lock), == 0);
+                                return READ_MESSAGE;
+                        } else {
+                                break;
+                        }
+                }
                 packets[i]->error = trace_read_packet(libtrace, packets[i]);
 
@@ -892 +905 @@
                         /* We'll catch this next time if we have already got packets */
                         if ( i==0 ) {
-                                ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
+                                ASSERT_RET(pthread_mutex_unlock(&libtrace->read_packet_lock), == 0);
                                 return packets[i]->error;
                         } else {
@@ -908 +921 @@
                 store_first_packet(libtrace, packets[0], t);
         }
-        ASSERT_RET(pthread_mutex_unlock(&libtrace->libtrace_lock), == 0);
+        ASSERT_RET(pthread_mutex_unlock(&libtrace->read_packet_lock), == 0);
         /* XXX TODO this needs to be inband with packets, or we don't bother in this case
         if (tick_hit) {