Merged in dstaesse/ouroboros/irmd-threads (pull request #70)

irmd: use pthread_cond_wait

3 files changed, 321 insertions, 137 deletions

diff --git a/src/irmd/main.c b/src/irmd/main.c
index 946ed13d..b660511c 100644
--- a/src/irmd/main.c
+++ b/src/irmd/main.c
@@ -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 */
@@ -83,9 +81,11 @@ struct reg_name_entry {
         bool   accept;
         char * req_ap_name;
         char * req_ae_name;
+        int    response;
         bool   flow_arrived;
 
-        pthread_mutex_t fa_lock;
+        pthread_cond_t  acc_signal;
+        pthread_mutex_t acc_lock;
 };
 
 /* keeps track of port_id's between N and N - 1 */
@@ -97,6 +97,9 @@ struct port_map_entry {
         pid_t n_pid;
         pid_t n_1_pid;
 
+        pthread_cond_t  res_signal;
+        pthread_mutex_t res_lock;
+
         enum flow_state state;
 };
 
@@ -117,9 +120,33 @@ 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;
+        }
+
+        if (pthread_mutex_init(&e->res_lock, NULL)) {
+                free(e);
+                return NULL;
+        }
+
+        return e;
+}
+
 static struct port_map_entry * get_port_map_entry(int port_id)
 {
         struct list_head * pos = NULL;
@@ -160,7 +187,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;
 }
@@ -254,7 +280,16 @@ static struct reg_name_entry * reg_name_entry_create()
         e->req_ae_name  = NULL;
         e->flow_arrived = false;
 
-        pthread_mutex_init(&e->fa_lock, NULL);
+        if (pthread_cond_init(&e->acc_signal, NULL)) {
+                free(e);
+                return NULL;
+        }
+
+        if (pthread_mutex_init(&e->acc_lock, NULL)) {
+                free(e);
+                return NULL;
+        }
+
         INIT_LIST_HEAD(&e->next);
 
         return e;
@@ -388,13 +423,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;
 }
@@ -403,6 +438,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;
@@ -415,6 +452,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.");
 
@@ -428,7 +466,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;
 }
@@ -438,33 +476,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);
 
@@ -479,14 +525,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;
         }
@@ -496,6 +546,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;
         }
 
@@ -504,13 +555,18 @@ 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;
         }
 
-        if (ipcp_enroll(entry->api->id, member, n_1_difs[0])) {
+        pthread_mutex_unlock(&instance->r_lock);
+
+        if (ipcp_enroll(api->id, member, n_1_difs[0])) {
                 LOG_ERR("Could not enroll IPCP.");
+                pthread_mutex_lock(&instance->r_lock);
                 free(entry->dif_name);
                 entry->dif_name = NULL;
+                pthread_mutex_unlock(&instance->r_lock);
                 return -1;
         }
 
@@ -545,50 +601,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,
@@ -602,24 +614,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
@@ -656,12 +679,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;
 }
 
@@ -670,62 +696,104 @@ 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;
+                        }
+                }
+        }
+
+        reg_name_entry_del_name(rne->name);
 
-        return ap_unreg_id(ap_id, difs, len);
+        pthread_mutex_unlock(&instance->r_lock);
+
+        return ret;
 }
 
 static struct port_map_entry * flow_accept(pid_t    pid,
                                            char **  ap_name,
                                            char **  ae_name)
 {
-        bool arrived = false;
+        struct port_map_entry * pme;
+        struct reg_name_entry * rne = NULL;
 
-        struct timespec         ts = {0, 100000};
+        pthread_mutex_lock(&instance->r_lock);
 
-        struct port_map_entry * pme;
-        struct reg_name_entry * rne = get_reg_name_entry_by_id(pid);
+        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_unlock(&instance->r_lock);
+        pthread_mutex_lock(&rne->acc_lock);
 
-        /* FIXME: wait for a thread that runs select() on flow_arrived */
-        while (!arrived) {
-                /* FIXME: this needs locking */
-                rne = get_reg_name_entry_by_id(pid);
-                if (rne == NULL)
-                        return NULL;
-                arrived = rne->flow_arrived;
-                nanosleep(&ts, NULL);
-        }
+        while (!rne->flow_arrived)
+                pthread_cond_wait(&rne->acc_signal, &rne->acc_lock);
+
+        pthread_mutex_unlock(&rne->acc_lock);
+        pthread_mutex_lock(&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.");
                 return NULL;
         }
 
-        pthread_mutex_lock(&rne->fa_lock);
         *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;
 }
@@ -734,14 +802,20 @@ static int flow_alloc_resp(pid_t n_pid,
                            int   port_id,
                            int   response)
 {
-        struct reg_name_entry * rne = get_reg_name_entry_by_id(n_pid);
-        struct port_map_entry * pme = get_port_map_entry(port_id);
+        struct port_map_entry * pme = NULL;
+        struct reg_name_entry * rne = NULL;
+
+        pthread_mutex_lock(&instance->r_lock);
 
-        if (rne == NULL || pme == NULL)
+        rne = get_reg_name_entry_by_id(n_pid);
+        if (rne == NULL) {
+                pthread_mutex_unlock(&instance->r_lock);
                 return -1;
+        }
 
         /* FIXME: check all instances associated with the name */
         if (!rne->accept) {
+                pthread_mutex_unlock(&instance->r_lock);
                 LOG_ERR("No process listening for this name.");
                 return -1;
         }
@@ -751,11 +825,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->flow_arrived = false;
         rne->accept       = false;
+        rne->flow_arrived = false;
 
-        if (!response)
+        if (!response) {
+                pme = get_port_map_entry(port_id);
                 pme->state = FLOW_ALLOCATED;
+        }
+
+        pthread_mutex_unlock(&instance->r_lock);
 
         return ipcp_flow_alloc_resp(pme->n_1_pid,
                                     port_id,
@@ -769,53 +847,104 @@ 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->n_pid   = pid;
+        pme->state   = FLOW_PENDING;
+
+        pthread_mutex_lock(&instance->r_lock);
+
+        pme->port_id = bmp_allocate(instance->port_ids);
+        pme->n_1_pid = get_ipcp_by_dst_name(dst_name)->id;
+
+        list_add(&pme->next, &instance->port_map);
+
+        ipcp = get_ipcp_by_dst_name(dst_name);
+
+        pthread_mutex_unlock(&instance->r_lock);
 
-        list_add(&e->next, &instance->port_map);
+        if (ipcp == NULL) {
+                LOG_DBG("unknown ipcp");
+                return NULL;
+        }
 
-        if (ipcp_flow_alloc(get_ipcp_by_dst_name(dst_name)->id,
-                            e->port_id,
-                            e->n_pid,
+        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;
+        }
+
+        pthread_mutex_unlock(&instance->r_lock);
+        pthread_mutex_lock(&e->res_lock);
+
+        while (true) {
+                pthread_cond_wait(&e->res_signal, &e->res_lock);
+
+                pthread_mutex_unlock(&e->res_lock);
+                pthread_mutex_lock(&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;
 }
 
@@ -823,13 +952,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);
@@ -843,33 +981,45 @@ 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.");
                 return NULL;
         }
 
-        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);
+        pme->port_id = bmp_allocate(instance->port_ids);
+
+        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_lock(&rne->acc_lock);
+
         rne->flow_arrived = true;
 
-        pthread_mutex_unlock(&rne->fa_lock);
+        if (pthread_cond_signal(&rne->acc_signal))
+                LOG_ERR("Failed to send signal.");
+
+        pthread_mutex_unlock(&rne->acc_lock);
+
+        pthread_mutex_unlock(&instance->r_lock);
+
 
         return pme;
 }
@@ -879,26 +1029,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;
         }
 
-        /* FIXME: does this need to be propagated to the IPCP? */
+        pthread_mutex_lock(&e->res_lock);
+
+        if (!response)
+                e->state = FLOW_ALLOCATED;
+
+        else
+                e->state = FLOW_NULL;
+
+        if (pthread_cond_signal(&e->res_signal))
+                LOG_ERR("Failed to send signal.");
+
+        pthread_mutex_unlock(&e->res_lock);
+
+        pthread_mutex_unlock(&instance->r_lock);
 
         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;
@@ -927,8 +1099,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) {
@@ -1053,6 +1225,7 @@ void * mainloop()
                         e = flow_accept(msg->pid,
                                         &ret_msg.ap_name,
                                         &ret_msg.ae_name);
+
                         if (e == NULL)
                                 break;
 
@@ -1180,7 +1353,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,