Changeset 639d952 for lib

Timestamp:

11/13/17 17:24:27 (3 years ago)

Author:

Shane Alcock <salcock@…>

Branches:

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

Children:

5d8280a

Parents:

b34f924

Message:

Minor performance tweaks for format_ndag

Most notably, split receive_encap_records() into two separate
versions (blocking and nonblocking) which should avoid a few
"if (block)" checks.

Also removed extraneous call to receiver_read_messages when using
the pread function.

File:

• lib/format_ndag.c (modified) (11 diffs)

lib/format_ndag.c

@@ -736 +736 @@
                 return 0;
         }
+        /*
         if (ssock->nextread - ssock->saved[ssock->nextreadind] >=
                         ssock->savedsize[ssock->nextreadind]) {
                 return 0;
         }
+        */
         return 1;
 
@@ -862 +864 @@
         int avail, ret, ndagstat, i;
         int toret = 0;
-        struct timespec ts;
 
         if (ssock->sock == -1) {
@@ -876 +877 @@
         }
 
-        ts.tv_sec = 0;
-        ts.tv_nsec = 10000;
-
         ret = recvmmsg(ssock->sock, ssock->mmsgbufs, avail,
-                        MSG_DONTWAIT, &ts);
-
-        /*
-           ret = recvfrom(rt->sources[i].sock, rt->sources[i].saved[nw],
-           ENCAP_BUFSIZE, MSG_DONTWAIT,
-           rt->sources[i].srcaddr->ai_addr,
-           &(rt->sources[i].srcaddr->ai_addrlen));
-         */
+                        MSG_DONTWAIT, NULL);
+
         if (ret < 0) {
                 /* Nothing to receive right now, but we should still
@@ -955 +947 @@
 
 
-static int receive_encap_records(libtrace_t *libtrace, recvstream_t *rt,
-                libtrace_packet_t *packet, int block) {
+static int receive_encap_records_block(libtrace_t *libtrace, recvstream_t *rt,
+                libtrace_packet_t *packet) {
 
         int iserr = 0;
@@ -972 +964 @@
 
                 /* Check for any messages from the control thread */
-                if (block) {
-                        iserr = receiver_read_messages(rt);
-
-                        if (iserr <= 0) {
-                                return iserr;
-                        }
-                }
-
-                /* If non-blocking and there are no sources, just break */
-                if (!block && rt->sourcecount == 0) {
-                        iserr = 0;
-                        break;
+                iserr = receiver_read_messages(rt);
+
+                if (iserr <= 0) {
+                        return iserr;
                 }
 
@@ -989 +973 @@
                  * checking for messages again.
                  */
-                if (block && rt->sourcecount == 0) {
+                if (rt->sourcecount == 0) {
                         usleep(10000);
                         continue;
@@ -1005 +989 @@
                  * a short break rather than immediately trying again.
                  */
-                if (block && iserr == 0) {
+                if (iserr == 0) {
                         usleep(100);
                 }
 
-        } while (block);
+        } while (1);
 
         return iserr;
+}
+
+static int receive_encap_records_nonblock(libtrace_t *libtrace, recvstream_t *rt,
+                libtrace_packet_t *packet) {
+
+        int iserr = 0;
+
+        if (packet->buf_control == TRACE_CTRL_PACKET) {
+                free(packet->buffer);
+                packet->buffer = NULL;
+        }
+
+        /* Make sure we shouldn't be halting */
+        if ((iserr = is_halted(libtrace)) != -1) {
+                return iserr;
+        }
+
+        /* If non-blocking and there are no sources, just break */
+        if (rt->sourcecount == 0) {
+                return 0;
+        }
+
+        return receive_from_sockets(rt);
 }
 
@@ -1047 +1054 @@
         int rem;
         streamsock_t *nextavail = NULL;
-        rem = receive_encap_records(libtrace, &(FORMAT_DATA->receivers[0]),
-                        packet, 1);
+        rem = receive_encap_records_block(libtrace, &(FORMAT_DATA->receivers[0]),
+                        packet);
 
         if (rem <= 0) {
@@ -1077 +1084 @@
         rt = (recvstream_t *)t->format_data;
 
-        /* Only check for messages once per batch */
-        rem = receiver_read_messages(rt);
-        if (rem <= 0) {
-                return rem;
-        }
 
         do {
-                rem = receive_encap_records(libtrace, rt,
-                                packets[read_packets],
-                                read_packets == 0 ? 1 : 0);
+                /* Only check for messages once per batch */
+                if (read_packets == 0) {
+                        rem = receive_encap_records_block(libtrace, rt,
+                                packets[read_packets]);
+                } else {
+                        rem = receive_encap_records_nonblock(libtrace, rt,
+                                packets[read_packets]);
+                }
+
                 if (rem < 0) {
                         return rem;
@@ -1122 +1130 @@
         streamsock_t *nextavail = NULL;
 
-        /* Only check for messages once per batch */
+        /* Only check for messages once per call */
         rem = receiver_read_messages(&(FORMAT_DATA->receivers[0]));
         if (rem <= 0) {
@@ -1130 +1138 @@
 
         do {
-                rem = receive_encap_records(libtrace,
-                                &(FORMAT_DATA->receivers[0]), packet, 0);
+                rem = receive_encap_records_nonblock(libtrace,
+                                &(FORMAT_DATA->receivers[0]), packet);
 
                 if (rem < 0) {