Changeset aa7db84

Timestamp:

11/07/17 13:46:40 (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:

4e427da

Parents:

8fa0167

Message:

Added support for seqno wrapping and improved performance.

Performance improvements are as follows:

• Use larger receive buffer for the multicast recv socket.
• Remove select in main receive function with a non-blocking recvfrom.
• Only sleep if no sources have data available, thus avoiding constant 100% CPU usage due to the above change.

File:

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

lib/format_ndag.c

@@ -89 +89 @@
 
 
+static inline int seq_cmp(uint32_t seq_a, uint32_t seq_b) {
+
+        /* Calculate seq_a - seq_b, taking wraparound into account */
+        if (seq_a == seq_b) return 0;
+
+        if (seq_a > seq_b) {
+                return (int) (seq_a - seq_b);
+        }
+        return (int) (0xffffffff - ((seq_b - seq_a) - 1));
+}
+
 static inline uint8_t check_ndag_header(char *msgbuf, uint32_t msgsize) {
         ndag_common_t *header = (ndag_common_t *)msgbuf;
@@ -123 +134 @@
         char pstr[16];
         struct group_req greq;
+        int bufsize;
 
         int sock;
@@ -182 +194 @@
                 fprintf(stderr,
                         "Failed to join multicast group %s:%s -- %s\n",
+                                groupaddr, portstr, strerror(errno));
+                close(sock);
+                sock = -1;
+                goto sockcreateover;
+        }
+
+        bufsize = 16 * 1024 * 1024;
+        if (setsockopt(sock, SOL_SOCKET, SO_RCVBUF, &bufsize,
+                                (socklen_t)sizeof(int)) < 0) {
+
+                fprintf(stderr,
+                        "Failed to increase buffer size for multicast group %s:%s -- %s\n",
                                 groupaddr, portstr, strerror(errno));
                 close(sock);
@@ -637 +661 @@
 static int receive_from_sockets(recvstream_t *rt) {
 
-        fd_set allgroups;
-        int maxfd, i, ret, readybufs, availsocks;
-        struct timeval timeout;
-
-        maxfd = 0;
-        timeout.tv_sec = 0;
-        timeout.tv_usec = 10000;
+        int i, ret, readybufs, availsocks, successrecv;
 
         /* TODO maybe need a way to tidy up "dead" sockets and signal back
@@ -652 +670 @@
         readybufs = 0;
         availsocks = 0;
-
-        FD_ZERO(&allgroups);
+        successrecv = 0;
+
         for (i = 0; i < rt->sourcecount; i ++) {
                 if (read_required(rt->sources[i])) {
-                        FD_SET(rt->sources[i].sock, &allgroups);
-                        if (rt->sources[i].sock > maxfd) {
-                                maxfd = rt->sources[i].sock;
-                        }
                         availsocks += 1;
                 } else if (rt->sources[i].sock != -1) {
@@ -679 +693 @@
          * sockets.
          */
-        if (select(maxfd + 1, &allgroups, NULL, NULL, &timeout) < 0) {
-                fprintf(stderr, "Error waiting to receive records: %s\n",
-                                strerror(errno));
-                return -1;
-        }
-
-
         for (i = 0; i < rt->sourcecount; i ++) {
                 if (!read_required(rt->sources[i])) {
@@ -697 +704 @@
                 rt->sources[i].nextread = rt->sources[i].saved;
 
-                if (!FD_ISSET(rt->sources[i].sock, &allgroups)) {
-                        continue;
-                }
-
                 ret = recvfrom(rt->sources[i].sock, rt->sources[i].saved,
-                                ENCAP_BUFSIZE, 0,
+                                ENCAP_BUFSIZE, MSG_DONTWAIT,
                                 rt->sources[i].srcaddr->ai_addr,
                                 &(rt->sources[i].srcaddr->ai_addrlen));
                 if (ret < 0) {
+                        if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                                continue;
+                        }
+
                         fprintf(stderr,
-                                        "Error receiving encapsulated records from %s:%u -- %s\n",
+                                        "Error receiving encapsulated records from %s:%u -- %s \n",
                                         rt->sources[i].groupaddr,
                                         rt->sources[i].port,
@@ -726 +733 @@
 
                 rt->sources[i].savedsize = ret;
+                successrecv ++;
 
                 /* Check that we have a valid nDAG encap record */
@@ -743 +751 @@
 
                 if (rt->sources[i].expectedseq != 0) {
-                        if (ntohl(rt->sources[i].encaphdr->seqno) !=
-                                        rt->sources[i].expectedseq) {
-                                /* TODO deal with wrapping */
-                                rt->missing_records += (ntohl(rt->sources[i].encaphdr->seqno) - rt->sources[i].expectedseq);
-                        }
+                        rt->missing_records += seq_cmp(
+                                        ntohl(rt->sources[i].encaphdr->seqno),
+                                        rt->sources[i].expectedseq);
                 }
                 rt->sources[i].expectedseq =
@@ -809 +815 @@
                 }
 
+                /* None of our sources have anything available, we can take
+                 * a short break rather than immediately trying again.
+                 */
+                if (block && iserr == 0) {
+                        usleep(100);
+                }
+
         } while (block);
 
@@ -826 +839 @@
                 }
 
-                daghdr = (dag_record_t *)rt->sources[i].nextread;
+                daghdr = (dag_record_t *)(rt->sources[i].nextread);
                 currentts = bswap_le_to_host64(daghdr->ts);