Changeset de825b5

Timestamp:

11/07/17 15:31:47 (4 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:

e68325b

Parents:

4e427da

Message:

Add idle timeout for ndag receiving sockets.

Remove extraneous read_required checks in receive_from_sockets().
This used to make sense when we had a select() that we wanted to
skip whenever possible, but now it is just extra effort for no
benefit.

File:

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

lib/format_ndag.c

@@ -23 +23 @@
 #include "format_ndag.h"
 
+#define NDAG_IDLE_TIMEOUT (600)
 #define ENCAP_BUFSIZE (10000)
 #define CTRL_BUF_SIZE (10000)
@@ -53 +54 @@
         int nextwriteind;
         int savedsize[ENCAP_BUFFERS];
+        uint32_t startidle;
         uint64_t recordcount;
 
@@ -615 +617 @@
         int i;
 
+        /* TODO consider replacing this with a list or vector so we can
+         * easily remove sources that are no longer in use, rather than
+         * just setting the sock to -1 and having to check them every
+         * time we want to read a packet.
+         */
         if (rt->sourcecount == 0) {
                 rt->sources = (streamsock_t *)malloc(sizeof(streamsock_t) * 10);
@@ -636 +643 @@
         ssock->monitorid = src.monitor;
         ssock->saved = (char **)malloc(sizeof(char *) * ENCAP_BUFFERS);
-
+        ssock->startidle = 0;
 
         for (i = 0; i < ENCAP_BUFFERS; i++) {
@@ -708 +715 @@
 static int receive_from_sockets(recvstream_t *rt) {
 
-        int i, ret, readybufs, availsocks, successrecv;
-
-        /* TODO maybe need a way to tidy up "dead" sockets and signal back
-         * to the control thread that we've killed the socket for a particular
-         * port.
-         */
+        int i, ret, readybufs, gottime;
+        struct timeval tv;
 
         readybufs = 0;
-        availsocks = 0;
-        successrecv = 0;
-
-        for (i = 0; i < rt->sourcecount; i ++) {
-                if (read_required(rt->sources[i])) {
-                        availsocks += 1;
-                } else if (rt->sources[i].sock != -1) {
-                        readybufs += 1;
-                        availsocks += 1;
-                }
-        }
-
-        /* If all of our sockets already have data sitting in their
-         * buffers then we can save ourselves some 'select'ing.
-         */
-        if (availsocks == readybufs) {
-                return readybufs;
-        }
-
-        /* Otherwise, at least one active socket has an empty buffer so
-         * we better try to read some data from those sockets (just in
-         * case the correct 'next' packet is waiting on one of those
-         * sockets.
-         */
+        gottime = 0;
+
         for (i = 0; i < rt->sourcecount; i ++) {
                 int nw;
@@ -764 +745 @@
                                         readybufs ++;
                                 }
+                                if (!gottime) {
+                                        gettimeofday(&tv, NULL);
+                                }
+                                if (rt->sources[i].startidle == 0) {
+                                        rt->sources[i].startidle = tv.tv_sec;
+                                } else if (tv.tv_sec - rt->sources[i].startidle > NDAG_IDLE_TIMEOUT) {
+                                        fprintf(stderr, "Closing channel %s:%u due to inactivity.\n",
+                                                        rt->sources[i].groupaddr,
+                                                        rt->sources[i].port);
+
+                                        close(rt->sources[i].sock);
+                                        rt->sources[i].sock = -1;
+                                }
                                 continue;
                         }
@@ -776 +770 @@
                         continue;
                 }
-
-                if (ret == 0) {
-                        fprintf(stderr, "Received zero bytes on the channel for %s:%u.\n",
-                                        rt->sources[i].groupaddr,
-                                        rt->sources[i].port);
-                        close(rt->sources[i].sock);
-                        rt->sources[i].sock = -1;
-                        continue;
-                }
-
-                rt->sources[i].savedsize[nw] = ret;
-                rt->sources[i].nextwriteind =
-                                (rt->sources[i].nextwriteind + 1) % ENCAP_BUFFERS;
-                successrecv ++;
+                rt->sources[i].startidle = 0;
 
                 /* Check that we have a valid nDAG encap record */
@@ -801 +782 @@
                         continue;
                 }
+
+                rt->sources[i].savedsize[nw] = ret;
+                rt->sources[i].nextwriteind =
+                                (rt->sources[i].nextwriteind + 1) % ENCAP_BUFFERS;
 
                 /* Save the useful info from the encap header */