Changeset 3004d6c

Timestamp:

02/14/18 17:11:08 (3 years ago)

Author:

Shane Alcock <salcock@…>

Branches:

cachetimestamps, develop, etsilive, master, rc-4.0.3, rc-4.0.4, ringdecrementfix, ringperformance

Children:

a857389

Parents:

7ebf768

Message:

format_ndag: use select to skip sockets with no data

A non-blocking call to recvmmsg still requires a lot of initial
effort, e.g. setting up buffers etc., but there is no need to do
so if we use select() to tell us in advance which sockets are
worth trying.

File:

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

lib/format_ndag.c

@@ -113 +113 @@
         uint64_t missing_records;
         uint64_t received_packets;
+
+        fd_set allsocks;
+        int maxfd;
 } recvstream_t;
 
@@ -319 +322 @@
 
 }
-        
+
 static int ndag_parse_control_message(libtrace_t *libtrace, char *msgbuf,
                 int msgsize, uint16_t *ptmap) {
@@ -485 +488 @@
                 FORMAT_DATA->receivers[i].received_packets = 0;
                 FORMAT_DATA->receivers[i].missing_records = 0;
+                FD_ZERO(&(FORMAT_DATA->receivers[i].allsocks));
+                FORMAT_DATA->receivers[i].maxfd = -1;
 
                 libtrace_message_queue_init(&(FORMAT_DATA->receivers[i].mqueue),
@@ -766 +771 @@
         ssock->recordcount = 0;
         rt->sourcecount += 1;
+        FD_SET(ssock->sock, &(rt->allsocks));
+        if (ssock->sock > rt->maxfd) {
+                rt->maxfd = ssock->sock;
+        }
 
         fprintf(stderr, "Added new stream %s:%u to thread %d\n",
@@ -872 +881 @@
                 fprintf(stderr, "Received invalid record on the channel for %s:%u.\n",
                                 ssock->groupaddr, ssock->port);
+                FD_CLR(ssock->sock, &(rt->allsocks));
                 close(ssock->sock);
                 ssock->sock = -1;
@@ -977 +987 @@
                                         ssock->port);
 
+                                FD_CLR(ssock->sock, &(rt->allsocks));
                                 close(ssock->sock);
                                 ssock->sock = -1;
@@ -986 +997 @@
                                 ssock->groupaddr, ssock->port,
                                 strerror(errno));
+                        FD_CLR(ssock->sock, &(rt->allsocks));
                         close(ssock->sock);
                         ssock->sock = -1;
@@ -1022 +1034 @@
         int i, readybufs, gottime;
         struct timeval tv;
+        fd_set fds;
+        struct timeval zerotv;
 
         readybufs = 0;
         gottime = 0;
 
-        for (i = 0; i < rt->sourcecount; i ++) {
+        fds = rt->allsocks;
+
+        if (rt->maxfd == -1) {
+                return 0;
+        }
+
+        zerotv.tv_sec = 0;
+        zerotv.tv_usec = 0;
+
+        if (select(rt->maxfd + 1, &fds, NULL, NULL, &zerotv) == -1) {
+                /* log the error? XXX */
+                return -1;
+        }
+
+        for (i = 0; i < rt->sourcecount; i++) {
+                if (!FD_ISSET(rt->sources[i].sock, &fds)) {
+                        continue;
+                }
                 readybufs += receive_from_single_socket(&(rt->sources[i]),
                                 &tv, &gottime, rt);
@@ -1125 +1156 @@
                 }
                 return NULL;
-        }       
+        }
 
 
@@ -1220 +1251 @@
         } while (1);
 
-        
         for (i = 0; i < rt->sourcecount; i++) {
                 streamsock_t *src = &(rt->sources[i]);