Changeset b15cc48

Timestamp:

11/25/16 14:50:28 (4 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:

c94f107

Parents:

e3b6476

Message:

Fixes bug in DPDK trace_event handling

Reading small bursts of packets is not supported by DPDK and hangs.
Instead we read a larger burst and cache, which is better for performance also.

Fixes #29 broken tracetop with DPDK

File:

• lib/format_dpdk.c (modified) (3 diffs)

lib/format_dpdk.c

@@ -237 +237 @@
 /* For single threaded libtrace we read packets as a batch/burst
  * this is the maximum size of said burst */
-#define BURST_SIZE 50
+#define BURST_SIZE 32
 
 #define MBUF(x) ((struct rte_mbuf *) x)
@@ -2254 +2254 @@
                                             libtrace_packet_t *packet) {
         libtrace_eventobj_t event = {0,0,0.0,0};
-        int nb_rx; /* Number of receive packets we've read */
-        struct rte_mbuf* pkts_burst[1]; /* Array of 1 pointer(s) to rx buffers */
+        size_t nb_rx; /* Number of received packets we've read */
 
         do {
 
-                /* See if we already have a packet waiting */
-                nb_rx = rte_eth_rx_burst(FORMAT(trace)->port,
-                                         FORMAT_DATA_FIRST(trace)->queue_id,
-                                         pkts_burst, 1);
-
-                if (nb_rx > 0) {
+                /* No packets waiting in our buffer? Try and read some more */
+                if (FORMAT(trace)->burst_size == FORMAT(trace)->burst_offset) {
+                        nb_rx = rte_eth_rx_burst(FORMAT(trace)->port,
+                                                 FORMAT_DATA_FIRST(trace)->queue_id,
+                                                 FORMAT(trace)->burst_pkts, BURST_SIZE);
+                        if (nb_rx > 0) {
+                                dpdk_ready_pkts(trace, FORMAT_DATA_FIRST(trace),
+                                                FORMAT(trace)->burst_pkts, nb_rx);
+                                FORMAT(trace)->burst_size = nb_rx;
+                                FORMAT(trace)->burst_offset = 0;
+                        }
+                }
+
+                /* Now do we have packets waiting? */
+                if (FORMAT(trace)->burst_size != FORMAT(trace)->burst_offset) {
                         /* Free the last packet buffer */
                         if (packet->buffer != NULL) {
@@ -2276 +2284 @@
                         packet->type = TRACE_RT_DATA_DPDK;
                         event.type = TRACE_EVENT_PACKET;
-                        dpdk_ready_pkts(trace, FORMAT_DATA_FIRST(trace), pkts_burst, 1);
-                        packet->buffer = FORMAT(trace)->burst_pkts[0];
+                        packet->buffer = FORMAT(trace)->burst_pkts[
+                                             FORMAT(trace)->burst_offset++];
                         dpdk_prepare_packet(trace, packet, packet->buffer, packet->type, 0);
                         event.size = 1; // TODO should be bytes read, which essentially useless anyway