Changeset f96ad61

Timestamp:

10/24/18 15:20:19 (2 years ago)

Author:

Shane Alcock <salcock@…>

Branches:

ringperformance

Children:

84dbd79

Parents:

928db3d

Message:

Attempt to improve ring per-packet performance in parallel libtrace

• Try to read multiple packets on each pread call, instead of just one.
• Inline most of the prepare packet workload to avoid function call overhead when reading packets.

File:

• lib/format_linux_ring.c (modified) (8 diffs)

lib/format_linux_ring.c

@@ -320 +320 @@
 #ifdef HAVE_PACKET_FANOUT
 static int linuxring_pstart_input(libtrace_t *libtrace) {
-        /* Only because our pread is hard-coded to only do one
-         * packet at a time anyway */
-        libtrace->config.burst_size = 1;
         return linuxcommon_pstart_input(libtrace, linuxring_start_input_stream);
 }
@@ -465 +462 @@
 }
 
-static int linuxring_prepare_packet(libtrace_t *libtrace UNUSED,
+static inline int _linuxring_prepare_packet(
                                     libtrace_packet_t *packet, void *buffer,
                                     libtrace_rt_types_t rt_type, uint32_t flags)
@@ -491 +488 @@
 }
 
+static inline int linuxring_prepare_packet(libtrace_t *libtrace UNUSED,
+                                    libtrace_packet_t *packet, void *buffer,
+                                    libtrace_rt_types_t rt_type, uint32_t flags)
+{
+        return _linuxring_prepare_packet(packet, buffer, rt_type, flags);
+}
 #ifdef HAVE_NETPACKET_PACKET_H
 #define LIBTRACE_MIN(a,b) ((a)<(b) ? (a) : (b))
@@ -500 +503 @@
                                         libtrace_packet_t *packet,
                                         struct linux_per_stream_t *stream,
-                                        libtrace_message_queue_t *queue) {
+                                        libtrace_message_queue_t *queue,
+                                        int block) {
 
         struct tpacket2_hdr *header;
@@ -518 +522 @@
          * ready for consumption.
          */
+
         while (!(header->tp_status & TP_STATUS_USER) ||
-               header->tp_status == TP_STATUS_LIBTRACE) {
+                        header->tp_status == TP_STATUS_LIBTRACE) {
+                if (!block) {
+                        return 0;
+                }
+
                 if ((ret=is_halted(libtrace)) != -1)
                         return ret;
@@ -594 +603 @@
         /* We just need to get prepare_packet to set all our packet pointers
          * appropriately */
-        if (linuxring_prepare_packet(libtrace, packet, packet->buffer,
+        if (_linuxring_prepare_packet(packet, packet->buffer,
                                 packet->type, 0))
                 return -1;
@@ -603 +612 @@
 
 static int linuxring_read_packet(libtrace_t *libtrace, libtrace_packet_t *packet) {
-        return linuxring_read_stream(libtrace, packet, FORMAT_DATA_FIRST, NULL);
+        return linuxring_read_stream(libtrace, packet, FORMAT_DATA_FIRST,
+                        NULL, 1);
 }
 
@@ -610 +620 @@
                                    libtrace_thread_t *t,
                                    libtrace_packet_t *packets[],
-                                   UNUSED size_t nb_packets) {
-        /* For now just read one packet */
-        /* If we change this to actually read nb_packets, make sure
-         * we remove the burst_size override in linuxring_pstart_input()
-         */
-        packets[0]->error = linuxring_read_stream(libtrace, packets[0],
-                                                  t->format_data, &t->messages);
-        if (packets[0]->error >= 1)
-                return 1;
-        else
-                return packets[0]->error;
+                                   size_t nb_packets) {
+        size_t i;
+
+        for (i = 0; i < nb_packets; i++) {
+                packets[i]->error = linuxring_read_stream(libtrace, packets[i],
+                                t->format_data, &t->messages, i == 0 ? 1 : 0);
+
+                if (packets[i]->error < 0) {
+                        return packets[i]->error;
+                }
+
+                if (packets[i]->error == 0) {
+                        return i;
+                }
+        }
+
+        return nb_packets;
 }
 #endif