Merged in dstaesse/ouroboros/be-cancel (pull request #609)

ipcpd: Cancel threads in IPCPs

3 files changed, 85 insertions, 72 deletions

diff --git a/src/ipcpd/local/main.c b/src/ipcpd/local/main.c
index 7f44a2c6..b98e144c 100644
--- a/src/ipcpd/local/main.c
+++ b/src/ipcpd/local/main.c
@@ -44,8 +44,8 @@
 #include <sys/wait.h>
 #include <assert.h>
 
-#define EVENT_WAIT_TIMEOUT 100 /* us */
-#define THIS_TYPE IPCP_LOCAL
+#define THIS_TYPE     IPCP_LOCAL
+#define ALLOC_TIMEOUT 10 /* ms */
 
 struct {
         struct shim_data * shim_data;
@@ -95,15 +95,13 @@ static void local_data_fini(void){
 
 static void * ipcp_local_sdu_loop(void * o)
 {
-        struct timespec timeout = {0, EVENT_WAIT_TIMEOUT * 1000};
-
         (void) o;
 
-        while (ipcp_get_state() == IPCP_OPERATIONAL) {
+        while (true) {
                 int     fd;
                 ssize_t idx;
 
-                fevent(local_data.flows, local_data.fq, &timeout);
+                fevent(local_data.flows, local_data.fq, NULL);
 
                 while ((fd = fqueue_next(local_data.fq)) >= 0) {
                         idx = local_flow_read(fd);
@@ -181,7 +179,7 @@ static int ipcp_local_flow_alloc(int             fd,
                                  const uint8_t * dst,
                                  qoscube_t       cube)
 {
-        struct timespec ts     = {0, EVENT_WAIT_TIMEOUT * 1000};
+        struct timespec ts     = {0, ALLOC_TIMEOUT * MILLION};
         struct timespec abstime;
         int             out_fd = -1;
 
@@ -237,7 +235,7 @@ static int ipcp_local_flow_alloc(int             fd,
 static int ipcp_local_flow_alloc_resp(int fd,
                                       int response)
 {
-        struct timespec ts     = {0, EVENT_WAIT_TIMEOUT * 1000};
+        struct timespec ts     = {0, ALLOC_TIMEOUT * MILLION};
         struct timespec abstime;
         int             out_fd = -1;
         int             ret    = -1;
@@ -359,8 +357,10 @@ int main(int    argc,
 
         ipcp_shutdown();
 
-        if (ipcp_get_state() == IPCP_SHUTDOWN)
+        if (ipcp_get_state() == IPCP_SHUTDOWN) {
+                pthread_cancel(local_data.sduloop);
                 pthread_join(local_data.sduloop, NULL);
+        }
 
         local_data_fini();
 
diff --git a/src/ipcpd/shim-eth-llc/main.c b/src/ipcpd/shim-eth-llc/main.c
index 36d101e5..ca0c8166 100644
--- a/src/ipcpd/shim-eth-llc/main.c
+++ b/src/ipcpd/shim-eth-llc/main.c
@@ -102,7 +102,7 @@
 #define ETH_FRAME_SIZE            (ETH_HEADER_SIZE + LLC_HEADER_SIZE \
                                    + SHIM_ETH_LLC_MAX_SDU_SIZE)
 #define SHIM_ETH_LLC_MAX_SDU_SIZE (1500 - LLC_HEADER_SIZE)
-#define EVENT_WAIT_TIMEO          10000 /* us */
+#define ALLOC_TIMEO               10    /* ms */
 #define NAME_QUERY_TIMEO          2000  /* ms */
 #define MGMT_TIMEO                100   /* ms */
 
@@ -421,7 +421,7 @@ static int eth_llc_ipcp_sap_req(uint8_t         r_sap,
                                 const uint8_t * dst,
                                 qoscube_t       cube)
 {
-        struct timespec ts = {0, EVENT_WAIT_TIMEO * 100};
+        struct timespec ts = {0, ALLOC_TIMEO * MILLION};
         struct timespec abstime;
         int             fd;
 
@@ -585,12 +585,12 @@ static void * eth_llc_ipcp_mgmt_handler(void * o)
 
         (void) o;
 
+        pthread_cleanup_push((void (*)(void *)) pthread_mutex_unlock,
+                             (void *) &eth_llc_data.mgmt_lock);
+
         while (true) {
                 ret = 0;
 
-                if (ipcp_get_state() != IPCP_OPERATIONAL)
-                        return (void *) 0;
-
                 clock_gettime(PTHREAD_COND_CLOCK, &abstime);
                 ts_add(&abstime, &timeout, &abstime);
 
@@ -620,6 +620,8 @@ static void * eth_llc_ipcp_mgmt_handler(void * o)
                 eth_llc_ipcp_mgmt_frame(frame->buf, frame->len, frame->r_addr);
                 free(frame);
         }
+
+        pthread_cleanup_pop(false);
 }
 
 static void * eth_llc_ipcp_sdu_reader(void * o)
@@ -646,10 +648,8 @@ static void * eth_llc_ipcp_sdu_reader(void * o)
         memset(br_addr, 0xff, MAC_SIZE * sizeof(uint8_t));
 
         while (true) {
-                if (ipcp_get_state() != IPCP_OPERATIONAL)
-                        return (void *) 0;
 #if defined(HAVE_NETMAP)
-                if (poll(&eth_llc_data.poll_in, 1, EVENT_WAIT_TIMEO / 1000) < 0)
+                if (poll(&eth_llc_data.poll_in, 1, -1) < 0)
                         continue;
                 if (eth_llc_data.poll_in.revents == 0) /* TIMED OUT */
                         continue;
@@ -740,7 +740,6 @@ static void * eth_llc_ipcp_sdu_reader(void * o)
 
 static void * eth_llc_ipcp_sdu_writer(void * o)
 {
-        struct timespec      timeout = {0, EVENT_WAIT_TIMEO * 1000};
         int                  fd;
         struct shm_du_buff * sdb;
         uint8_t              ssap;
@@ -749,9 +748,8 @@ static void * eth_llc_ipcp_sdu_writer(void * o)
 
         (void) o;
 
-        while (fevent(eth_llc_data.np1_flows, eth_llc_data.fq, &timeout)) {
-                if (ipcp_get_state() != IPCP_OPERATIONAL)
-                        return (void *) 0;
+        while (true) {
+                fevent(eth_llc_data.np1_flows, eth_llc_data.fq, NULL);
 
                 pthread_rwlock_rdlock(&eth_llc_data.flows_lock);
                 while ((fd = fqueue_next(eth_llc_data.fq)) >= 0) {
@@ -843,7 +841,7 @@ static void * eth_llc_ipcp_if_monitor(void * o)
                 return (void *) -1;
         }
 
-        while (ipcp_get_state() == IPCP_OPERATIONAL) {
+        while (true) {
                 status = recvmsg(fd, &msg, 0);
                 if (status < 0)
                         continue;
@@ -916,9 +914,6 @@ static int eth_llc_ipcp_bootstrap(const struct ipcp_config * conf)
         int              enable  = 1;
         int              disable = 0;
         int              blen;
-        struct timeval   tv = {0, EVENT_WAIT_TIMEO};
-#elif defined(HAVE_RAW_SOCKETS)
-        struct timeval   tv = {0, EVENT_WAIT_TIMEO};
 #endif /* HAVE_NETMAP */
 
 #if defined(__FreeBSD__) || defined(__APPLE__)
@@ -1037,11 +1032,6 @@ static int eth_llc_ipcp_bootstrap(const struct ipcp_config * conf)
                 goto fail_device;
         }
 
-        if (ioctl(eth_llc_data.bpf, BIOCSRTIMEOUT, &tv) < 0) {
-                log_err("Failed to set BIOCSRTIMEOUT.");
-                goto fail_device;
-        }
-
         if (ioctl(eth_llc_data.bpf, BIOCIMMEDIATE, &enable) < 0) {
                 log_err("Failed to set BIOCIMMEDIATE.");
                 goto fail_device;
@@ -1070,12 +1060,6 @@ static int eth_llc_ipcp_bootstrap(const struct ipcp_config * conf)
                 goto fail_device;
         }
 
-        if (setsockopt(eth_llc_data.s_fd, SOL_SOCKET, SO_RCVTIMEO,
-                       &tv, sizeof(tv))) {
-                log_err("Failed to set socket timeout: %s.", strerror(errno));
-                goto fail_device;
-        }
-
 #endif /* HAVE_NETMAP */
         ipcp_set_state(IPCP_OPERATIONAL);
 
@@ -1119,11 +1103,14 @@ static int eth_llc_ipcp_bootstrap(const struct ipcp_config * conf)
         return 0;
 
  fail_sdu_writer:
+        pthread_cancel(eth_llc_data.sdu_reader);
         pthread_join(eth_llc_data.sdu_reader, NULL);
  fail_sdu_reader:
+        pthread_cancel(eth_llc_data.mgmt_handler);
         pthread_join(eth_llc_data.mgmt_handler, NULL);
  fail_mgmt_handler:
 #ifdef __linux__
+        pthread_cancel(eth_llc_data.if_monitor);
         pthread_join(eth_llc_data.if_monitor, NULL);
 #endif
  fail_device:
@@ -1246,7 +1233,7 @@ static int eth_llc_ipcp_flow_alloc(int             fd,
 static int eth_llc_ipcp_flow_alloc_resp(int fd,
                                         int response)
 {
-        struct timespec ts    = {0, EVENT_WAIT_TIMEO * 100};
+        struct timespec ts    = {0, ALLOC_TIMEO * MILLION};
         struct timespec abstime;
         uint8_t         ssap  = 0;
         uint8_t         r_sap = 0;
@@ -1369,6 +1356,12 @@ int main(int    argc,
         ipcp_shutdown();
 
         if (ipcp_get_state() == IPCP_SHUTDOWN) {
+                pthread_cancel(eth_llc_data.sdu_writer);
+                pthread_cancel(eth_llc_data.sdu_reader);
+                pthread_cancel(eth_llc_data.mgmt_handler);
+#ifdef __linux__
+                pthread_cancel(eth_llc_data.if_monitor);
+#endif
                 pthread_join(eth_llc_data.sdu_writer, NULL);
                 pthread_join(eth_llc_data.sdu_reader, NULL);
                 pthread_join(eth_llc_data.mgmt_handler, NULL);
diff --git a/src/ipcpd/shim-udp/main.c b/src/ipcpd/shim-udp/main.c
index 8f4bd80d..f47bb01a 100644
--- a/src/ipcpd/shim-udp/main.c
+++ b/src/ipcpd/shim-udp/main.c
@@ -399,16 +399,10 @@ static void * ipcp_udp_listener(void * o)
         ssize_t            n   = 0;
         struct sockaddr_in c_saddr;
         int                sfd = udp_data.s_fd;
-        struct timeval     ltv = {(SOCKET_TIMEOUT / 1000),
-                                  (SOCKET_TIMEOUT % 1000) * 1000};
 
         (void) o;
 
-        if (setsockopt(sfd, SOL_SOCKET, SO_RCVTIMEO,
-                       (void *) &ltv, sizeof(ltv)))
-                log_warn("Failed to set timeout on socket.");
-
-        while (ipcp_get_state() == IPCP_OPERATIONAL) {
+        while (true) {
                 shim_udp_msg_t * msg = NULL;
 
                 memset(&buf, 0, SHIM_UDP_MSG_SIZE);
@@ -456,19 +450,19 @@ static void * ipcp_udp_listener(void * o)
 
 static void * ipcp_udp_sdu_reader(void * o)
 {
-        ssize_t n;
-        int skfd;
-        int fd;
+        ssize_t            n;
+        int                skfd;
+        int                fd;
         /* FIXME: avoid this copy */
-        char buf[SHIM_UDP_MAX_SDU_SIZE];
+        char               buf[SHIM_UDP_MAX_SDU_SIZE];
         struct sockaddr_in r_saddr;
-        fd_set read_fds;
-        int flags;
-        struct timeval tv = {0, FD_UPDATE_TIMEOUT};
+        struct timeval     tv = {0, FD_UPDATE_TIMEOUT};
+        fd_set             read_fds;
+        int                flags;
 
         (void) o;
 
-        while (ipcp_get_state() == IPCP_OPERATIONAL) {
+        while (true) {
                 pthread_rwlock_rdlock(&udp_data.flows_lock);
                 pthread_mutex_lock(&udp_data.fd_set_lock);
 
@@ -512,13 +506,12 @@ static void * ipcp_udp_sdu_reader(void * o)
 static void * ipcp_udp_sdu_loop(void * o)
 {
         int fd;
-        struct timespec timeout = {0, FD_UPDATE_TIMEOUT * 1000};
         struct shm_du_buff * sdb;
 
         (void) o;
 
-        while (ipcp_get_state() == IPCP_OPERATIONAL) {
-                fevent(udp_data.np1_flows, udp_data.fq, &timeout);
+        while (true) {
+                fevent(udp_data.np1_flows, udp_data.fq, NULL);
                 while ((fd = fqueue_next(udp_data.fq)) >= 0) {
                         if (ipcp_flow_read(fd, &sdb)) {
                                 log_err("Bad read from fd %d.", fd);
@@ -531,12 +524,15 @@ static void * ipcp_udp_sdu_loop(void * o)
 
                         pthread_rwlock_unlock(&udp_data.flows_lock);
 
+                        pthread_cleanup_push((void (*)(void *)) ipcp_sdb_release,
+                                             (void *) sdb);
+
                         if (send(fd, shm_du_buff_head(sdb),
                                  shm_du_buff_tail(sdb) - shm_du_buff_head(sdb),
                                  0) < 0)
                                 log_err("Failed to send SDU.");
 
-                        ipcp_sdb_release(sdb);
+                        pthread_cleanup_pop(true);
                 }
         }
 
@@ -570,7 +566,7 @@ static int ipcp_udp_bootstrap(const struct ipcp_config * conf)
                         log_err("Failed to convert DNS address");
                         return -1;
                 }
-#ifndef CONFIG_OUROBOROS_ENABLE_DNS
+#ifndef HAVE_DDNS
                 log_warn("DNS disabled at compile time, address ignored");
 #endif
         } else {
@@ -580,7 +576,7 @@ static int ipcp_udp_bootstrap(const struct ipcp_config * conf)
         /* UDP listen server */
         if ((fd = socket(AF_INET, SOCK_DGRAM, IPPROTO_UDP)) == -1) {
                 log_err("Can't create socket.");
-                return -1;
+                goto fail_socket;
         }
 
         if (setsockopt(fd,
@@ -599,8 +595,7 @@ static int ipcp_udp_bootstrap(const struct ipcp_config * conf)
                  (struct sockaddr *) &udp_data.s_saddr,
                  sizeof(udp_data.s_saddr)) < 0) {
                 log_err("Couldn't bind to %s.", ipstr);
-                close(fd);
-                return -1;
+                goto fail_bind;
         }
 
         udp_data.s_fd     = fd;
@@ -611,25 +606,46 @@ static int ipcp_udp_bootstrap(const struct ipcp_config * conf)
 
         ipcp_set_state(IPCP_OPERATIONAL);
 
-        pthread_create(&udp_data.handler,
-                       NULL,
-                       ipcp_udp_listener,
-                       NULL);
-        pthread_create(&udp_data.sdu_reader,
-                       NULL,
-                       ipcp_udp_sdu_reader,
-                       NULL);
+        if (pthread_create(&udp_data.handler,
+                           NULL,
+                           ipcp_udp_listener,
+                           NULL)) {
+                ipcp_set_state(IPCP_INIT);
+                goto fail_bind;
+        }
+
+        if (pthread_create(&udp_data.sdu_reader,
+                           NULL,
+                           ipcp_udp_sdu_reader,
+                           NULL)) {
+                ipcp_set_state(IPCP_INIT);
+                goto fail_sdu_reader;
+        }
 
-        pthread_create(&udp_data.sduloop,
-                       NULL,
-                       ipcp_udp_sdu_loop,
-                       NULL);
+        if (pthread_create(&udp_data.sduloop,
+                           NULL,
+                           ipcp_udp_sdu_loop,
+                           NULL)) {
+                ipcp_set_state(IPCP_INIT);
+                goto fail_sduloop;
+        }
 
         log_dbg("Bootstrapped shim IPCP over UDP with api %d.", getpid());
         log_dbg("Bound to IP address %s.", ipstr);
         log_dbg("DNS server address is %s.", dnsstr);
 
         return 0;
+
+ fail_sduloop:
+        pthread_cancel(udp_data.sdu_reader);
+        pthread_join(udp_data.sdu_reader, NULL);
+ fail_sdu_reader:
+        pthread_cancel(udp_data.handler);
+        pthread_join(udp_data.handler, NULL);
+ fail_bind:
+        close(fd);
+ fail_socket:
+        return -1;
 }
 
 #ifdef HAVE_DDNS
@@ -1128,6 +1144,10 @@ int main(int    argc,
         ipcp_shutdown();
 
         if (ipcp_get_state() == IPCP_SHUTDOWN) {
+                pthread_cancel(udp_data.sduloop);
+                pthread_cancel(udp_data.handler);
+                pthread_cancel(udp_data.sdu_reader);
+
                 pthread_join(udp_data.sduloop, NULL);
                 pthread_join(udp_data.handler, NULL);
                 pthread_join(udp_data.sdu_reader, NULL);