irmd: introduced locking

This commit adds locking to the IRMd with a single global lock.
It also fixes some issues in cleaning up the daemon.

3 files changed, 287 insertions, 142 deletions

diff --git a/src/irmd/main.c b/src/irmd/main.c
index 8cf03400..2be5e1b6 100644
--- a/src/irmd/main.c
+++ b/src/irmd/main.c
@@ -53,7 +53,7 @@
 #endif
 
 #ifndef IRMD_THREADPOOL_SIZE
-  #define IRMD_THREADPOOL_SIZE 3
+  #define IRMD_THREADPOOL_SIZE 5
 #endif
 
 enum flow_state {
@@ -66,8 +66,6 @@ struct ipcp_entry {
         struct list_head  next;
         instance_name_t * api;
         char *            dif_name;
-
-        pthread_mutex_t   lock;
 };
 
 /* currently supports only registering whatevercast groups of a single AP-I */
@@ -84,9 +82,9 @@ struct reg_name_entry {
         char * req_ap_name;
         char * req_ae_name;
         int    response;
+        bool   flow_arrived;
 
-        pthread_cond_t flow_arrived;
-        pthread_mutex_t fa_lock;
+        pthread_cond_t  acc_signal;
 };
 
 /* keeps track of port_id's between N and N - 1 */
@@ -98,6 +96,8 @@ struct port_map_entry {
         pid_t n_pid;
         pid_t n_1_pid;
 
+        pthread_cond_t  res_signal;
+
         enum flow_state state;
 };
 
@@ -118,9 +118,28 @@ struct irm {
 
         pthread_t * threadpool;
 
-        pthread_mutex_t lock;
+        pthread_mutex_t r_lock;
 } * instance = NULL;
 
+static struct port_map_entry * port_map_entry_create()
+{
+        struct port_map_entry * e = malloc(sizeof(*e));
+        if (e == NULL)
+                return NULL;
+
+        e->n_pid   = 0;
+        e->n_1_pid = 0;
+        e->port_id = 0;
+        e->state   = FLOW_NULL;
+
+        if (pthread_cond_init(&e->res_signal, NULL)) {
+                free(e);
+                return NULL;
+        }
+
+        return e;
+}
+
 static struct port_map_entry * get_port_map_entry(int port_id)
 {
         struct list_head * pos = NULL;
@@ -161,7 +180,6 @@ static struct ipcp_entry * ipcp_entry_create()
         e->dif_name = NULL;
 
         INIT_LIST_HEAD(&e->next);
-        pthread_mutex_init(&e->lock, NULL);
 
         return e;
 }
@@ -251,13 +269,14 @@ static struct reg_name_entry * reg_name_entry_create()
         e->name         = NULL;
         e->api          = NULL;
         e->accept       = false;
-        /* pending response */
-        e->response     = 1;
         e->req_ap_name  = NULL;
         e->req_ae_name  = NULL;
+        e->flow_arrived = false;
 
-        pthread_mutex_init(&e->fa_lock, NULL);
-        pthread_cond_init(&e->flow_arrived, NULL);
+        if (pthread_cond_init(&e->acc_signal, NULL)) {
+                free(e);
+                return NULL;
+        }
 
         INIT_LIST_HEAD(&e->next);
 
@@ -290,8 +309,6 @@ static int reg_name_entry_destroy(struct reg_name_entry * e)
         if (e->req_ae_name != NULL)
                 free(e->req_ae_name);
 
-        pthread_mutex_destroy(&e->fa_lock);
-
         free(e);
 
         return 0;
@@ -394,13 +411,13 @@ static pid_t create_ipcp(char *         ap_name,
 
         tmp->dif_name = NULL;
 
-        pthread_mutex_lock(&instance->lock);
+        pthread_mutex_lock(&instance->r_lock);
 
         list_add(&tmp->next, &instance->ipcps);
 
-        pthread_mutex_unlock(&instance->lock);
+        pthread_mutex_unlock(&instance->r_lock);
 
-        LOG_INFO("Created IPCP %s-%d ", ap_name, pid);
+        LOG_INFO("Created IPCP %s-%d.", ap_name, pid);
 
         return pid;
 }
@@ -409,6 +426,8 @@ static int destroy_ipcp(instance_name_t * api)
 {
         struct list_head * pos = NULL;
         struct list_head * n = NULL;
+        pid_t pid = 0;
+
 
         if (api == NULL)
                 return 0;
@@ -421,6 +440,7 @@ static int destroy_ipcp(instance_name_t * api)
                 return 0;
         }
 
+        pid =api->id;
         if (ipcp_destroy(api->id))
                 LOG_ERR("Could not destroy IPCP.");
 
@@ -434,7 +454,7 @@ static int destroy_ipcp(instance_name_t * api)
                 ipcp_entry_destroy(tmp);
         }
 
-        LOG_INFO("Destroyed IPCP %s-%d.", api->name, api->id);
+        LOG_INFO("Destroyed IPCP %d.", pid);
 
         return 0;
 }
@@ -444,33 +464,41 @@ static int bootstrap_ipcp(instance_name_t *  api,
 {
         struct ipcp_entry * entry = NULL;
 
+        pthread_mutex_lock(&instance->r_lock);
+
         if (api->id == 0)
                 api = get_ipcp_by_name(api->name);
 
         if (api == NULL) {
+                pthread_mutex_unlock(&instance->r_lock);
                 LOG_ERR("No such IPCP in the system.");
                 return -1;
         }
 
         entry = get_ipcp_entry_by_name(api);
         if (entry == NULL) {
+                pthread_mutex_unlock(&instance->r_lock);
                 LOG_ERR("No such IPCP.");
                 return -1;
         }
 
         entry->dif_name = strdup(conf->dif_name);
         if (entry->dif_name == NULL) {
+                pthread_mutex_unlock(&instance->r_lock);
                 LOG_ERR("Failed to strdup.");
                 return -1;
         }
 
         if (ipcp_bootstrap(entry->api->id, conf)) {
+                pthread_mutex_unlock(&instance->r_lock);
                 LOG_ERR("Could not bootstrap IPCP.");
                 free(entry->dif_name);
                 entry->dif_name = NULL;
                 return -1;
         }
 
+        pthread_mutex_unlock(&instance->r_lock);
+
         LOG_INFO("Bootstrapped IPCP %s-%d in DIF %s.",
                  api->name, api->id, conf->dif_name);
 
@@ -485,14 +513,18 @@ static int enroll_ipcp(instance_name_t  * api,
         ssize_t n_1_difs_size = 0;
         struct ipcp_entry * entry = NULL;
 
+        pthread_mutex_lock(&instance->r_lock);
+
         entry = get_ipcp_entry_by_name(api);
         if (entry == NULL) {
+                pthread_mutex_unlock(&instance->r_lock);
                 LOG_ERR("No such IPCP.");
                 return -1;
         }
 
         entry->dif_name = strdup(dif_name);
         if (entry->dif_name == NULL) {
+                pthread_mutex_unlock(&instance->r_lock);
                 LOG_ERR("Failed to strdup.");
                 return -1;
         }
@@ -502,6 +534,7 @@ static int enroll_ipcp(instance_name_t  * api,
                 LOG_ERR("Could not find a member of that DIF.");
                 free(entry->dif_name);
                 entry->dif_name = NULL;
+                pthread_mutex_unlock(&instance->r_lock);
                 return -1;
         }
 
@@ -510,6 +543,7 @@ static int enroll_ipcp(instance_name_t  * api,
                 LOG_ERR("Could not find N-1 DIFs.");
                 free(entry->dif_name);
                 entry->dif_name = NULL;
+                pthread_mutex_unlock(&instance->r_lock);
                 return -1;
         }
 
@@ -517,9 +551,12 @@ static int enroll_ipcp(instance_name_t  * api,
                 LOG_ERR("Could not enroll IPCP.");
                 free(entry->dif_name);
                 entry->dif_name = NULL;
+                pthread_mutex_unlock(&instance->r_lock);
                 return -1;
         }
 
+        pthread_mutex_unlock(&instance->r_lock);
+
         LOG_INFO("Enrolled IPCP %s-%d in DIF %s.",
                  api->name, api->id, dif_name);
 
@@ -551,50 +588,6 @@ static int unreg_ipcp(instance_name_t  * api,
         return 0;
 }
 
-static int ap_unreg_id(pid_t    pid,
-                       char **  difs,
-                       size_t   len)
-{
-        int i;
-        int ret = 0;
-        struct reg_name_entry * rne = NULL;
-        struct list_head      * pos = NULL;
-
-        rne = get_reg_name_entry_by_id(pid);
-        if (rne == NULL)
-                return 0; /* no such id */
-
-        if (instance->ipcps.next == NULL) {
-                LOG_ERR("No IPCPs in this system.");
-                return 0;
-        }
-
-        if (strcmp(difs[0], ALL_DIFS) == 0) {
-                  list_for_each(pos, &instance->ipcps) {
-                        struct ipcp_entry * e =
-                                list_entry(pos, struct ipcp_entry, next);
-
-                        if (ipcp_name_unreg(e->api->id, rne->name)) {
-                                LOG_ERR("Could not unregister %s in DIF %s.",
-                                        rne->name, e->dif_name);
-                                --ret;
-                        }
-                }
-        } else {
-                for (i = 0; i < len; ++i) {
-                        if (ipcp_name_unreg(pid, rne->name)) {
-                                LOG_ERR("Could not unregister %s in DIF %s.",
-                                        rne->name, difs[i]);
-                                --ret;
-                        }
-                }
-        }
-
-        reg_name_entry_del_name(rne->name);
-
-        return ret;
-}
-
 static int ap_reg(char *  ap_name,
                   pid_t   ap_id,
                   char ** difs,
@@ -608,24 +601,35 @@ static int ap_reg(char *  ap_name,
         instance_name_t * api   = NULL;
         instance_name_t * ipcpi = NULL;
 
-        if (instance->ipcps.next == NULL)
-                return -1;
 
-        /* check if this ap_name is already registered */
-        rne = get_reg_name_entry_by_name(ap_name);
-        if (rne != NULL)
-                return -1; /* can only register one instance for now */
+        pthread_mutex_lock(&instance->r_lock);
+
+        if (instance->ipcps.next == NULL) {
+                pthread_mutex_unlock(&instance->r_lock);
+                return -1;
+        }
 
         api = instance_name_create();
         if (api == NULL) {
+                pthread_mutex_unlock(&instance->r_lock);
                 return -1;
         }
 
         if (instance_name_init_from(api, ap_name, ap_id) == NULL) {
+                pthread_mutex_unlock(&instance->r_lock);
                 instance_name_destroy(api);
                 return -1;
         }
 
+        /* check if this ap_name is already registered */
+
+        rne = get_reg_name_entry_by_name(ap_name);
+        if (rne != NULL) {
+                instance_name_destroy(api);
+                pthread_mutex_unlock(&instance->r_lock);
+                return -1; /* can only register one instance for now */
+        }
+
         /*
          * for now, the whatevercast name is the same as the ap_name and
          * contains a single instance only
@@ -662,12 +666,15 @@ static int ap_reg(char *  ap_name,
 
         if (ret ==  0) {
                 instance_name_destroy(api);
+                pthread_mutex_unlock(&instance->r_lock);
                 return -1;
         }
         /* for now, we register single instances */
         ret = reg_name_entry_add_name_instance(strdup(ap_name),
                                                api);
 
+        pthread_mutex_unlock(&instance->r_lock);
+
         return ret;
 }
 
@@ -676,17 +683,57 @@ static int ap_unreg(char *  ap_name,
                     char ** difs,
                     size_t  len)
 {
-        struct reg_name_entry * tmp = NULL;
+        int i;
+        int ret = 0;
+        struct reg_name_entry * rne = NULL;
+        struct list_head      * pos = NULL;
+
+        pthread_mutex_lock(&instance->r_lock);
 
         /* check if ap_name is registered */
-        tmp = get_reg_name_entry_by_id(ap_id);
-        if (tmp == NULL)
+        rne = get_reg_name_entry_by_id(ap_id);
+        if (rne == NULL) {
+                pthread_mutex_unlock(&instance->r_lock);
+                return 0; /* no such id */
+        }
+
+        if (strcmp(ap_name, rne->api->name)) {
+                pthread_mutex_unlock(&instance->r_lock);
                 return 0;
+        }
 
-        if (strcmp(ap_name, tmp->api->name))
+        if (instance->ipcps.next == NULL) {
+                pthread_mutex_unlock(&instance->r_lock);
+                LOG_ERR("No IPCPs in this system.");
                 return 0;
+        }
+
+        if (strcmp(difs[0], ALL_DIFS) == 0) {
+                  list_for_each(pos, &instance->ipcps) {
+                        struct ipcp_entry * e =
+                                list_entry(pos, struct ipcp_entry, next);
+
+                        if (ipcp_name_unreg(e->api->id, rne->name)) {
+                                LOG_ERR("Could not unregister %s in DIF %s.",
+                                        rne->name, e->dif_name);
+                                --ret;
+                        }
+                }
+        } else {
+                for (i = 0; i < len; ++i) {
+                        if (ipcp_name_unreg(ap_id, rne->name)) {
+                                LOG_ERR("Could not unregister %s in DIF %s.",
+                                        rne->name, difs[i]);
+                                --ret;
+                        }
+                }
+        }
 
-        return ap_unreg_id(ap_id, difs, len);
+        reg_name_entry_del_name(rne->name);
+
+        pthread_mutex_unlock(&instance->r_lock);
+
+        return ret;
 }
 
 static struct port_map_entry * flow_accept(pid_t    pid,
@@ -694,40 +741,40 @@ static struct port_map_entry * flow_accept(pid_t    pid,
                                            char **  ae_name)
 {
         struct port_map_entry * pme;
-        struct reg_name_entry * rne = get_reg_name_entry_by_id(pid);
+        struct reg_name_entry * rne = NULL;
+
+        pthread_mutex_lock(&instance->r_lock);
+
+        rne = get_reg_name_entry_by_id(pid);
         if (rne == NULL) {
+                pthread_mutex_unlock(&instance->r_lock);
                 LOG_DBGF("Unregistered AP calling accept().");
                 return NULL;
         }
-
         if (rne->accept) {
+                pthread_mutex_unlock(&instance->r_lock);
                 LOG_DBGF("This AP still has a pending accept().");
                 return NULL;
         }
 
-        rne->accept = true;
+        rne->accept       = true;
+        rne->flow_arrived = false;
 
-        pthread_mutex_lock(&rne->fa_lock);
-
-        pthread_cond_wait(&rne->flow_arrived, &rne->fa_lock);
-
-        if (rne->response == -1) {
-                list_del(&rne->next);
-                reg_name_entry_destroy(rne);
-                return NULL;
-        }
+        while (!rne->flow_arrived)
+                pthread_cond_wait(&rne->acc_signal, &instance->r_lock);
 
         pme = get_port_map_entry_n(pid);
         if (pme == NULL) {
+                pthread_mutex_unlock(&instance->r_lock);
                 LOG_ERR("Port_id was not created yet.");
-                pthread_mutex_unlock(&rne->fa_lock);
                 return NULL;
         }
 
         *ap_name = rne->req_ap_name;
         if (ae_name != NULL)
                 *ae_name = rne->req_ae_name;
-        pthread_mutex_unlock(&rne->fa_lock);
+
+        pthread_mutex_unlock(&instance->r_lock);
 
         return pme;
 }
@@ -737,14 +784,19 @@ static int flow_alloc_resp(pid_t n_pid,
                            int   response)
 {
         struct port_map_entry * pme = NULL;
-        struct reg_name_entry * rne = get_reg_name_entry_by_id(n_pid);
-        if (rne == NULL)
+        struct reg_name_entry * rne = NULL;
+
+        pthread_mutex_lock(&instance->r_lock);
+
+        rne = get_reg_name_entry_by_id(n_pid);
+        if (rne == NULL) {
+                pthread_mutex_unlock(&instance->r_lock);
                 return -1;
+        }
 
-        pthread_mutex_lock(&rne->fa_lock);
         /* FIXME: check all instances associated with the name */
         if (!rne->accept) {
-                pthread_mutex_unlock(&rne->fa_lock);
+                pthread_mutex_unlock(&instance->r_lock);
                 LOG_ERR("No process listening for this name.");
                 return -1;
         }
@@ -754,15 +806,15 @@ static int flow_alloc_resp(pid_t n_pid,
          * once we can handle a list of AP-I's, remove it from the list
          */
 
-        rne->accept   = false;
-        rne->response = response;
+        rne->accept       = false;
+        rne->flow_arrived = false;
 
         if (!response) {
                 pme = get_port_map_entry(port_id);
                 pme->state = FLOW_ALLOCATED;
         }
 
-        pthread_mutex_unlock(&rne->fa_lock);
+        pthread_mutex_unlock(&instance->r_lock);
 
         return ipcp_flow_alloc_resp(pme->n_1_pid,
                                     port_id,
@@ -776,53 +828,96 @@ static struct port_map_entry * flow_alloc(pid_t  pid,
                                           char * src_ae_name,
                                           struct qos_spec * qos)
 {
-        struct port_map_entry * e = malloc(sizeof(*e));
-        if (e == NULL) {
+        struct port_map_entry * pme;
+        instance_name_t * ipcp;
+
+        pme = port_map_entry_create();
+        if (pme == NULL) {
                 LOG_ERR("Failed malloc of port_map_entry.");
                 return NULL;
         }
 
-        e->port_id = bmp_allocate(instance->port_ids);
-        e->n_pid   = pid;
-        e->state   = FLOW_PENDING;
-        e->n_1_pid = get_ipcp_by_dst_name(dst_name)->id;
+        pme->port_id = bmp_allocate(instance->port_ids);
+        pme->n_pid   = pid;
+        pme->state   = FLOW_PENDING;
+        pme->n_1_pid = get_ipcp_by_dst_name(dst_name)->id;
+
+        pthread_mutex_lock(&instance->r_lock);
+
+        list_add(&pme->next, &instance->port_map);
+
+        ipcp = get_ipcp_by_dst_name(dst_name);
 
-        list_add(&e->next, &instance->port_map);
+        pthread_mutex_unlock(&instance->r_lock);
 
-        if (ipcp_flow_alloc(get_ipcp_by_dst_name(dst_name)->id,
-                            e->port_id,
-                            e->n_pid,
+        if (ipcp == NULL) {
+                LOG_DBG("unknown ipcp");
+                return NULL;
+        }
+
+        if (ipcp_flow_alloc(ipcp->id,
+                            pme->port_id,
+                            pme->n_pid,
                             dst_name,
                             src_ap_name,
                             src_ae_name,
                             qos) < 0) {
-                list_del(&e->next);
-                bmp_release(instance->port_ids, e->port_id);
-                free(e);
+                pthread_mutex_lock(&instance->r_lock);
+
+                list_del(&pme->next);
+
+                pthread_mutex_unlock(&instance->r_lock);
+
+                bmp_release(instance->port_ids, pme->port_id);
+                free(pme);
+
                 return NULL;
         }
 
-        return e;
+        return pme;
 }
 
 static int flow_alloc_res(int port_id)
 {
-        bool allocated = false;
         struct port_map_entry * e;
-        struct timespec ts = {0,100000};
 
-        while (!allocated) {
-                /* FIXME: this needs locking */
+        pthread_mutex_lock(&instance->r_lock);
+
+        e = get_port_map_entry(port_id);
+        if (e == NULL) {
+                pthread_mutex_unlock(&instance->r_lock);
+                return -1;
+        }
+
+        if (e->state == FLOW_ALLOCATED) {
+                pthread_mutex_unlock(&instance->r_lock);
+                return 0;
+        }
+
+        while (true) {
+                pthread_cond_wait(&e->res_signal, &instance->r_lock);
                 e = get_port_map_entry(port_id);
                 if (e == NULL) {
-                        LOG_DBGF("Could not locate port_id %d", port_id);
+                        pthread_mutex_unlock(&instance->r_lock);
                         return -1;
                 }
-                if (e->state == FLOW_ALLOCATED)
-                        allocated = true;
-                nanosleep(&ts, NULL);
+                if (e->state == FLOW_ALLOCATED) {
+                        pthread_mutex_unlock(&instance->r_lock);
+                        LOG_DBGF("Returning 0.");
+                        return 0;
+                }
+                if (e->state == FLOW_NULL) {
+                        list_del(&e->next);
+                        pthread_mutex_unlock(&instance->r_lock);
+                        free(e);
+                        return -1;
+
+                }
+                /* still pending, spurious wake */
         }
 
+        pthread_mutex_unlock(&instance->r_lock);
+
         return 0;
 }
 
@@ -830,13 +925,22 @@ static int flow_dealloc(int port_id)
 {
         pid_t    n_1_pid;
 
-        struct port_map_entry * e = get_port_map_entry(port_id);
-        if (e == NULL)
+        struct port_map_entry * e = NULL;
+
+        pthread_mutex_lock(&instance->r_lock);
+
+        e = get_port_map_entry(port_id);
+        if (e == NULL) {
+                pthread_mutex_unlock(&instance->r_lock);
                 return 0;
+        }
 
         n_1_pid = e->n_1_pid;
 
         list_del(&e->next);
+
+        pthread_mutex_unlock(&instance->r_lock);
+
         free(e);
 
         return ipcp_flow_dealloc(n_1_pid, port_id);
@@ -850,12 +954,6 @@ static struct port_map_entry * flow_req_arr(pid_t  pid,
         struct reg_name_entry * rne;
         struct port_map_entry * pme;
 
-        rne = get_reg_name_entry_by_name(dst_name);
-        if (rne == NULL) {
-                LOG_DBGF("Destination name %s unknown.", dst_name);
-                return NULL;
-        }
-
         pme = malloc(sizeof(*pme));
         if (pme == NULL) {
                 LOG_ERR("Failed malloc of port_map_entry.");
@@ -863,19 +961,32 @@ static struct port_map_entry * flow_req_arr(pid_t  pid,
         }
 
         pme->port_id = bmp_allocate(instance->port_ids);
-        pme->n_pid   = rne->api->id;
         pme->state   = FLOW_PENDING;
         pme->n_1_pid = pid;
 
-        list_add(&pme->next, &instance->port_map);
+        pthread_mutex_lock(&instance->r_lock);
 
-        pthread_mutex_lock(&rne->fa_lock);
+        rne = get_reg_name_entry_by_name(dst_name);
+        if (rne == NULL) {
+                pthread_mutex_unlock(&instance->r_lock);
+                LOG_DBGF("Destination name %s unknown.", dst_name);
+                free(pme);
+                return NULL;
+        }
+
+        pme->n_pid   = rne->api->id;
+
+        list_add(&pme->next, &instance->port_map);
 
         rne->req_ap_name = strdup(ap_name);
         rne->req_ae_name = strdup(ae_name);
 
-        pthread_mutex_unlock(&rne->fa_lock);
-        pthread_cond_signal(&rne->flow_arrived);
+        rne->flow_arrived = true;
+
+        pthread_mutex_unlock(&instance->r_lock);
+
+        if (pthread_cond_signal(&rne->acc_signal))
+                LOG_ERR("Failed to send signal.");
 
         return pme;
 }
@@ -885,26 +996,48 @@ static int flow_alloc_reply(int port_id,
 {
         struct port_map_entry * e;
 
-        /* FIXME: do this under lock */
-        if (!response) {
-                e = get_port_map_entry(port_id);
-                if (e == NULL)
-                        return -1;
-                e->state = FLOW_ALLOCATED;
+        pthread_mutex_lock(&instance->r_lock);
+
+        e = get_port_map_entry(port_id);
+        if (e == NULL) {
+                pthread_mutex_unlock(&instance->r_lock);
+                return -1;
+        }
+
+        if (e->state == FLOW_ALLOCATED) {
+                pthread_mutex_unlock(&instance->r_lock);
+                return 0;
         }
 
-        /* FIXME: does this need to be propagated to the IPCP? */
+        if (!response)
+                e->state = FLOW_ALLOCATED;
+
+        else
+                e->state = FLOW_NULL;
+
+        pthread_mutex_unlock(&instance->r_lock);
+        if (pthread_cond_signal(&e->res_signal))
+                LOG_ERR("Failed to send signal.");
 
         return 0;
 }
 
 static int flow_dealloc_ipcp(int port_id)
 {
-        struct port_map_entry * e = get_port_map_entry(port_id);
-        if (e == NULL)
+        struct port_map_entry * e = NULL;
+
+        pthread_mutex_lock(&instance->r_lock);
+
+        e = get_port_map_entry(port_id);
+        if (e == NULL) {
+                pthread_mutex_unlock(&instance->r_lock);
                 return 0;
+        }
 
         list_del(&e->next);
+
+        pthread_mutex_unlock(&instance->r_lock);
+
         free(e);
 
         return 0;
@@ -933,8 +1066,8 @@ static void irm_destroy(struct irm * irm)
                 struct reg_name_entry * e = list_entry(h,
                                                        struct reg_name_entry,
                                                        next);
-                char * difs [1] = {ALL_DIFS};
-                ap_unreg_id(e->api->id, difs, 1);
+                list_del(&e->next);
+                free(e);
         }
 
         list_for_each_safe(h, t, &irm->port_map) {
@@ -1059,6 +1192,7 @@ void * mainloop()
                         e = flow_accept(msg->pid,
                                         &ret_msg.ap_name,
                                         &ret_msg.ae_name);
+
                         if (e == NULL)
                                 break;
 
@@ -1186,7 +1320,15 @@ static struct irm * irm_create()
                 return NULL;
         }
 
-        pthread_mutex_init(&i->lock, NULL);
+        if (pthread_mutex_init(&i->r_lock, NULL)) {
+                irm_destroy(i);
+                return NULL;
+        }
+
+        if (pthread_mutex_init(&i->r_lock, NULL)) {
+                irm_destroy(i);
+                return NULL;
+        }
 
         return i;
 }
diff --git a/src/lib/dev.c b/src/lib/dev.c
index d574363b..c1cfe043 100644
--- a/src/lib/dev.c
+++ b/src/lib/dev.c
@@ -64,6 +64,7 @@ struct ap_data {
 
 int ap_init(char * ap_name)
 {
+        int i = 0;
         _ap_instance = malloc(sizeof(struct ap_data));
         if (_ap_instance == NULL) {
                 return -1;
@@ -106,6 +107,10 @@ int ap_init(char * ap_name)
                 return -1;
         }
 
+        for (i = 0; i < AP_MAX_FLOWS; ++i)
+                _ap_instance->flows[i].rb = NULL;
+
+
         return 0;
 }
 
diff --git a/src/lib/ipcp.c b/src/lib/ipcp.c
index 843582b9..1f1e5c99 100644
--- a/src/lib/ipcp.c
+++ b/src/lib/ipcp.c
@@ -148,8 +148,6 @@ pid_t ipcp_create(char *         ipcp_name,
         strcat(full_name, exec_name);
         full_name[len] = '\0';
 
-        LOG_DBG("Full name is %s", full_name);
-
         char * argv[] = {full_name,
                          irmd_pid,
                          ipcp_name,