Merged in sandervrijders/ouroboros/be-enrolment (pull request #182)

Be enrolment

13 files changed, 453 insertions, 94 deletions

diff --git a/src/ipcpd/ipcp.h b/src/ipcpd/ipcp.h
index bbf1d1f7..630f7922 100644
--- a/src/ipcpd/ipcp.h
+++ b/src/ipcpd/ipcp.h
@@ -33,6 +33,7 @@
 
 enum ipcp_state {
         IPCP_INIT = 0,
+        IPCP_PENDING_ENROLL,
         IPCP_ENROLLED,
         IPCP_DISCONNECTED,
         IPCP_SHUTDOWN
diff --git a/src/ipcpd/local/main.c b/src/ipcpd/local/main.c
index fcf5f879..4b9dcbbc 100644
--- a/src/ipcpd/local/main.c
+++ b/src/ipcpd/local/main.c
@@ -30,7 +30,6 @@
 #include <ouroboros/utils.h>
 #include <ouroboros/ipcp.h>
 #include <ouroboros/irm_config.h>
-#include <ouroboros/sockets.h>
 #include <ouroboros/bitmap.h>
 #include <ouroboros/shared.h>
 #include <ouroboros/dev.h>
diff --git a/src/ipcpd/normal/dt_const.h b/src/ipcpd/normal/dt_const.h
index 65cde5f7..b33f16c5 100644
--- a/src/ipcpd/normal/dt_const.h
+++ b/src/ipcpd/normal/dt_const.h
@@ -31,9 +31,10 @@ struct dt_const {
         uint8_t cep_id_size;
         uint8_t pdu_length_size;
         uint8_t seqno_size;
-        uint8_t qos_id_size;
         uint8_t ttl_size;
         uint8_t chk_size;
+        uint32_t min_pdu_size;
+        uint32_t max_pdu_size;
 };
 
 #endif /* IPCP_DT_CONST_H */
diff --git a/src/ipcpd/normal/fmgr.c b/src/ipcpd/normal/fmgr.c
index 9521805f..a539b289 100644
--- a/src/ipcpd/normal/fmgr.c
+++ b/src/ipcpd/normal/fmgr.c
@@ -35,6 +35,9 @@
 #include "fmgr.h"
 #include "ribmgr.h"
 #include "frct.h"
+#include "ipcp.h"
+
+extern struct ipcp * _ipcp;
 
 struct n_1_flow {
         int fd;
@@ -48,7 +51,7 @@ struct fmgr {
         struct list_head n_1_flows;
         pthread_mutex_t n_1_flows_lock;
 
-} * instance = NULL;
+} * fmgr = NULL;
 
 static int add_n_1_fd(int fd,
                       char * ae_name)
@@ -65,9 +68,9 @@ static int add_n_1_fd(int fd,
         tmp->fd = fd;
         tmp->ae_name = ae_name;
 
-        pthread_mutex_lock(&instance->n_1_flows_lock);
-        list_add(&tmp->next, &instance->n_1_flows);
-        pthread_mutex_unlock(&instance->n_1_flows_lock);
+        pthread_mutex_lock(&fmgr->n_1_flows_lock);
+        list_add(&tmp->next, &fmgr->n_1_flows);
+        pthread_mutex_unlock(&fmgr->n_1_flows_lock);
 
         return 0;
 }
@@ -77,10 +80,21 @@ static void * fmgr_listen(void * o)
         int fd;
         char * ae_name;
 
-        while (true) {
+        /* FIXME: Avoid busy wait and react to pthread_cond_t */
+        pthread_rwlock_rdlock(&_ipcp->state_lock);
+        while (_ipcp->state != IPCP_ENROLLED ||
+               _ipcp->state != IPCP_SHUTDOWN) {
+                pthread_rwlock_unlock(&_ipcp->state_lock);
+                sched_yield();
+                pthread_rwlock_rdlock(&_ipcp->state_lock);
+        }
+
+        while (_ipcp->state != IPCP_SHUTDOWN) {
+                pthread_rwlock_unlock(&_ipcp->state_lock);
                 fd = flow_accept(&ae_name);
                 if (fd < 0) {
                         LOG_ERR("Failed to accept flow.");
+                        pthread_rwlock_rdlock(&_ipcp->state_lock);
                         continue;
                 }
 
@@ -89,37 +103,46 @@ static void * fmgr_listen(void * o)
                         if (flow_alloc_resp(fd, -1))
                                 LOG_ERR("Failed to reply to flow allocation.");
                         flow_dealloc(fd);
+                        pthread_rwlock_rdlock(&_ipcp->state_lock);
                         continue;
                 }
 
                 if (flow_alloc_resp(fd, 0)) {
                         LOG_ERR("Failed to reply to flow allocation.");
                         flow_dealloc(fd);
+                        pthread_rwlock_rdlock(&_ipcp->state_lock);
                         continue;
                 }
 
                 LOG_DBG("Accepted new flow allocation request for AE %s.",
                         ae_name);
 
-                if (strcmp(ae_name, MGMT_AE) == 0 &&
-                    ribmgr_mgmt_flow(fd)) {
-                        LOG_ERR("Failed to hand file descriptor to RIB.");
-                        flow_dealloc(fd);
-                        continue;
+                if (strcmp(ae_name, MGMT_AE) == 0) {
+                        if (ribmgr_add_flow(fd)) {
+                                LOG_ERR("Failed to hand fd to RIB.");
+                                flow_dealloc(fd);
+                                pthread_rwlock_rdlock(&_ipcp->state_lock);
+                                continue;
+                        }
                 }
 
-                if (strcmp(ae_name, DT_AE) == 0 &&
-                    frct_dt_flow(fd)) {
-                        LOG_ERR("Failed to hand file descriptor to FRCT.");
-                        flow_dealloc(fd);
-                        continue;
+                if (strcmp(ae_name, DT_AE) == 0) {
+                        if (frct_dt_flow(fd)) {
+                                LOG_ERR("Failed to hand fd to FRCT.");
+                                flow_dealloc(fd);
+                                pthread_rwlock_rdlock(&_ipcp->state_lock);
+                                continue;
+                        }
                 }
 
                 if (add_n_1_fd(fd, ae_name)) {
                         LOG_ERR("Failed to add file descriptor to list.");
                         flow_dealloc(fd);
+                        pthread_rwlock_rdlock(&_ipcp->state_lock);
                         continue;
                 }
+
+                pthread_rwlock_rdlock(&_ipcp->state_lock);
         }
 
         return (void *) 0;
@@ -127,16 +150,15 @@ static void * fmgr_listen(void * o)
 
 int fmgr_init()
 {
-        instance = malloc(sizeof(*instance));
-        if (instance == NULL) {
+        fmgr = malloc(sizeof(*fmgr));
+        if (fmgr == NULL)
                 return -1;
-        }
 
-        INIT_LIST_HEAD(&instance->n_1_flows);
+        INIT_LIST_HEAD(&fmgr->n_1_flows);
 
-        pthread_mutex_init(&instance->n_1_flows_lock, NULL);
+        pthread_mutex_init(&fmgr->n_1_flows_lock, NULL);
 
-        pthread_create(&instance->listen_thread,
+        pthread_create(&fmgr->listen_thread,
                        NULL,
                        fmgr_listen,
                        NULL);
@@ -148,20 +170,21 @@ int fmgr_fini()
 {
         struct list_head * pos = NULL;
 
-        pthread_cancel(instance->listen_thread);
+        pthread_cancel(fmgr->listen_thread);
 
-        pthread_join(instance->listen_thread,
+        pthread_join(fmgr->listen_thread,
                      NULL);
 
-        list_for_each(pos, &instance->n_1_flows) {
+        list_for_each(pos, &fmgr->n_1_flows) {
                 struct n_1_flow * e =
                         list_entry(pos, struct n_1_flow, next);
                 if (e->ae_name != NULL)
                         free(e->ae_name);
-                flow_dealloc(e->fd);
+                if (ribmgr_remove_flow(e->fd))
+                    LOG_ERR("Failed to remove management flow.");
         }
 
-        free(instance);
+        free(fmgr);
 
         return 0;
 }
@@ -185,7 +208,7 @@ int fmgr_mgmt_flow(char * dst_name)
                 return -1;
         }
 
-        if (ribmgr_mgmt_flow(fd)) {
+        if (ribmgr_add_flow(fd)) {
                 LOG_ERR("Failed to hand file descriptor to RIB manager");
                 flow_dealloc(fd);
                 return -1;
@@ -233,10 +256,3 @@ int fmgr_flow_dealloc(int port_id)
 
         return -1;
 }
-
-int fmgr_flow_msg()
-{
-        LOG_MISSING;
-
-        return -1;
-}
diff --git a/src/ipcpd/normal/fmgr.h b/src/ipcpd/normal/fmgr.h
index 867cbff6..dc88bbdf 100644
--- a/src/ipcpd/normal/fmgr.h
+++ b/src/ipcpd/normal/fmgr.h
@@ -52,7 +52,4 @@ int fmgr_flow_alloc_resp(pid_t n_api,
 
 int fmgr_flow_dealloc(int port_id);
 
-/* RIB Manager calls this (param will be of type fmgr_msg_t) */
-int fmgr_flow_msg();
-
 #endif
diff --git a/src/ipcpd/normal/frct.c b/src/ipcpd/normal/frct.c
index 22f8a9fc..ba465540 100644
--- a/src/ipcpd/normal/frct.c
+++ b/src/ipcpd/normal/frct.c
@@ -22,24 +22,36 @@
 
 #define OUROBOROS_PREFIX "flow-rtx-control"
 
+#include <stdlib.h>
+
 #include <ouroboros/logs.h>
 
 #include "frct.h"
 
 struct frct_i {
-
 };
 
-int frct_init(struct dt_const * dt_const)
+struct frct {
+        struct dt_const * dtc;
+} * frct = NULL;
+
+int frct_init(struct dt_const * dtc)
 {
-        LOG_MISSING;
+        if (dtc == NULL)
+                return -1;
+
+        frct = malloc(sizeof(*frct));
+        if (frct == NULL)
+                return -1;
+
+        frct->dtc = dtc;
 
         return 0;
 }
 
 int frct_fini()
 {
-        LOG_MISSING;
+        free(frct);
 
         return 0;
 }
diff --git a/src/ipcpd/normal/frct.h b/src/ipcpd/normal/frct.h
index 07fd2c65..515bed3f 100644
--- a/src/ipcpd/normal/frct.h
+++ b/src/ipcpd/normal/frct.h
@@ -29,7 +29,7 @@
 
 struct frct_i;
 
-int             frct_init(struct dt_const * dt_const);
+int             frct_init(struct dt_const * dtc);
 int             frct_fini();
 
 struct frct_i * frct_i_create(int port_id,
diff --git a/src/ipcpd/normal/main.c b/src/ipcpd/normal/main.c
index 2d97f435..57fb72df 100644
--- a/src/ipcpd/normal/main.c
+++ b/src/ipcpd/normal/main.c
@@ -1,3 +1,25 @@
+/*
+ * Ouroboros - Copyright (C) 2016
+ *
+ * Normal IPC Process
+ *
+ *    Sander Vrijders <sander.vrijders@intec.ugent.be>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; either version 2 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, write to the Free Software
+ * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ */
+
 #define OUROBOROS_PREFIX "normal-ipcp"
 
 #include <ouroboros/config.h>
@@ -58,14 +80,14 @@ void ipcp_sig_handler(int sig, siginfo_t * info, void * c)
 
                         pthread_cancel(normal_data(_ipcp)->mainloop);
 
+                        if (fmgr_fini())
+                                LOG_ERR("Failed to finalize flow manager.");
+
                         if (ribmgr_fini())
                                 LOG_ERR("Failed to finalize RIB manager.");
 
                         if (frct_fini())
                                 LOG_ERR("Failed to finalize FRCT.");
-
-                        if (fmgr_fini())
-                                LOG_ERR("Failed to finalize flow manager.");
                 }
         default:
                 return;
@@ -108,16 +130,48 @@ static int normal_ipcp_name_unreg(char * name)
 
 static int normal_ipcp_enroll(char * dif_name)
 {
-        LOG_MISSING;
+        pthread_rwlock_rdlock(&_ipcp->state_lock);
 
-        return -1;
+        if (_ipcp->state != IPCP_INIT) {
+                pthread_rwlock_unlock(&_ipcp->state_lock);
+                LOG_DBGF("Won't enroll an IPCP that is not in INIT.");
+                return -1; /* -ENOTINIT */
+        }
+
+        pthread_rwlock_unlock(&_ipcp->state_lock);
+
+        if (fmgr_mgmt_flow(dif_name)) {
+                pthread_rwlock_unlock(&_ipcp->state_lock);
+                LOG_ERR("Failed to establish management flow.");
+                return -1;
+        }
+
+        /* FIXME: Wait until state changed to ENROLLED */
+
+        return 0;
 }
 
 static int normal_ipcp_bootstrap(struct dif_config * conf)
 {
-        LOG_MISSING;
+        pthread_rwlock_rdlock(&_ipcp->state_lock);
 
-        return -1;
+        if (_ipcp->state != IPCP_INIT) {
+                pthread_rwlock_unlock(&_ipcp->state_lock);
+                LOG_DBGF("Won't bootstrap an IPCP that is not in INIT.");
+                return -1; /* -ENOTINIT */
+        }
+
+        if (ribmgr_bootstrap(conf)) {
+                pthread_rwlock_unlock(&_ipcp->state_lock);
+                LOG_ERR("Failed to bootstrap RIB manager.");
+                return -1;
+        }
+
+        _ipcp->state = IPCP_ENROLLED;
+
+        pthread_rwlock_unlock(&_ipcp->state_lock);
+
+        return 0;
 }
 
 static struct ipcp_ops normal_ops = {
diff --git a/src/ipcpd/normal/ribmgr.c b/src/ipcpd/normal/ribmgr.c
index 98ed38c6..8bb320c0 100644
--- a/src/ipcpd/normal/ribmgr.c
+++ b/src/ipcpd/normal/ribmgr.c
@@ -22,41 +22,324 @@
 
 #define OUROBOROS_PREFIX "rib-manager"
 
+#include <ouroboros/config.h>
 #include <ouroboros/logs.h>
+#include <ouroboros/cdap.h>
+#include <ouroboros/list.h>
+
+#include <stdlib.h>
+#include <pthread.h>
+#include <string.h>
 
 #include "ribmgr.h"
+#include "dt_const.h"
+#include "frct.h"
+#include "ipcp.h"
+
+#define ENROLLMENT "enrollment"
+
+extern struct ipcp * _ipcp;
+
+enum cdap_opcode {
+        READ = 0,
+        WRITE,
+        START,
+        STOP,
+        CREATE,
+        DELETE
+};
+
+struct cdap_request {
+        enum cdap_opcode code;
+        char *           name;
+        int              invoke_id;
+        struct cdap *    instance;
+
+        struct list_head next;
+};
+
+struct mgmt_flow {
+        struct cdap *    instance;
+        int              fd;
+        struct list_head next;
+};
+
+struct rib {
+        struct dt_const  dtc;
+
+        struct list_head flows;
+        pthread_rwlock_t flows_lock;
+
+        struct list_head cdap_reqs;
+        pthread_mutex_t  cdap_reqs_lock;
+} * rib = NULL;
+
+/* Call while holding cdap_reqs_lock */
+int cdap_request_add(struct cdap * instance,
+                     enum cdap_opcode code,
+                     char * name,
+                     int invoke_id)
+{
+        struct cdap_request * req;
+
+        req = malloc(sizeof(*req));
+        if (req == NULL)
+                return -1;
+
+        req->code = code;
+        req->invoke_id = invoke_id;
+        req->instance = instance;
+
+        req->name = strdup(name);
+        if (req->name == NULL) {
+                free(req);
+                return -1;
+        }
+
+        INIT_LIST_HEAD(&req->next);
+
+        list_add(&req->next, &rib->cdap_reqs);
+
+        return 0;
+}
 
 int ribmgr_init()
 {
-        LOG_MISSING;
+        rib = malloc(sizeof(*rib));
+        if (rib == NULL)
+                return -1;
+
+        INIT_LIST_HEAD(&rib->flows);
+        INIT_LIST_HEAD(&rib->cdap_reqs);
+
+        if (pthread_rwlock_init(&rib->flows_lock, NULL)) {
+                LOG_ERR("Failed to initialize rwlock.");
+                free(rib);
+                return -1;
+        }
+
+        if (pthread_mutex_init(&rib->cdap_reqs_lock, NULL)) {
+                LOG_ERR("Failed to initialize mutex.");
+                free(rib);
+                return -1;
+        }
 
         return 0;
 }
 
 int ribmgr_fini()
 {
-        LOG_MISSING;
+        struct list_head * pos = NULL;
+        struct list_head * n = NULL;
+
+        pthread_mutex_lock(&rib->cdap_reqs_lock);
+        list_for_each_safe(pos, n, &rib->cdap_reqs) {
+                struct cdap_request * req =
+                        list_entry(pos, struct cdap_request, next);
+
+                free(req->name);
+                list_del(&req->next);
+                free(req);
+        }
+        pthread_mutex_unlock(&rib->cdap_reqs_lock);
+
+        pthread_rwlock_wrlock(&rib->flows_lock);
+        list_for_each_safe(pos, n, &rib->flows) {
+                struct mgmt_flow * flow =
+                        list_entry(pos, struct mgmt_flow, next);
+                if (cdap_destroy(flow->instance))
+                        LOG_ERR("Failed to destroy CDAP instance.");
+                list_del(&flow->next);
+                free(flow);
+        }
+        pthread_rwlock_unlock(&rib->flows_lock);
+
+        free(rib);
 
         return 0;
 }
 
-int ribmgr_mgmt_flow(int fd)
+int ribmgr_cdap_reply(struct cdap * instance,
+                      int           invoke_id,
+                      int           result,
+                      buffer_t *    val,
+                      size_t        len)
 {
         LOG_MISSING;
 
+        /* FIXME: Check all cdap_reqs here to see if we expect a reply */
+
         return -1;
 }
 
-int ribmgr_bootstrap(struct dif_config * conf)
+int ribmgr_cdap_read(struct cdap * instance,
+                     char *        name)
+{
+        LOG_MISSING;
+
+        return -1;
+}
+
+int ribmgr_cdap_write(struct cdap * instance,
+                      char *        name,
+                      buffer_t *    val,
+                      size_t        len,
+                      uint32_t      flags)
+{
+        LOG_MISSING;
+
+        return -1;
+}
+
+int ribmgr_cdap_create(struct cdap * instance,
+                       char *        name,
+                       buffer_t      val)
 {
         LOG_MISSING;
 
         return -1;
 }
 
-int ribmgr_fmgr_msg()
+int ribmgr_cdap_delete(struct cdap * instance,
+                       char *        name,
+                       buffer_t      val)
 {
         LOG_MISSING;
 
         return -1;
 }
+
+int ribmgr_cdap_start(struct cdap * instance,
+                      char *        name)
+{
+        LOG_MISSING;
+
+        /* FIXME: Handle enrollment request here */
+
+        return -1;
+}
+
+int ribmgr_cdap_stop(struct cdap * instance,
+                     char *        name)
+{
+        LOG_MISSING;
+
+        return -1;
+}
+
+static struct cdap_ops ribmgr_ops = {
+        .cdap_reply  = ribmgr_cdap_reply,
+        .cdap_read   = ribmgr_cdap_read,
+        .cdap_write  = ribmgr_cdap_write,
+        .cdap_create = ribmgr_cdap_create,
+        .cdap_delete = ribmgr_cdap_delete,
+        .cdap_start  = ribmgr_cdap_start,
+        .cdap_stop   = ribmgr_cdap_stop
+};
+
+int ribmgr_add_flow(int fd)
+{
+        struct cdap * instance = NULL;
+        struct mgmt_flow * flow;
+        int iid = 0;
+
+        flow = malloc(sizeof(*flow));
+        if (flow == NULL)
+                return -1;
+
+        instance = cdap_create(&ribmgr_ops, fd);
+        if (instance == NULL) {
+                LOG_ERR("Failed to create CDAP instance");
+                free(flow);
+                return -1;
+        }
+
+        INIT_LIST_HEAD(&flow->next);
+        flow->instance = instance;
+        flow->fd = fd;
+
+        pthread_rwlock_rdlock(&_ipcp->state_lock);
+        pthread_rwlock_wrlock(&rib->flows_lock);
+        if (list_empty(&rib->flows) &&
+            (_ipcp->state == IPCP_INIT ||
+             _ipcp->state == IPCP_DISCONNECTED)) {
+                _ipcp->state = IPCP_PENDING_ENROLL;
+                pthread_rwlock_unlock(&_ipcp->state_lock);
+
+                pthread_mutex_lock(&rib->cdap_reqs_lock);
+                iid = cdap_send_start(instance,
+                                      ENROLLMENT);
+                if (iid < 0) {
+                        pthread_mutex_unlock(&rib->cdap_reqs_lock);
+                        pthread_rwlock_unlock(&rib->flows_lock);
+                        LOG_ERR("Failed to start enrollment.");
+                        cdap_destroy(instance);
+                        free(flow);
+                        return -1;
+                }
+
+                if (cdap_request_add(instance, START, ENROLLMENT, iid)) {
+                        pthread_mutex_unlock(&rib->cdap_reqs_lock);
+                        pthread_rwlock_unlock(&rib->flows_lock);
+                        LOG_ERR("Failed to add CDAP request to list.");
+                        cdap_destroy(instance);
+                        free(flow);
+                        return -1;
+                }
+                pthread_mutex_unlock(&rib->cdap_reqs_lock);
+        }
+        pthread_rwlock_unlock(&_ipcp->state_lock);
+
+        list_add(&flow->next, &rib->flows);
+        pthread_rwlock_unlock(&rib->flows_lock);
+
+        return 0;
+}
+
+int ribmgr_remove_flow(int fd)
+{
+        struct list_head * pos, * n = NULL;
+
+        pthread_rwlock_wrlock(&rib->flows_lock);
+        list_for_each_safe(pos, n, &rib->flows) {
+                struct mgmt_flow * flow =
+                        list_entry(pos, struct mgmt_flow, next);
+                if (flow->fd == fd) {
+                        if (cdap_destroy(flow->instance))
+                                LOG_ERR("Failed to destroy CDAP instance.");
+                        list_del(&flow->next);
+                        free(flow);
+                        return 0;
+                }
+        }
+        pthread_rwlock_unlock(&rib->flows_lock);
+
+        return -1;
+}
+
+int ribmgr_bootstrap(struct dif_config * conf)
+{
+        if (conf == NULL ||
+            conf->type != IPCP_NORMAL) {
+                LOG_ERR("Bad DIF configuration.");
+                return -1;
+        }
+
+        rib->dtc.addr_size = conf->addr_size;
+        rib->dtc.cep_id_size  = conf->cep_id_size;
+        rib->dtc.pdu_length_size = conf->pdu_length_size;
+        rib->dtc.seqno_size = conf->seqno_size;
+        rib->dtc.ttl_size = conf->ttl_size;
+        rib->dtc.chk_size = conf->chk_size;
+        rib->dtc.min_pdu_size = conf->min_pdu_size;
+        rib->dtc.max_pdu_size = conf->max_pdu_size;
+
+        if (frct_init(&rib->dtc)) {
+                LOG_ERR("Failed to initialize FRCT.");
+                return -1;
+        }
+
+        LOG_DBG("Bootstrapped RIB Manager.");
+
+        return 0;
+}
diff --git a/src/ipcpd/normal/ribmgr.h b/src/ipcpd/normal/ribmgr.h
index 335189f9..e85c65be 100644
--- a/src/ipcpd/normal/ribmgr.h
+++ b/src/ipcpd/normal/ribmgr.h
@@ -28,10 +28,9 @@
 int ribmgr_init();
 int ribmgr_fini();
 
-int ribmgr_mgmt_flow(int fd);
-int ribmgr_bootstrap(struct dif_config * conf);
+int ribmgr_add_flow(int fd);
+int ribmgr_remove_flow(int fd);
 
-/* Called by Flow Manager (param of type fmgr_msg_t) */
-int ribmgr_fmgr_msg();
+int ribmgr_bootstrap(struct dif_config * conf);
 
 #endif
diff --git a/src/lib/cdap.c b/src/lib/cdap.c
index 4275bfc7..4599fd8b 100644
--- a/src/lib/cdap.c
+++ b/src/lib/cdap.c
@@ -69,6 +69,28 @@ static ssize_t cdap_msg_to_buffer(cdap_t * msg,
         return len;
 }
 
+static int next_invoke_id(struct cdap * instance)
+{
+        int ret;
+
+        pthread_mutex_lock(&instance->ids_lock);
+        ret = bmp_allocate(instance->ids);
+        pthread_mutex_unlock(&instance->ids_lock);
+
+        return ret;
+}
+
+static int release_invoke_id(struct cdap * instance,
+                             int id)
+{
+        int ret;
+
+        pthread_mutex_lock(&instance->ids_lock);
+        ret = bmp_release(instance->ids, id);
+        pthread_mutex_unlock(&instance->ids_lock);
+
+        return ret;
+}
 
 static void * sdu_reader(void * o)
 {
@@ -146,6 +168,7 @@ static void * sdu_reader(void * o)
                                                           msg->result,
                                                           val,
                                                           length);
+                                release_invoke_id(instance, msg->invoke_id);
                                 free(val);
                         }
                         break;
@@ -189,6 +212,7 @@ struct cdap * cdap_create(struct cdap_ops * ops,
         }
 
         instance->ops = ops;
+        instance->fd = fd;
 
         instance->ids = bmp_create(IDS_SIZE, 0);
         if (instance->ids == NULL) {
@@ -211,6 +235,9 @@ int cdap_destroy(struct cdap * instance)
 
         pthread_cancel(instance->reader);
 
+        if (flow_dealloc(instance->fd))
+                return -1;
+
         pthread_mutex_lock(&instance->ids_lock);
 
         bmp_destroy(instance->ids);
@@ -225,29 +252,6 @@ int cdap_destroy(struct cdap * instance)
         return 0;
 }
 
-static int next_invoke_id(struct cdap * instance)
-{
-        int ret;
-
-        pthread_mutex_lock(&instance->ids_lock);
-        ret = bmp_allocate(instance->ids);
-        pthread_mutex_unlock(&instance->ids_lock);
-
-        return ret;
-}
-
-static int release_invoke_id(struct cdap * instance,
-                             int id)
-{
-        int ret;
-
-        pthread_mutex_lock(&instance->ids_lock);
-        ret = bmp_release(instance->ids, id);
-        pthread_mutex_unlock(&instance->ids_lock);
-
-        return ret;
-}
-
 static int write_msg(struct cdap * instance,
                      cdap_t * msg)
 {
diff --git a/src/lib/shm_ap_rbuff.c b/src/lib/shm_ap_rbuff.c
index 605e5678..84f7617a 100644
--- a/src/lib/shm_ap_rbuff.c
+++ b/src/lib/shm_ap_rbuff.c
@@ -296,7 +296,7 @@ struct rb_entry * shm_ap_rbuff_read(struct shm_ap_rbuff * rb)
         while (tail_el_ptr->port_id < 0)
                 *rb->ptr_tail = (*rb->ptr_tail + 1) & (SHM_RBUFF_SIZE -1);
 
-        while(shm_rbuff_empty(rb))
+        while (shm_rbuff_empty(rb))
                 if (pthread_cond_wait(rb->work, rb->shm_mutex)
                     == EOWNERDEAD) {
                 LOG_DBGF("Recovering dead mutex.");
diff --git a/src/tools/irm/irm_ipcp_bootstrap.c b/src/tools/irm/irm_ipcp_bootstrap.c
index 94b7f81e..d5fa97da 100644
--- a/src/tools/irm/irm_ipcp_bootstrap.c
+++ b/src/tools/irm/irm_ipcp_bootstrap.c
@@ -41,7 +41,6 @@
 #define DEFAULT_ADDR_SIZE 4
 #define DEFAULT_CEP_ID_SIZE 2
 #define DEFAULT_PDU_LEN_SIZE 2
-#define DEFAULT_QOS_ID_SIZE 1
 #define DEFAULT_SEQ_NO_SIZE 4
 #define DEFAULT_TTL_SIZE 1
 #define DEFAULT_CHK_SIZE 2
@@ -62,7 +61,6 @@ static void usage()
                "                [addr <address size> (default: %d)]\n"
                "                [cep_id <CEP-id size> (default: %d)]\n"
                "                [pdu_len <PDU length size> (default: %d)]\n"
-               "                [qos_id <QoS-id size> (default: %d)]\n"
                "                [seqno <sequence number size> (default: %d)]\n"
                "                [ttl <time to live size>  (default: %d)]\n"
                "                [chk <checksum size>  (default: %d)]\n"
@@ -75,10 +73,9 @@ static void usage()
                "if TYPE == " SHIM_ETH_LLC "\n"
                "                if_name <interface name>\n",
                DEFAULT_ADDR_SIZE, DEFAULT_CEP_ID_SIZE,
-               DEFAULT_PDU_LEN_SIZE, DEFAULT_QOS_ID_SIZE,
-               DEFAULT_SEQ_NO_SIZE, DEFAULT_TTL_SIZE,
-               DEFAULT_CHK_SIZE, DEFAULT_MIN_PDU_SIZE,
-               DEFAULT_MAX_PDU_SIZE, DEFAULT_DDNS);
+               DEFAULT_PDU_LEN_SIZE, DEFAULT_SEQ_NO_SIZE,
+               DEFAULT_TTL_SIZE, DEFAULT_CHK_SIZE,
+               DEFAULT_MIN_PDU_SIZE, DEFAULT_MAX_PDU_SIZE, DEFAULT_DDNS);
 }
 
 int do_bootstrap_ipcp(int argc, char ** argv)
@@ -88,7 +85,6 @@ int do_bootstrap_ipcp(int argc, char ** argv)
         uint8_t addr_size = DEFAULT_ADDR_SIZE;
         uint8_t cep_id_size = DEFAULT_CEP_ID_SIZE;
         uint8_t pdu_length_size = DEFAULT_PDU_LEN_SIZE;
-        uint8_t qos_id_size = DEFAULT_QOS_ID_SIZE;
         uint8_t seqno_size = DEFAULT_SEQ_NO_SIZE;
         uint8_t ttl_size = DEFAULT_TTL_SIZE;
         uint8_t chk_size = DEFAULT_CHK_SIZE;
@@ -128,8 +124,6 @@ int do_bootstrap_ipcp(int argc, char ** argv)
                         cep_id_size = atoi(*(argv + 1));
                 } else if (matches(*argv, "pdu_len") == 0) {
                         pdu_length_size = atoi(*(argv + 1));
-                } else if (matches(*argv, "qos_id") == 0) {
-                        qos_id_size = atoi(*(argv + 1));
                 } else if (matches(*argv, "seqno") == 0) {
                         seqno_size = atoi(*(argv + 1));
                 } else if (matches(*argv, "ttl") == 0) {
@@ -162,7 +156,6 @@ int do_bootstrap_ipcp(int argc, char ** argv)
                 conf.addr_size = addr_size;
                 conf.cep_id_size = cep_id_size;
                 conf.pdu_length_size = pdu_length_size;
-                conf.qos_id_size = qos_id_size;
                 conf.seqno_size = seqno_size;
                 conf.ttl_size = ttl_size;
                 conf.chk_size = chk_size;