irmd: Revise IRMd internals

This is a full revision of the IRMd internal implementation.

The registry is now a proper subcomponent managing its own internal
lock (a single mutex). Some tests are added for the registry and its
data structures. Some macros for tests are added in <ouroboros/test.h>.

Flow allocation is now more symmetric between the client side (alloc)
and server size (accept). Each will create a flow in pending state
(ALLOC_PENDING/ACCEPT_PENDING) that is potentially fulfilled by an
IPCP using respond_alloc and respond_accept primitives. Deallocation
is split in flow_dealloc (application side) and ipcp_flow_dealloc
(IPCP side) to get the flow in DEALLOC_PENDING and DEALLOCATED state.

Cleanup of failed flow allocation is now properly handled instead of
relying on the sanitizer thread. The new sanitizer only needs to
monitor crashed processes.

On shutdown, the IRMd will now detect hanging processes and SIGKILL
them and clean up their fuse mountpoints if needed.

A lot of other things have been cleaned up and shuffled around a bit.

Signed-off-by: Dimitri Staessens <dimitri@ouroboros.rocks>
Signed-off-by: Sander Vrijders <sander@ouroboros.rocks>

1 files changed, 168 insertions, 226 deletions

diff --git a/src/irmd/ipcp.c b/src/irmd/ipcp.c
index 6f4d2c0f..3253a8f3 100644
--- a/src/irmd/ipcp.c
+++ b/src/irmd/ipcp.c
@@ -26,179 +26,124 @@
 
 #define OUROBOROS_PREFIX "irmd/ipcp"
 
-#include <ouroboros/logs.h>
 #include <ouroboros/errno.h>
-#include <ouroboros/utils.h>
+#include <ouroboros/flow.h>
+#include <ouroboros/logs.h>
 #include <ouroboros/sockets.h>
+#include <ouroboros/time.h>
+#include <ouroboros/utils.h>
 
 #include "ipcp.h"
 
-#include <stdlib.h>
-#include <string.h>
+#include <fcntl.h>
+#include <pthread.h>
 #include <signal.h>
+#include <spawn.h>
 #include <stdbool.h>
-#include <pthread.h>
+#include <stdlib.h>
+#include <string.h>
 #include <sys/types.h>
 #include <sys/wait.h>
 #include <sys/socket.h>
 #include <sys/time.h>
-#include <spawn.h>
 
 ipcp_msg_t * send_recv_ipcp_msg(pid_t        pid,
                                 ipcp_msg_t * msg)
 {
-       int            sockfd    = 0;
-       uint8_t        buf[SOCK_BUF_SIZE];
-       char *         sock_path = NULL;
-       ssize_t        len;
-       ipcp_msg_t *   recv_msg  = NULL;
-       struct timeval tv;
-
-       if (kill(pid, 0) < 0)
-               return NULL;
-
-       sock_path = ipcp_sock_path(pid);
-       if (sock_path == NULL)
-               return NULL;
-
-       sockfd = client_socket_open(sock_path);
-       if (sockfd < 0) {
-               free(sock_path);
-               return NULL;
-       }
-
-       free(sock_path);
-
-       len = ipcp_msg__get_packed_size(msg);
-       if (len == 0) {
-               close(sockfd);
-               return NULL;
-       }
-
-       switch (msg->code) {
-       case IPCP_MSG_CODE__IPCP_BOOTSTRAP:
-               tv.tv_sec  = BOOTSTRAP_TIMEOUT / 1000;
-               tv.tv_usec = (BOOTSTRAP_TIMEOUT % 1000) * 1000;
-               break;
-       case IPCP_MSG_CODE__IPCP_ENROLL:
-               tv.tv_sec  = ENROLL_TIMEOUT / 1000;
-               tv.tv_usec = (ENROLL_TIMEOUT % 1000) * 1000;
-               break;
-       case IPCP_MSG_CODE__IPCP_REG:
-               tv.tv_sec  = REG_TIMEOUT / 1000;
-               tv.tv_usec = (REG_TIMEOUT % 1000) * 1000;
-               break;
-       case IPCP_MSG_CODE__IPCP_QUERY:
-               tv.tv_sec  = QUERY_TIMEOUT / 1000;
-               tv.tv_usec = (QUERY_TIMEOUT % 1000) * 1000;
-               break;
-       case IPCP_MSG_CODE__IPCP_CONNECT:
-               tv.tv_sec  = CONNECT_TIMEOUT / 1000;
-               tv.tv_usec = (CONNECT_TIMEOUT % 1000) * 1000;
-               break;
-       default:
-               tv.tv_sec  = SOCKET_TIMEOUT / 1000;
-               tv.tv_usec = (SOCKET_TIMEOUT % 1000) * 1000;
-
-               break;
-       }
-
-       if (setsockopt(sockfd, SOL_SOCKET, SO_RCVTIMEO,
-                      (void *) &tv, sizeof(tv)))
-               log_warn("Failed to set timeout on socket.");
-
-       pthread_cleanup_push(__cleanup_close_ptr, (void *) &sockfd);
-
-       ipcp_msg__pack(msg, buf);
-
-       if (write(sockfd, buf, len) != -1)
-               len = read(sockfd, buf, SOCK_BUF_SIZE);
-
-       if (len > 0)
-               recv_msg = ipcp_msg__unpack(NULL, len, buf);
-
-       pthread_cleanup_pop(true);
-
-       return recv_msg;
-}
+        int             sockfd;
+        uint8_t         buf[SOCK_BUF_SIZE];
+        char *          sock_path;
+        ssize_t         len;
+        ipcp_msg_t *    recv_msg;
+        struct timeval  tv;
+        struct timespec tic;
+        struct timespec toc;
+
+        if (kill(pid, 0) < 0)
+                return NULL;
+
+        sock_path = ipcp_sock_path(pid);
+        if (sock_path == NULL)
+                return NULL;
+
+        sockfd = client_socket_open(sock_path);
+        if (sockfd < 0) {
+                free(sock_path);
+                return NULL;
+        }
 
-pid_t ipcp_create(const struct ipcp_info * info)
-{
-        pid_t  pid;
-        char * exec_name = NULL;
-        char   irmd_pid[10];
-        char   full_name[256];
-        char * argv[5];
-
-        switch(info->type) {
-        case IPCP_UNICAST:
-                exec_name = IPCP_UNICAST_EXEC;
-                break;
-        case IPCP_BROADCAST:
-                exec_name = IPCP_BROADCAST_EXEC;
+        free(sock_path);
+
+        len = ipcp_msg__get_packed_size(msg);
+        if (len == 0) {
+                close(sockfd);
+                return NULL;
+        }
+
+        switch (msg->code) {
+        case IPCP_MSG_CODE__IPCP_BOOTSTRAP:
+                tv.tv_sec  = BOOTSTRAP_TIMEOUT / 1000;
+                tv.tv_usec = (BOOTSTRAP_TIMEOUT % 1000) * 1000;
                 break;
-        case IPCP_UDP:
-                exec_name = IPCP_UDP_EXEC;
+        case IPCP_MSG_CODE__IPCP_ENROLL:
+                tv.tv_sec  = ENROLL_TIMEOUT / 1000;
+                tv.tv_usec = (ENROLL_TIMEOUT % 1000) * 1000;
                 break;
-        case IPCP_ETH_LLC:
-                exec_name = IPCP_ETH_LLC_EXEC;
+        case IPCP_MSG_CODE__IPCP_REG:
+                tv.tv_sec  = REG_TIMEOUT / 1000;
+                tv.tv_usec = (REG_TIMEOUT % 1000) * 1000;
                 break;
-        case IPCP_ETH_DIX:
-                exec_name = IPCP_ETH_DIX_EXEC;
+        case IPCP_MSG_CODE__IPCP_QUERY:
+                tv.tv_sec  = QUERY_TIMEOUT / 1000;
+                tv.tv_usec = (QUERY_TIMEOUT % 1000) * 1000;
                 break;
-        case IPCP_LOCAL:
-                exec_name = IPCP_LOCAL_EXEC;
+        case IPCP_MSG_CODE__IPCP_CONNECT:
+                tv.tv_sec  = CONNECT_TIMEOUT / 1000;
+                tv.tv_usec = (CONNECT_TIMEOUT % 1000) * 1000;
                 break;
         default:
-                return -1;
+                tv.tv_sec  = SOCKET_TIMEOUT / 1000;
+                tv.tv_usec = (SOCKET_TIMEOUT % 1000) * 1000;
+                break;
         }
 
-        if (exec_name == NULL) {
-                log_err("IPCP type not installed.");
-                return -1;
-        }
+        if (setsockopt(sockfd, SOL_SOCKET, SO_RCVTIMEO,
+                        (void *) &tv, sizeof(tv)))
+                log_warn("Failed to set timeout on socket.");
 
-        sprintf(irmd_pid, "%u", getpid());
+        pthread_cleanup_push(__cleanup_close_ptr, (void *) &sockfd);
 
-        strcpy(full_name, INSTALL_PREFIX"/"INSTALL_SBINDIR"/");
-        strcat(full_name, exec_name);
+        ipcp_msg__pack(msg, buf);
 
-        /* log_file to be placed at the end */
-        argv[0] = full_name;
-        argv[1] = irmd_pid;
-        argv[2] = (char *) info->name;
-        if (log_syslog)
-                argv[3] = "1";
-        else
-                argv[3] = NULL;
+        clock_gettime(CLOCK_REALTIME, &tic);
 
-        argv[4] = NULL;
+        if (write(sockfd, buf, len) != -1)
+                len = read(sockfd, buf, SOCK_BUF_SIZE);
 
-        if (posix_spawn(&pid, argv[0], NULL, NULL, argv, NULL) != 0) {
-                log_err("Failed to spawn new process");
-                return -1;
-        }
+        clock_gettime(CLOCK_REALTIME, &toc);
 
-        return pid;
-}
+        pthread_cleanup_pop(true); /* close socket */
 
-int ipcp_destroy(pid_t pid)
-{
-        if (kill(pid, SIGTERM)) {
-                log_err("Failed to destroy IPCP");
-                return -1;
+        if (len > 0)
+                recv_msg = ipcp_msg__unpack(NULL, len, buf);
+        else {
+                if (errno == EAGAIN) {
+                        int diff = ts_diff_ms(&tic, &toc);
+                        log_warn("IPCP command timed out after %d ms.", diff);
+                }
+                return NULL;
         }
 
-        return 0;
+        return recv_msg;
 }
 
 int ipcp_bootstrap(pid_t                pid,
                    struct ipcp_config * conf,
                    struct layer_info *  info)
 {
-        ipcp_msg_t   msg      = IPCP_MSG__INIT;
-        ipcp_msg_t * recv_msg = NULL;
-        int          ret      = -1;
+        ipcp_msg_t   msg = IPCP_MSG__INIT;
+        ipcp_msg_t * recv_msg;
+        int          ret;
 
         if (conf == NULL)
                 return -EINVAL;
@@ -240,9 +185,9 @@ int ipcp_enroll(pid_t               pid,
                 const char *        dst,
                 struct layer_info * info)
 {
-        ipcp_msg_t   msg      = IPCP_MSG__INIT;
-        ipcp_msg_t * recv_msg = NULL;
-        int          ret      = -1;
+        ipcp_msg_t   msg = IPCP_MSG__INIT;
+        ipcp_msg_t * recv_msg;
+        int          ret;
 
         if (dst == NULL)
                 return -EINVAL;
@@ -283,9 +228,9 @@ int ipcp_connect(pid_t        pid,
                  const char * component,
                  qosspec_t    qs)
 {
-        ipcp_msg_t    msg    = IPCP_MSG__INIT;
-        int           ret    = -1;
+        ipcp_msg_t    msg = IPCP_MSG__INIT;
         ipcp_msg_t *  recv_msg;
+        int           ret;
 
         msg.code    = IPCP_MSG_CODE__IPCP_CONNECT;
         msg.dst     = (char *) dst;
@@ -314,9 +259,9 @@ int ipcp_disconnect(pid_t        pid,
                     const char * dst,
                     const char * component)
 {
-        ipcp_msg_t   msg      = IPCP_MSG__INIT;
-        ipcp_msg_t * recv_msg = NULL;
-        int          ret      = -1;
+        ipcp_msg_t   msg = IPCP_MSG__INIT;
+        ipcp_msg_t * recv_msg;
+        int          ret;
 
         msg.code    = IPCP_MSG_CODE__IPCP_DISCONNECT;
         msg.dst     = (char *) dst;
@@ -339,20 +284,17 @@ int ipcp_disconnect(pid_t        pid,
         return ret;
 }
 
-int ipcp_reg(pid_t           pid,
-             const uint8_t * hash,
-             size_t          len)
+int ipcp_reg(pid_t          pid,
+             const buffer_t hash)
 {
-        ipcp_msg_t   msg      = IPCP_MSG__INIT;
-        ipcp_msg_t * recv_msg = NULL;
-        int          ret      = -1;
-
-        assert(hash);
+        ipcp_msg_t   msg = IPCP_MSG__INIT;
+        ipcp_msg_t * recv_msg;
+        int          ret;
 
         msg.code      = IPCP_MSG_CODE__IPCP_REG;
         msg.has_hash  = true;
-        msg.hash.len  = len;
-        msg.hash.data = (uint8_t *)hash;
+        msg.hash.data = (uint8_t *) hash.data;
+        msg.hash.len  = hash.len;
 
         recv_msg = send_recv_ipcp_msg(pid, &msg);
         if (recv_msg == NULL)
@@ -369,18 +311,17 @@ int ipcp_reg(pid_t           pid,
         return ret;
 }
 
-int ipcp_unreg(pid_t           pid,
-               const uint8_t * hash,
-               size_t          len)
+int ipcp_unreg(pid_t          pid,
+               const buffer_t hash)
 {
-        ipcp_msg_t   msg      = IPCP_MSG__INIT;
-        ipcp_msg_t * recv_msg = NULL;
-        int          ret      = -1;
+        ipcp_msg_t   msg = IPCP_MSG__INIT;
+        ipcp_msg_t * recv_msg;
+        int          ret;
 
         msg.code      = IPCP_MSG_CODE__IPCP_UNREG;
         msg.has_hash  = true;
-        msg.hash.len  = len;
-        msg.hash.data = (uint8_t *) hash;
+        msg.hash.data = (uint8_t *) hash.data;
+        msg.hash.len  = hash.len;
 
         recv_msg = send_recv_ipcp_msg(pid, &msg);
         if (recv_msg == NULL)
@@ -397,18 +338,17 @@ int ipcp_unreg(pid_t           pid,
         return ret;
 }
 
-int ipcp_query(pid_t           pid,
-               const uint8_t * hash,
-               size_t          len)
+int ipcp_query(pid_t          pid,
+               const buffer_t dst)
 {
-        ipcp_msg_t   msg      = IPCP_MSG__INIT;
-        ipcp_msg_t * recv_msg = NULL;
-        int          ret      = -1;
+        ipcp_msg_t   msg = IPCP_MSG__INIT;
+        ipcp_msg_t * recv_msg;
+        int          ret;
 
         msg.code      = IPCP_MSG_CODE__IPCP_QUERY;
         msg.has_hash  = true;
-        msg.hash.len  = len;
-        msg.hash.data = (uint8_t *) hash;
+        msg.hash.data = (uint8_t *) dst.data;
+        msg.hash.len  = dst.len;
 
         recv_msg = send_recv_ipcp_msg(pid, &msg);
         if (recv_msg == NULL)
@@ -425,36 +365,24 @@ int ipcp_query(pid_t           pid,
         return ret;
 }
 
-static int __ipcp_flow_alloc(pid_t           pid,
-                             int             flow_id,
-                             pid_t           n_pid,
-                             const uint8_t * dst,
-                             size_t          len,
-                             qosspec_t       qs,
-                             bool            join,
-                             const buffer_t  data)
+int ipcp_flow_join(const struct flow_info * flow,
+                   const buffer_t           dst)
 {
-        ipcp_msg_t    msg      = IPCP_MSG__INIT;
-        ipcp_msg_t *  recv_msg = NULL;
-        int           ret      = -1;
-
-        assert(dst);
+        ipcp_msg_t   msg = IPCP_MSG__INIT;
+        ipcp_msg_t * recv_msg;
+        int          ret;
 
-        msg.code = join ? IPCP_MSG_CODE__IPCP_FLOW_JOIN
-                        : IPCP_MSG_CODE__IPCP_FLOW_ALLOC;
+        msg.code         = IPCP_MSG_CODE__IPCP_FLOW_JOIN;
         msg.has_flow_id  = true;
-        msg.flow_id      = flow_id;
+        msg.flow_id      = flow->id;
         msg.has_pid      = true;
-        msg.pid          = n_pid;
+        msg.pid          = flow->n_pid;
         msg.has_hash     = true;
-        msg.hash.len     = len;
-        msg.hash.data    = (uint8_t *) dst;
-        msg.qosspec      = qos_spec_s_to_msg(&qs);;
-        msg.has_pk       = true;
-        msg.pk.data      = data.data;
-        msg.pk.len       = data.len;
+        msg.hash.data    = (uint8_t *) dst.data;
+        msg.hash.len     = dst.len;
+        msg.has_pk       = false;
 
-        recv_msg = send_recv_ipcp_msg(pid, &msg);
+        recv_msg = send_recv_ipcp_msg(flow->n_1_pid, &msg);
         free(msg.qosspec);
         if (recv_msg == NULL)
                 return -EIPCP;
@@ -470,52 +398,66 @@ static int __ipcp_flow_alloc(pid_t           pid,
         return ret;
 }
 
-int ipcp_flow_alloc(pid_t           pid,
-                    int             flow_id,
-                    pid_t           n_pid,
-                    const uint8_t * dst,
-                    size_t          len,
-                    qosspec_t       qs,
-                    const buffer_t  data)
+int ipcp_flow_alloc(const struct flow_info * flow,
+                    const buffer_t           dst,
+                    const buffer_t           data)
 {
-        return __ipcp_flow_alloc(pid, flow_id, n_pid, dst,
-                                 len, qs, false, data);
-}
+        ipcp_msg_t   msg = IPCP_MSG__INIT;
+        ipcp_msg_t * recv_msg;
+        int          ret;
 
-int ipcp_flow_join(pid_t           pid,
-                   int             flow_id,
-                   pid_t           n_pid,
-                   const uint8_t * dst,
-                   size_t          len,
-                   qosspec_t       qs)
-{
-        buffer_t data = {NULL, 0};
-        return __ipcp_flow_alloc(pid, flow_id, n_pid, dst,
-                                 len, qs, true, data);
+        msg.code         = IPCP_MSG_CODE__IPCP_FLOW_ALLOC;
+        msg.has_flow_id  = true;
+        msg.flow_id      = flow->id;
+        msg.has_pid      = true;
+        msg.pid          = flow->n_pid;
+        msg.qosspec      = qos_spec_s_to_msg(&flow->qs);
+        msg.has_hash     = true;
+        msg.hash.data    = (uint8_t *) dst.data;
+        msg.hash.len     = dst.len;
+        msg.has_pk       = true;
+        msg.pk.data      = data.data;
+        msg.pk.len       = data.len;
+
+        recv_msg = send_recv_ipcp_msg(flow->n_1_pid, &msg);
+        free(msg.qosspec);
+        if (recv_msg == NULL) {
+                log_err("Did not receive message.");
+                return -EIPCP;
+        }
+
+        if (!recv_msg->has_result) {
+                log_err("Message has no result");
+                ipcp_msg__free_unpacked(recv_msg, NULL);
+                return -EIPCP;
+        }
+
+        ret = recv_msg->result;
+        ipcp_msg__free_unpacked(recv_msg, NULL);
+
+        return ret;
 }
 
-int ipcp_flow_alloc_resp(pid_t          pid,
-                         int            flow_id,
-                         pid_t          n_pid,
-                         int            response,
-                         const buffer_t data)
+int ipcp_flow_alloc_resp(const struct flow_info * flow,
+                         int                      response,
+                         const buffer_t           data)
 {
-        ipcp_msg_t   msg      = IPCP_MSG__INIT;
-        ipcp_msg_t * recv_msg = NULL;
-        int          ret      = -1;
+        ipcp_msg_t   msg = IPCP_MSG__INIT;
+        ipcp_msg_t * recv_msg;
+        int          ret;
 
         msg.code         = IPCP_MSG_CODE__IPCP_FLOW_ALLOC_RESP;
         msg.has_flow_id  = true;
-        msg.flow_id      = flow_id;
+        msg.flow_id      = flow->id;
         msg.has_pid      = true;
-        msg.pid          = n_pid;
+        msg.pid          = flow->n_pid;
         msg.has_response = true;
         msg.response     = response;
         msg.has_pk       = true;
         msg.pk.data      = data.data;
         msg.pk.len       = data.len;
 
-        recv_msg = send_recv_ipcp_msg(pid, &msg);
+        recv_msg = send_recv_ipcp_msg(flow->n_1_pid, &msg);
         if (recv_msg == NULL)
                 return -EIPCP;
 
@@ -534,9 +476,9 @@ int ipcp_flow_dealloc(pid_t  pid,
                       int    flow_id,
                       time_t timeo)
 {
-        ipcp_msg_t   msg      = IPCP_MSG__INIT;
-        ipcp_msg_t * recv_msg = NULL;
-        int          ret      = -1;
+        ipcp_msg_t   msg = IPCP_MSG__INIT;
+        ipcp_msg_t * recv_msg;
+        int          ret;
 
         msg.code          = IPCP_MSG_CODE__IPCP_FLOW_DEALLOC;
         msg.has_flow_id   = true;