irmd: Graceful shutdown

Threads from the threadpool will now check the state of the IRMd and
exit gracefully without a need to cancel them. This avoids the unsafe
call of pthread_cancel in the signal handler.

6 files changed, 83 insertions, 80 deletions

diff --git a/include/ouroboros/config.h.in b/include/ouroboros/config.h.in
index 46685569..60654676 100644
--- a/include/ouroboros/config.h.in
+++ b/include/ouroboros/config.h.in
@@ -47,6 +47,7 @@
 #define SHM_AP_RBUFF_PREFIX    "/ouroboros.rbuff."
 #define IRMD_MAX_FLOWS         4096
 #define IRMD_THREADPOOL_SIZE   5
+#define IRMD_ACCEPT_TIMEOUT    100 /* ms */
 #define IRMD_FLOW_TIMEOUT      5000 /* ms */
 #define LOG_DIR                "/@LOG_DIR@/"
 
diff --git a/src/irmd/irm_flow.c b/src/irmd/irm_flow.c
index bf67f2ce..d9fe3fb3 100644
--- a/src/irmd/irm_flow.c
+++ b/src/irmd/irm_flow.c
@@ -65,13 +65,10 @@ void irm_flow_destroy(struct irm_flow * f)
 
         pthread_cond_signal(&f->state_cond);
 
-        pthread_cleanup_push((void (*)(void *)) pthread_mutex_unlock,
-                             (void *) &f->state_lock);
-
         while (f->state != FLOW_NULL)
                 pthread_cond_wait(&f->state_cond, &f->state_lock);
 
-        pthread_cleanup_pop(true);
+        pthread_mutex_unlock(&f->state_lock);
 
         pthread_cond_destroy(&f->state_cond);
         pthread_mutex_destroy(&f->state_lock);
diff --git a/src/irmd/main.c b/src/irmd/main.c
index f9f34416..c2058ed5 100644
--- a/src/irmd/main.c
+++ b/src/irmd/main.c
@@ -67,8 +67,7 @@ struct ipcp_entry {
 
 enum irm_state {
         IRMD_NULL = 0,
-        IRMD_RUNNING,
-        IRMD_SHUTDOWN
+        IRMD_RUNNING
 };
 
 struct spawned_api {
@@ -718,7 +717,14 @@ static struct irm_flow * flow_accept(pid_t   api,
         pthread_rwlock_unlock(&irmd->reg_lock);
         pthread_rwlock_unlock(&irmd->state_lock);
 
-        reg_api_sleep(rgi);
+        while (reg_api_sleep(rgi) == -ETIMEDOUT) {
+                pthread_rwlock_rdlock(&irmd->state_lock);
+                if (irmd->state != IRMD_RUNNING) {
+                        pthread_rwlock_unlock(&irmd->state_lock);
+                        break;
+                }
+                pthread_rwlock_unlock(&irmd->state_lock);
+        }
 
         pthread_rwlock_rdlock(&irmd->state_lock);
         pthread_rwlock_rdlock(&irmd->reg_lock);
@@ -887,15 +893,6 @@ static struct irm_flow * flow_alloc(pid_t  api,
         return f;
 }
 
-static void cleanup_alloc_res(void * o)
-{
-        struct irm_flow * f = (struct irm_flow *) o;
-        if (f->state == FLOW_PENDING)
-                f->state = FLOW_NULL;
-        pthread_cond_broadcast(&f->state_cond);
-        pthread_mutex_unlock(&f->state_lock);
-}
-
 static int flow_alloc_res(int port_id)
 {
         struct irm_flow * f;
@@ -933,12 +930,11 @@ static int flow_alloc_res(int port_id)
         pthread_rwlock_unlock(&irmd->state_lock);
 
         pthread_mutex_lock(&f->state_lock);
-        pthread_cleanup_push(cleanup_alloc_res, (void *) f);
 
         while (f->state == FLOW_PENDING)
                 pthread_cond_wait(&f->state_cond, &f->state_lock);
 
-        pthread_cleanup_pop(true);
+        pthread_mutex_unlock(&f->state_lock);
 
         pthread_rwlock_rdlock(&irmd->state_lock);
         pthread_rwlock_wrlock(&irmd->flows_lock);
@@ -1103,13 +1099,11 @@ static struct irm_flow * flow_req_arr(pid_t  api,
                 pthread_rwlock_unlock(&irmd->state_lock);
 
                 pthread_mutex_lock(&rne->state_lock);
-                pthread_cleanup_push((void (*)(void *)) pthread_mutex_unlock,
-                                     (void *) &rne->state_lock);
 
                 while (rne->state == REG_NAME_AUTO_EXEC)
                         pthread_cond_wait(&rne->state_cond, &rne->state_lock);
 
-                pthread_cleanup_pop(true);
+                pthread_mutex_unlock(&rne->state_lock);
 
                 pthread_rwlock_rdlock(&irmd->state_lock);
                 pthread_rwlock_rdlock(&irmd->reg_lock);
@@ -1165,13 +1159,11 @@ static struct irm_flow * flow_req_arr(pid_t  api,
         reg_api_wake(rgi);
 
         pthread_mutex_lock(&rne->state_lock);
-        pthread_cleanup_push((void (*)(void *)) pthread_mutex_unlock,
-                             (void *) &rne->state_lock);
 
         while (rne->state == REG_NAME_FLOW_ARRIVED)
                 pthread_cond_wait(&rne->state_cond, &rne->state_lock);
 
-        pthread_cleanup_pop(true);
+        pthread_mutex_unlock(&rne->state_lock);
 
         return f;
 }
@@ -1304,26 +1296,17 @@ static void irm_destroy()
 
 void irmd_sig_handler(int sig, siginfo_t * info, void * c)
 {
-        int i;
-
         switch(sig) {
         case SIGINT:
         case SIGTERM:
         case SIGHUP:
+                LOG_INFO("IRMd shutting down...");
+
                 pthread_rwlock_wrlock(&irmd->state_lock);
 
                 irmd->state = IRMD_NULL;
 
                 pthread_rwlock_unlock(&irmd->state_lock);
-
-                if (irmd->threadpool != NULL) {
-                        for (i = 0; i < IRMD_THREADPOOL_SIZE; i++)
-                                pthread_cancel(irmd->threadpool[i]);
-
-                }
-
-                pthread_cancel(irmd->shm_sanitize);
-                pthread_cancel(irmd->cleanup_flows);
                 break;
         case SIGPIPE:
                 LOG_DBG("Ignored SIGPIPE.");
@@ -1349,7 +1332,7 @@ void * irm_flow_cleaner()
 
                 pthread_rwlock_rdlock(&irmd->state_lock);
 
-                if (irmd->state == IRMD_NULL) {
+                if (irmd->state != IRMD_RUNNING) {
                         pthread_rwlock_unlock(&irmd->state_lock);
                         return (void *) 0;
                 }
@@ -1427,16 +1410,6 @@ void * irm_flow_cleaner()
         }
 }
 
-static void clean_msg(void * msg)
-{
-        irm_msg__free_unpacked(msg, NULL);
-}
-
-static void close_ptr(void * o)
-{
-        close(*((int *) o));
-}
-
 void * mainloop()
 {
         uint8_t buf[IRM_MSG_BUF_SIZE];
@@ -1450,13 +1423,18 @@ void * mainloop()
                 struct irm_flow * e = NULL;
                 pid_t * apis = NULL;
 
+                pthread_rwlock_rdlock(&irmd->state_lock);
+                if (irmd->state != IRMD_RUNNING) {
+                        pthread_rwlock_unlock(&irmd->state_lock);
+                        break;
+                }
+                pthread_rwlock_unlock(&irmd->state_lock);
+
                 ret_msg.code = IRM_MSG_CODE__IRM_REPLY;
 
                 cli_sockfd = accept(irmd->sockfd, 0, 0);
-                if (cli_sockfd < 0) {
-                        LOG_ERR("Cannot accept new connection.");
+                if (cli_sockfd < 0)
                         continue;
-                }
 
                 count = read(cli_sockfd, buf, IRM_MSG_BUF_SIZE);
                 if (count <= 0) {
@@ -1471,9 +1449,6 @@ void * mainloop()
                         continue;
                 }
 
-                pthread_cleanup_push(close_ptr, &cli_sockfd);
-                pthread_cleanup_push(clean_msg, (void *) msg);
-
                 switch (msg->code) {
                 case IRM_MSG_CODE__IRM_CREATE_IPCP:
                         ret_msg.has_result = true;
@@ -1601,8 +1576,7 @@ void * mainloop()
                         break;
                 }
 
-                pthread_cleanup_pop(true);
-                pthread_cleanup_pop(false);
+                irm_msg__free_unpacked(msg, NULL);
 
                 buffer.len = irm_msg__get_packed_size(&ret_msg);
                 if (buffer.len == 0) {
@@ -1632,11 +1606,15 @@ void * mainloop()
                 free(buffer.data);
                 close(cli_sockfd);
         }
+
+        return (void *) 0;
 }
 
 static struct irm * irm_create()
 {
         struct stat st = {0};
+        struct timeval timeout = {(IRMD_ACCEPT_TIMEOUT / 1000),
+                                  (IRMD_ACCEPT_TIMEOUT % 1000) * 1000};
 
         irmd = malloc(sizeof(*irmd));
         if (irmd == NULL)
@@ -1693,6 +1671,16 @@ static struct irm * irm_create()
                 return NULL;
         }
 
+        if (setsockopt(irmd->sockfd,
+                       SOL_SOCKET,
+                       SO_RCVTIMEO,
+                       (char *) &timeout,
+                       sizeof(timeout)) < 0) {
+                LOG_ERR("Failed setting socket option.");
+                irm_destroy();
+                return NULL;
+        }
+
         if (chmod(IRM_SOCK_PATH, 0666)) {
                 LOG_ERR("Failed to chmod socket.");
                 irm_destroy();
@@ -1846,12 +1834,16 @@ int main(int argc, char ** argv)
         for (t = 0; t < IRMD_THREADPOOL_SIZE; ++t)
                 pthread_join(irmd->threadpool[t], NULL);
 
-        pthread_join(irmd->shm_sanitize, NULL);
         pthread_join(irmd->cleanup_flows, NULL);
 
+        pthread_cancel(irmd->shm_sanitize);
+        pthread_join(irmd->shm_sanitize, NULL);
+
         irm_destroy();
 
         close_logfile();
 
+        LOG_INFO("Bye.");
+
         exit(EXIT_SUCCESS);
 }
diff --git a/src/irmd/reg_api.c b/src/irmd/reg_api.c
index 648dc1b3..5b7750f7 100644
--- a/src/irmd/reg_api.c
+++ b/src/irmd/reg_api.c
@@ -21,10 +21,15 @@
  * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
  */
 
+#include <ouroboros/config.h>
+#include <ouroboros/time_utils.h>
+#include <ouroboros/errno.h>
+
 #include "reg_api.h"
 
 #include <stdlib.h>
 #include <stdbool.h>
+#include <time.h>
 
 struct reg_api * reg_api_create(pid_t api)
 {
@@ -51,18 +56,18 @@ void reg_api_destroy(struct reg_api * i)
 
         pthread_mutex_lock(&i->state_lock);
 
+        if (i->state == REG_I_INIT)
+                i->state = REG_I_NULL;
+
         if (i->state != REG_I_NULL)
                 i->state = REG_I_DESTROY;
 
         pthread_cond_signal(&i->state_cond);
 
-        pthread_cleanup_push((void(*)(void *)) pthread_mutex_unlock,
-                             (void *) &i->state_lock);
-
         while (i->state != REG_I_NULL)
                 pthread_cond_wait(&i->state_cond, &i->state_lock);
 
-        pthread_cleanup_pop(true);
+        pthread_mutex_unlock(&i->state_lock);
 
         pthread_cond_destroy(&i->state_cond);
         pthread_mutex_destroy(&i->state_lock);
@@ -70,32 +75,45 @@ void reg_api_destroy(struct reg_api * i)
         free(i);
 }
 
-static void cleanup_sleeper(void * o) {
-        struct reg_api * i = (struct reg_api *) o;
-        i->state = REG_I_NULL;
-        pthread_cond_broadcast(&i->state_cond);
-        pthread_mutex_unlock(&i->state_lock);
-}
-
-void reg_api_sleep(struct reg_api * i)
+int reg_api_sleep(struct reg_api * i)
 {
+        struct timespec timeout = {(IRMD_ACCEPT_TIMEOUT / 1000),
+                                   (IRMD_ACCEPT_TIMEOUT % 1000) * MILLION};
+        struct timespec now;
+        struct timespec dl;
+
+        int ret = 0;
+
         if (i == NULL)
-                return;
+                return -EINVAL;
+
+        clock_gettime(CLOCK_REALTIME, &now);
+
+        ts_add(&now, &timeout, &dl);
 
         pthread_mutex_lock(&i->state_lock);
         if (i->state != REG_I_INIT) {
                 pthread_mutex_unlock(&i->state_lock);
-                return;
+                return -EINVAL;
         }
 
         i->state = REG_I_SLEEP;
 
-        pthread_cleanup_push(cleanup_sleeper, (void *) i);
+        while (i->state == REG_I_SLEEP) {
+                if ((ret = -pthread_cond_timedwait(&i->state_cond,
+                                                   &i->state_lock,
+                                                   &dl)) == -ETIMEDOUT) {
+                        i->state = REG_I_INIT;
+                        break;
+                } else {
+                        i->state = REG_I_NULL;
+                        pthread_cond_broadcast(&i->state_cond);
+                }
+        }
 
-        while (i->state == REG_I_SLEEP)
-                pthread_cond_wait(&i->state_cond, &i->state_lock);
+        pthread_mutex_unlock(&i->state_lock);
 
-        pthread_cleanup_pop(true);
+        return ret;
 }
 
 void reg_api_wake(struct reg_api * i)
@@ -111,11 +129,8 @@ void reg_api_wake(struct reg_api * i)
 
         pthread_cond_broadcast(&i->state_cond);
 
-        pthread_cleanup_push((void(*)(void *)) pthread_mutex_unlock,
-                             (void *) &i->state_lock);
-
         while (i->state == REG_I_WAKE)
                 pthread_cond_wait(&i->state_cond, &i->state_lock);
 
-        pthread_cleanup_pop(true);
+        pthread_mutex_unlock(&i->state_lock);
 }
diff --git a/src/irmd/reg_api.h b/src/irmd/reg_api.h
index 73e1141b..2c84f4c1 100644
--- a/src/irmd/reg_api.h
+++ b/src/irmd/reg_api.h
@@ -49,7 +49,7 @@ struct reg_api {
 
 struct reg_api * reg_api_create(pid_t pid);
 void             reg_api_destroy(struct reg_api * i);
-void             reg_api_sleep(struct reg_api * i);
+int              reg_api_sleep(struct reg_api * i);
 void             reg_api_wake(struct reg_api * i);
 
 #endif /* OUROBOROS_IRMD_REG_API_H */
diff --git a/src/irmd/registry.c b/src/irmd/registry.c
index 04f60009..687ff30d 100644
--- a/src/irmd/registry.c
+++ b/src/irmd/registry.c
@@ -22,8 +22,6 @@
 
 #include "registry.h"
 
-#include <ouroboros/config.h>
-
 #define OUROBOROS_PREFIX "registry"
 
 #include <ouroboros/errno.h>