Merged in dstaesse/ouroboros/be-connmgr (pull request #559)

ipcpd: Improve locking in connmgr

5 files changed, 113 insertions, 102 deletions

diff --git a/src/ipcpd/normal/connmgr.c b/src/ipcpd/normal/connmgr.c
index f48175fa..b7ba90d9 100644
--- a/src/ipcpd/normal/connmgr.c
+++ b/src/ipcpd/normal/connmgr.c
@@ -60,15 +60,39 @@ struct {
         pthread_t        acceptor;
 
         struct list_head aes;
-        pthread_mutex_t  aes_lock;
+        pthread_rwlock_t aes_lock;
 } connmgr;
 
-static int add_ae_conn(struct ae *        ae,
+
+static int get_info_by_name(const char *       name,
+                            struct conn_info * info)
+{
+        struct list_head * p;
+
+        pthread_rwlock_rdlock(&connmgr.aes_lock);
+
+        list_for_each(p, &connmgr.aes) {
+                struct ae * ae = list_entry(p, struct ae, next);
+                if (strcmp(ae->info.ae_name, name) == 0) {
+                        memcpy(info, &ae->info, sizeof(*info));
+                        pthread_rwlock_unlock(&connmgr.aes_lock);
+                        return 0;
+                }
+        }
+
+        pthread_rwlock_unlock(&connmgr.aes_lock);
+
+        return -1;
+}
+
+static int add_ae_conn(const char *       name,
                        int                fd,
                        qosspec_t          qs,
                        struct conn_info * rcv_info)
 {
-        struct ae_conn * ae_conn = NULL;
+        struct ae_conn *   ae_conn;
+        struct list_head * p;
+        struct ae *        ae = NULL;
 
         ae_conn = malloc(sizeof(*ae_conn));
         if (ae_conn == NULL) {
@@ -76,40 +100,40 @@ static int add_ae_conn(struct ae *        ae,
                 return -1;
         }
 
-        ae_conn->conn.conn_info = *rcv_info;
+        ae_conn->conn.conn_info    = *rcv_info;
         ae_conn->conn.flow_info.fd = fd;
         ae_conn->conn.flow_info.qs = qs;
 
         list_head_init(&ae_conn->next);
 
-        pthread_mutex_lock(&ae->conn_lock);
-        list_add(&ae_conn->next, &ae->conn_list);
-        pthread_cond_signal(&ae->conn_cond);
-        pthread_mutex_unlock(&ae->conn_lock);
-
-        return 0;
-}
-
-static struct ae * find_ae_by_name(char * name)
-{
-        struct list_head * p = NULL;
+        pthread_rwlock_wrlock(&connmgr.aes_lock);
 
         list_for_each(p, &connmgr.aes) {
-                struct ae * ae = list_entry(p, struct ae, next);
+                ae = list_entry(p, struct ae, next);
                 if (strcmp(ae->info.ae_name, name) == 0)
-                        return ae;
+                        break;
         }
 
-        return NULL;
+        /* Check if entry was removed during allocation. */
+        if (ae == NULL || strcmp(ae->info.ae_name, name) != 0) {
+                pthread_rwlock_unlock(&connmgr.aes_lock);
+                return -1;
+        }
+
+        list_add(&ae_conn->next, &ae->conn_list);
+
+        pthread_rwlock_unlock(&connmgr.aes_lock);
+
+        return 0;
 }
 
 static void * flow_acceptor(void * o)
 {
         int               fd;
         qosspec_t         qs;
+        struct conn_info  snd_info;
         struct conn_info  rcv_info;
         struct conn_info  fail_info;
-        struct ae *       ae = NULL;
 
         (void) o;
 
@@ -134,25 +158,23 @@ static void * flow_acceptor(void * o)
                         continue;
                 }
 
-                pthread_mutex_lock(&connmgr.aes_lock);
-                ae = find_ae_by_name(rcv_info.ae_name);
-                pthread_mutex_unlock(&connmgr.aes_lock);
-
-                if (ae != NULL) {
-                        if (cacep_snd(fd, &ae->info)) {
-                                log_err("Failed to respond to req.");
-                                flow_dealloc(fd);
-                                continue;
-                        }
-
-                        if (add_ae_conn(ae, fd, qs, &rcv_info)) {
-                                log_err("Failed to add ae conn.");
-                                flow_dealloc(fd);
-                                continue;
-                        }
-                } else {
+                if (get_info_by_name(rcv_info.ae_name, &snd_info)) {
+                        log_err("Failed to get info for %s.", rcv_info.ae_name);
                         cacep_snd(fd, &fail_info);
                         flow_dealloc(fd);
+                        continue;
+                }
+
+                if (cacep_snd(fd, &snd_info)) {
+                        log_err("Failed to respond to request.");
+                        flow_dealloc(fd);
+                        continue;
+                }
+
+                if (add_ae_conn(rcv_info.ae_name, fd, qs, &rcv_info)) {
+                        log_err("Failed to add new connection.");
+                        flow_dealloc(fd);
+                        continue;
                 }
         }
 
@@ -161,11 +183,11 @@ static void * flow_acceptor(void * o)
 
 int connmgr_init(void)
 {
-        list_head_init(&connmgr.aes);
-
-        if (pthread_mutex_init(&connmgr.aes_lock, NULL))
+        if (pthread_rwlock_init(&connmgr.aes_lock, NULL))
                 return -1;
 
+        list_head_init(&connmgr.aes);
+
         return 0;
 }
 
@@ -180,13 +202,12 @@ int connmgr_start(void)
 void connmgr_stop(void)
 {
         pthread_cancel(connmgr.acceptor);
-        pthread_join(connmgr.acceptor, NULL);
 }
 
 static void destroy_ae(struct ae * ae)
 {
-        struct list_head * p = NULL;
-        struct list_head * h = NULL;
+        struct list_head * p;
+        struct list_head * h;
 
         pthread_mutex_lock(&ae->conn_lock);
 
@@ -206,20 +227,22 @@ static void destroy_ae(struct ae * ae)
 
 void connmgr_fini(void)
 {
-        struct list_head * p = NULL;
-        struct list_head * n = NULL;
+        struct list_head * p;
+        struct list_head * h;
+
+        pthread_join(connmgr.acceptor, NULL);
 
-        pthread_mutex_lock(&connmgr.aes_lock);
+        pthread_rwlock_wrlock(&connmgr.aes_lock);
 
-        list_for_each_safe(p, n, &connmgr.aes) {
+        list_for_each_safe(p, h, &connmgr.aes) {
                 struct ae * e = list_entry(p, struct ae, next);
                 list_del(&e->next);
                 destroy_ae(e);
         }
 
-        pthread_mutex_unlock(&connmgr.aes_lock);
+        pthread_rwlock_unlock(&connmgr.aes_lock);
 
-        pthread_mutex_destroy(&connmgr.aes_lock);
+        pthread_rwlock_destroy(&connmgr.aes_lock);
 }
 
 struct ae * connmgr_ae_create(struct conn_info info)
@@ -228,42 +251,46 @@ struct ae * connmgr_ae_create(struct conn_info info)
 
         ae = malloc(sizeof(*ae));
         if (ae == NULL)
-                return NULL;
+                goto fail_malloc;
 
         list_head_init(&ae->next);
         list_head_init(&ae->conn_list);
 
         ae->info = info;
 
-        if (pthread_mutex_init(&ae->conn_lock, NULL)) {
-                free(ae);
-                return NULL;
-        }
+        if (pthread_mutex_init(&ae->conn_lock, NULL))
+                goto fail_mutex_init;
 
-        if (pthread_cond_init(&ae->conn_cond, NULL)) {
-                pthread_mutex_destroy(&ae->conn_lock);
-                free(ae);
-                return NULL;
-        }
+        if (pthread_cond_init(&ae->conn_cond, NULL))
+                goto fail_cond_init;
+
+        pthread_rwlock_wrlock(&connmgr.aes_lock);
 
-        pthread_mutex_lock(&connmgr.aes_lock);
         list_add(&ae->next, &connmgr.aes);
-        pthread_mutex_unlock(&connmgr.aes_lock);
+
+        pthread_rwlock_unlock(&connmgr.aes_lock);
 
         return ae;
+
+ fail_cond_init:
+        pthread_mutex_destroy(&ae->conn_lock);
+ fail_mutex_init:
+        free(ae);
+ fail_malloc:
+        return NULL;
 }
 
 void connmgr_ae_destroy(struct ae * ae)
 {
         assert(ae);
 
-        pthread_mutex_lock(&connmgr.aes_lock);
+        pthread_rwlock_wrlock(&connmgr.aes_lock);
 
         list_del(&ae->next);
 
-        destroy_ae(ae);
+        pthread_rwlock_unlock(&connmgr.aes_lock);
 
-        pthread_mutex_unlock(&connmgr.aes_lock);
+        destroy_ae(ae);
 }
 
 int connmgr_alloc(struct ae *   ae,
@@ -313,7 +340,7 @@ int connmgr_alloc(struct ae *   ae,
 int connmgr_wait(struct ae *   ae,
                  struct conn * conn)
 {
-        struct ae_conn * ae_conn = NULL;
+        struct ae_conn * ae_conn;
 
         assert(ae);
         assert(conn);
diff --git a/src/ipcpd/normal/enroll.c b/src/ipcpd/normal/enroll.c
index 471230bc..a33239a0 100644
--- a/src/ipcpd/normal/enroll.c
+++ b/src/ipcpd/normal/enroll.c
@@ -324,6 +324,7 @@ int enroll_init(void)
 
 void enroll_fini(void)
 {
+        pthread_join(enroll.listener, NULL);
         cdap_destroy(enroll.cdap);
         connmgr_ae_destroy(enroll.ae);
 }
@@ -339,5 +340,4 @@ int enroll_start(void)
 void enroll_stop(void)
 {
         pthread_cancel(enroll.listener);
-        pthread_join(enroll.listener, NULL);
 }
diff --git a/src/ipcpd/normal/main.c b/src/ipcpd/normal/main.c
index 95b0ed56..53762415 100644
--- a/src/ipcpd/normal/main.c
+++ b/src/ipcpd/normal/main.c
@@ -159,6 +159,7 @@ static int boot_components(void)
         ipcp_set_state(IPCP_OPERATIONAL);
 
         if (connmgr_start()) {
+                ipcp_set_state(IPCP_INIT);
                 log_err("Failed to start AP connection manager.");
                 goto fail_connmgr_start;
         }
@@ -166,7 +167,6 @@ static int boot_components(void)
         return 0;
 
  fail_connmgr_start:
-        ipcp_set_state(IPCP_INIT);
         enroll_stop();
  fail_enroll_start:
         dir_fini();
diff --git a/src/irmd/ipcp.c b/src/irmd/ipcp.c
index b8f11508..e1689b91 100644
--- a/src/irmd/ipcp.c
+++ b/src/irmd/ipcp.c
@@ -52,9 +52,9 @@ ipcp_msg_t * send_recv_ipcp_msg(pid_t        api,
                                 ipcp_msg_t * msg)
 {
        int            sockfd    = 0;
-       buffer_t       buf;
+       uint8_t        buf[IPCP_MSG_BUF_SIZE];
        char *         sock_path = NULL;
-       ssize_t        count     = 0;
+       ssize_t        len;
        ipcp_msg_t *   recv_msg  = NULL;
        struct timeval tv;
 
@@ -73,14 +73,8 @@ ipcp_msg_t * send_recv_ipcp_msg(pid_t        api,
 
        free(sock_path);
 
-       buf.len = ipcp_msg__get_packed_size(msg);
-       if (buf.len == 0) {
-               close(sockfd);
-               return NULL;
-       }
-
-       buf.data = malloc(IPCP_MSG_BUF_SIZE);
-       if (buf.data == NULL) {
+       len = ipcp_msg__get_packed_size(msg);
+       if (len == 0) {
                close(sockfd);
                return NULL;
        }
@@ -113,18 +107,16 @@ ipcp_msg_t * send_recv_ipcp_msg(pid_t        api,
                log_warn("Failed to set timeout on socket.");
 
        pthread_cleanup_push(close_ptr, (void *) &sockfd);
-       pthread_cleanup_push((void (*)(void *)) free, (void *) buf.data);
 
-       ipcp_msg__pack(msg, buf.data);
+       ipcp_msg__pack(msg, buf);
 
-       if (write(sockfd, buf.data, buf.len) != -1)
-               count = read(sockfd, buf.data, IPCP_MSG_BUF_SIZE);
+       if (write(sockfd, buf, len) != -1)
+               len = read(sockfd, buf, IPCP_MSG_BUF_SIZE);
 
-       if (count > 0)
-               recv_msg = ipcp_msg__unpack(NULL, count, buf.data);
+       if (len > 0)
+               recv_msg = ipcp_msg__unpack(NULL, len, buf);
 
        pthread_cleanup_pop(true);
-       pthread_cleanup_pop(true);
 
        return recv_msg;
 }
diff --git a/src/lib/sockets.c b/src/lib/sockets.c
index 263d2356..9f1b326e 100644
--- a/src/lib/sockets.c
+++ b/src/lib/sockets.c
@@ -95,40 +95,32 @@ static void close_ptr(void * o)
 
 irm_msg_t * send_recv_irm_msg(irm_msg_t * msg)
 {
-        int sockfd;
-        buffer_t buf;
-        ssize_t count = 0;
-        irm_msg_t * recv_msg = NULL;
+        int         sockfd;
+        uint8_t     buf[IRM_MSG_BUF_SIZE];
+        ssize_t     len;
+        irm_msg_t * recv_msg;
 
         sockfd = client_socket_open(IRM_SOCK_PATH);
         if (sockfd < 0)
                 return NULL;
 
-        buf.len = irm_msg__get_packed_size(msg);
-        if (buf.len == 0) {
-                close(sockfd);
-                return NULL;
-        }
-
-        buf.data = malloc(IRM_MSG_BUF_SIZE);
-        if (buf.data == NULL) {
+        len = irm_msg__get_packed_size(msg);
+        if (len == 0) {
                 close(sockfd);
                 return NULL;
         }
 
         pthread_cleanup_push(close_ptr, &sockfd);
-        pthread_cleanup_push((void (*)(void *)) free, (void *) buf.data);
 
-        irm_msg__pack(msg, buf.data);
+        irm_msg__pack(msg, buf);
 
-        if (write(sockfd, buf.data, buf.len) != -1)
-                count = read(sockfd, buf.data, IRM_MSG_BUF_SIZE);
+        if (write(sockfd, buf, len) != -1)
+                len = read(sockfd, buf, IRM_MSG_BUF_SIZE);
 
-        if (count > 0)
-                recv_msg = irm_msg__unpack(NULL, count, buf.data);
+        if (len > 0)
+                recv_msg = irm_msg__unpack(NULL, len, buf);
 
         pthread_cleanup_pop(true);
-        pthread_cleanup_pop(true);
 
         return recv_msg;
 }