Implements some timeouts.

Author: larsbrinkhoff

src/ncp.c

@@ -15,6 +15,10 @@
 #include "imp.h"
 #include "wire.h"
 
+#define RFNM_TIMEOUT   10
+#define RRP_TIMEOUT    20
+#define ERP_TIMEOUT    20
+
 #define IMP_REGULAR       0
 #define IMP_LEADER_ERROR  1
 #define IMP_DOWN          2
@@ -71,6 +75,7 @@ static int fd;
 static struct sockaddr_un server;
 static struct sockaddr_un client;
 static socklen_t len;
+static unsigned long time_unit;
 
 typedef struct
 { 
@@ -86,9 +91,11 @@ struct
   int listen;
   struct { int link, size; uint32_t lsock, rsock; } rcv, snd;
   void (*rrp_callback) (int);
-  void (*rrp_timeout) (int);
+  void (*rrp_timed_out) (int);
+  unsigned long rrp_timeout;
   void (*rfnm_callback) (int);
-  void (*rfnm_timeout) (int);
+  void (*rfnm_timed_out) (int);
+  unsigned long rfnm_timeout;
 } connection[CONNECTIONS];
 
 struct
@@ -104,6 +111,7 @@ static struct
 #define HOST_ALIVE   0001
 
   client_t echo;
+  unsigned long erp_timeout;
   int outstanding_rfnm;
 } hosts[256];
 
@@ -128,10 +136,12 @@ static const char *type_name[] =
 static uint8_t packet[200];
 static uint8_t app[200];
 
-static void when_rrp (int i, void (*cb) (int), void (*to) (int))
+static void when_rrp (int i, void (*cb) (int),
+                      unsigned timeout, void (*to) (int))
 {
   connection[i].rrp_callback = cb;
-  connection[i].rrp_timeout = to;
+  connection[i].rrp_timed_out = to;
+  connection[i].rrp_timeout = time_unit + timeout;
 }
 
 static void check_rrp (int host)
@@ -141,18 +151,20 @@ static void check_rrp (int host)
   for (i = 0; i < CONNECTIONS; i++) {
     if (connection[i].host != host)
       continue;
-    if (connection[i].rrp_callback == NULL)
-      continue;
     cb = connection[i].rrp_callback;
+    if (cb == NULL)
+      continue;
     connection[i].rrp_callback = NULL;
     cb (i);
   }
 }
 
-static void when_rfnm (int i, void (*cb) (int), void (*to) (int))
+static void when_rfnm (int i, void (*cb) (int),
+                       unsigned timeout, void (*to) (int))
 {
   connection[i].rfnm_callback = cb;
-  connection[i].rfnm_timeout = to;
+  connection[i].rfnm_timed_out = to;
+  connection[i].rfnm_timeout = time_unit + timeout;
 }
 
 static void check_rfnm (int host)
@@ -312,6 +324,8 @@ static int make_open (int host,
   connection[i].snd.lsock = snd_lsock;
   connection[i].snd.rsock = snd_rsock;
   connection[i].flags = 0;
+  connection[i].rrp_timeout = time_unit - 1;
+  connection[i].rfnm_timeout = time_unit - 1;
 
   return i;
 }
@@ -730,12 +744,13 @@ static int process_cls (uint8_t source, uint8_t *data)
 
 static void just_drop (int i)
 {
+  fprintf (stderr, "NCP: RFNM timeout, drop connection %d.\n", i);
   destroy (i);
 }
 
 static void cls_and_drop (int i)
 {
-  fprintf (stderr, "NCP: Timeout, drop connection %d.\n", i);
+  fprintf (stderr, "NCP: RFNM timeout, close connection %d.\n", i);
   ncp_cls (connection[i].host,
            connection[i].snd.lsock, connection[i].snd.rsock);
   ncp_cls (connection[i].host,
@@ -764,7 +779,7 @@ static void send_str_and_rts (int i)
              connection[i].snd.rsock, connection[i].snd.size);
     connection[i].flags |= CONN_SENT_STR;
   }
-  when_rfnm (i, send_rts, cls_and_drop);
+  when_rfnm (i, send_rts, RFNM_TIMEOUT, cls_and_drop);
   check_rfnm (connection[i].host);
 }
 
@@ -825,8 +840,8 @@ static int process_all (uint8_t source, uint8_t *data)
     fprintf (stderr, "NCP: New connection %d.\n", j);
     connection[j].snd.size = 8;
     connection[j].rcv.link = 44;
-    when_rfnm (j, send_str_and_rts, cls_and_drop);
-    when_rfnm (i, send_cls_snd, just_drop);
+    when_rfnm (j, send_str_and_rts, RFNM_TIMEOUT, cls_and_drop);
+    when_rfnm (i, send_cls_snd, RFNM_TIMEOUT, just_drop);
     check_rfnm (source);
   }
   return 7;
@@ -1062,7 +1077,7 @@ static void process_regular (uint8_t *packet, int length)
     if (connection[i].flags & CONN_CLIENT) {
       uint32_t s = sock (&packet[9]);
       fprintf (stderr, "NCP: ICP link %u socket %u.\n", link, s);
-      when_rfnm (i, send_cls_rcv, just_drop);
+      when_rfnm (i, send_cls_rcv, RFNM_TIMEOUT, just_drop);
       connection[i].snd.rsock = s;
 
       j = find_rcv_sockets (source, connection[i].rcv.lsock+2, s+1);
@@ -1075,7 +1090,7 @@ static void process_regular (uint8_t *packet, int length)
         connection[j].snd.size = 8;
         connection[j].rcv.link = 45;
         fprintf (stderr, "NCP: New connection %d.\n", j);
-        when_rfnm (j, send_str_and_rts, cls_and_drop);
+        when_rfnm (j, send_str_and_rts, RFNM_TIMEOUT, cls_and_drop);
       }
       connection[j].listen = connection[i].rcv.rsock;
       connection[j].flags |= CONN_GOT_SOCKET;
@@ -1269,6 +1284,7 @@ static void app_echo (void)
 
   memcpy (&hosts[host].echo.addr, &client, len);
   hosts[host].echo.len = len;
+  hosts[host].erp_timeout = time_unit + ERP_TIMEOUT;
   ncp_eco (host, app[2]);
 }
 
@@ -1281,6 +1297,7 @@ static void app_open_send_rts (int i)
 
 static void app_open_fail (int i)
 {
+  fprintf (stderr, "NCP: Timed out waiting for RRP.\n");
   reply_open (connection[i].host, connection[i].rcv.rsock, 255);
 }
 
@@ -1305,7 +1322,7 @@ static void app_open (void)
     // We haven't communicated with this host yet.
     ncp_rst (host);
     // Wait for RRP, then send RTS.
-    when_rrp (i, app_open_send_rts, app_open_fail);
+    when_rrp (i, app_open_send_rts, RRP_TIMEOUT, app_open_fail);
   } else {
     // Ok to send RTS directly.
     app_open_send_rts (i);
@@ -1417,6 +1434,35 @@ static void application (void)
   }
 }
 
+static void tick (void)
+{
+  void (*to) (int);
+  int i;
+  time_unit++;
+  for (i = 0; i < CONNECTIONS; i++) {
+    to = connection[i].rrp_timed_out;
+    if (to != NULL && connection[i].rrp_timeout == time_unit) {
+      connection[i].rrp_timed_out = NULL;
+      connection[i].rrp_timeout = time_unit - 1;
+      to (i);
+    }
+    to = connection[i].rfnm_timed_out;
+    if (to != NULL && connection[i].rfnm_timeout == time_unit) {
+      connection[i].rfnm_timed_out = NULL;
+      connection[i].rrp_timeout = time_unit - 1;
+      to (i);
+    }
+  }
+  for (i = 0; i < 256; i++) {
+    if (hosts[i].echo.len == 0)
+      continue;
+    if (hosts[i].erp_timeout != time_unit)
+      continue;
+    reply_echo (i, 0, 0x20);
+    hosts[i].echo.len = 0;
+  }
+}
+
 static void cleanup (void)
 {
   unlink (server.sun_path);
@@ -1444,6 +1490,7 @@ void ncp_init (void)
   signal (SIGQUIT, sigcleanup);
   signal (SIGTERM, sigcleanup);
   atexit (cleanup);
+  time_unit = 0;
 }
 
 int main (int argc, char **argv)
@@ -1456,13 +1503,20 @@ int main (int argc, char **argv)
   for (;;) {
     int n;
     fd_set rfds;
+    struct timeval tv;
     FD_ZERO (&rfds);
     FD_SET (fd, &rfds);
     imp_fd_set (&rfds);
-    n = select (33, &rfds, NULL, NULL, NULL);
+    tv.tv_sec = 1;
+    tv.tv_usec = 0;
+    n = select (33, &rfds, NULL, NULL, &tv);
     if (n == -1)
       fprintf (stderr, "NCP: select error.\n");
-    else if (n > 0) {
+    else if (n == 0) {
+      tick ();
+      tv.tv_sec = 1;
+      tv.tv_usec = 0;
+    } else {
       if (imp_fd_isset (&rfds)) {
         memset (packet, 0, sizeof packet);
         imp_receive_message (packet, &n);