Merged in sandervrijders/ouroboros/be-aes (pull request #395)

ipcpd: normal: Add connection manager

13 files changed, 452 insertions, 91 deletions

diff --git a/include/ouroboros/cacep.h b/include/ouroboros/cacep.h
index c7b7f38c..b6fb8625 100644
--- a/include/ouroboros/cacep.h
+++ b/include/ouroboros/cacep.h
@@ -32,15 +32,12 @@ enum proto_concrete_syntax {
         PROTO_FIXED
 };
 
-struct conn_info{
+struct conn_info {
         char                       ae_name[64];
         char                       protocol[64];
         uint32_t                   pref_version;
         enum proto_concrete_syntax pref_syntax;
-        union {
-                char     name[64];
-                uint64_t addr;
-        } ae;
+        uint64_t                   addr;
 };
 
 int cacep_snd(int                      fd,
diff --git a/src/ipcpd/ipcp.h b/src/ipcpd/ipcp.h
index d0b5e022..07c72791 100644
--- a/src/ipcpd/ipcp.h
+++ b/src/ipcpd/ipcp.h
@@ -65,7 +65,7 @@ struct ipcp {
         enum ipcp_type     type;
         char *             dif_name;
 
-        uint64_t           address;
+        uint64_t           dt_addr;
 
         struct ipcp_ops *  ops;
         int                irmd_fd;
diff --git a/src/ipcpd/normal/CMakeLists.txt b/src/ipcpd/normal/CMakeLists.txt
index 7e10cc0d..70742336 100644
--- a/src/ipcpd/normal/CMakeLists.txt
+++ b/src/ipcpd/normal/CMakeLists.txt
@@ -20,6 +20,7 @@ set(SOURCE_FILES
   # Add source files here
   addr_auth.c
   cdap_flow.c
+  connmgr.c
   dir.c
   enroll.c
   fmgr.c
diff --git a/src/ipcpd/normal/connmgr.c b/src/ipcpd/normal/connmgr.c
new file mode 100644
index 00000000..387c38fd
--- /dev/null
+++ b/src/ipcpd/normal/connmgr.c
@@ -0,0 +1,350 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2017
+ *
+ * Handles the different AP connections
+ *
+ *    Sander Vrijders   <sander.vrijders@ugent.be>
+ *    Dimitri Staessens <dimitri.staessens@ugent.be>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2 as
+ * published by the Free Software Foundation.
+ *
+ * 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>
+#include <ouroboros/logs.h>
+#include <ouroboros/dev.h>
+#include <ouroboros/cacep.h>
+#include <ouroboros/cdap.h>
+#include <ouroboros/errno.h>
+
+#include "ae.h"
+#include "connmgr.h"
+#include "enroll.h"
+#include "fmgr.h"
+#include "frct.h"
+#include "ipcp.h"
+#include "ribmgr.h"
+
+#include <pthread.h>
+#include <string.h>
+#include <stdlib.h>
+#include <assert.h>
+
+#define FRCT_PROTO "frct"
+
+struct ae_conn {
+        struct list_head next;
+        struct conn      conn;
+};
+
+struct ae {
+        struct list_head next;
+        struct conn_info info;
+
+        struct list_head conn_list;
+        pthread_cond_t   conn_cond;
+        pthread_mutex_t  conn_lock;
+};
+
+struct {
+        pthread_t        acceptor;
+
+        struct list_head aes;
+        pthread_mutex_t  aes_lock;
+} connmgr;
+
+static int add_ae_conn(struct ae *        ae,
+                       int                fd,
+                       qosspec_t          qs,
+                       struct conn_info * rcv_info)
+{
+        struct ae_conn * ae_conn = NULL;
+
+        ae_conn = malloc(sizeof(*ae_conn));
+        if (ae_conn == NULL) {
+                log_err("Not enough memory.");
+                return -1;
+        }
+
+        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;
+
+        list_for_each(p, &connmgr.aes) {
+                struct ae * ae = list_entry(p, struct ae, next);
+                if (strcmp(ae->info.ae_name, name) == 0)
+                        return ae;
+        }
+
+        return NULL;
+}
+
+static void * flow_acceptor(void * o)
+{
+        int               fd;
+        qosspec_t         qs;
+        struct conn_info  rcv_info;
+        struct conn_info  fail_info;
+        struct ae *       ae = NULL;
+
+        (void) o;
+
+        memset(&fail_info, 0, sizeof(fail_info));
+
+        while (true) {
+                pthread_rwlock_rdlock(&ipcpi.state_lock);
+
+                if (ipcp_get_state() != IPCP_OPERATIONAL) {
+                        pthread_rwlock_unlock(&ipcpi.state_lock);
+                        log_info("Shutting down flow acceptor.");
+                        return 0;
+                }
+
+                pthread_rwlock_unlock(&ipcpi.state_lock);
+
+                fd = flow_accept(&qs);
+                if (fd < 0) {
+                        if (fd != -EIRMD)
+                                log_warn("Flow accept failed: %d", fd);
+                        continue;
+                }
+
+                if (flow_alloc_resp(fd, 0)) {
+                        log_err("Failed to respond to flow alloc request.");
+                        continue;
+                }
+
+                if (cacep_rcv(fd, &rcv_info)) {
+                        log_err("Error establishing application connection.");
+                        flow_dealloc(fd);
+                        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 {
+                        cacep_snd(fd, &fail_info);
+                        flow_dealloc(fd);
+                }
+        }
+
+        return (void *) 0;
+}
+
+int connmgr_init(void)
+{
+        list_head_init(&connmgr.aes);
+
+        if (pthread_mutex_init(&connmgr.aes_lock, NULL))
+                return -1;
+
+        return 0;
+}
+
+int connmgr_start(void)
+{
+        pthread_create(&connmgr.acceptor, NULL, flow_acceptor, NULL);
+
+        return 0;
+}
+
+void connmgr_stop(void)
+{
+        pthread_cancel(connmgr.acceptor);
+        pthread_join(connmgr.acceptor, NULL);
+}
+
+void connmgr_fini(void)
+{
+        struct list_head * p = NULL;
+        struct list_head * n = NULL;
+
+        pthread_mutex_lock(&connmgr.aes_lock);
+
+        list_for_each_safe(p, n, &connmgr.aes) {
+                struct ae * e = list_entry(p, struct ae, next);
+                connmgr_ae_destroy(e);
+        }
+
+        pthread_mutex_unlock(&connmgr.aes_lock);
+
+        pthread_mutex_destroy(&connmgr.aes_lock);
+}
+
+struct ae * connmgr_ae_create(struct conn_info info)
+{
+        struct ae * ae;
+
+        ae = malloc(sizeof(*ae));
+        if (ae == NULL)
+                return NULL;
+
+        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_cond_init(&ae->conn_cond, NULL)) {
+                pthread_mutex_destroy(&ae->conn_lock);
+                free(ae);
+                return NULL;
+        }
+
+        pthread_mutex_lock(&connmgr.aes_lock);
+        list_add(&ae->next, &connmgr.aes);
+        pthread_mutex_unlock(&connmgr.aes_lock);
+
+        return ae;
+}
+
+void connmgr_ae_destroy(struct ae * ae)
+{
+        struct list_head * p = NULL;
+        struct list_head * n = NULL;
+
+        assert(ae);
+
+        pthread_mutex_lock(&connmgr.aes_lock);
+        pthread_mutex_lock(&ae->conn_lock);
+
+        list_for_each_safe(p, n, &ae->conn_list) {
+                struct ae_conn * e = list_entry(p, struct ae_conn, next);
+                list_del(&e->next);
+                free(e);
+        }
+
+        pthread_mutex_unlock(&ae->conn_lock);
+
+        pthread_cond_destroy(&ae->conn_cond);
+        pthread_mutex_destroy(&ae->conn_lock);
+
+        list_del(&ae->next);
+
+        pthread_mutex_unlock(&connmgr.aes_lock);
+
+        free(ae);
+}
+
+int connmgr_alloc(struct ae *   ae,
+                  char *        dst_name,
+                  qosspec_t     qs,
+                  struct conn * conn)
+{
+        assert(ae);
+        assert(dst_name);
+        assert(conn);
+
+        memset(&conn->conn_info, 0, sizeof(conn->conn_info));
+
+        conn->flow_info.fd = flow_alloc(dst_name, &qs);
+        if (conn->flow_info.fd < 0) {
+                log_err("Failed to allocate flow to %s.", dst_name);
+                return -1;
+        }
+
+        conn->flow_info.qs = qs;
+
+        if (flow_alloc_res(conn->flow_info.fd)) {
+                log_err("Flow allocation to %s failed.", dst_name);
+                flow_dealloc(conn->flow_info.fd);
+                return -1;
+        }
+
+        if (cacep_snd(conn->flow_info.fd, &ae->info)) {
+                log_err("Failed to create application connection.");
+                flow_dealloc(conn->flow_info.fd);
+                return -1;
+        }
+
+        if (cacep_rcv(conn->flow_info.fd, &conn->conn_info)) {
+                log_err("Failed to connect to application.");
+                flow_dealloc(conn->flow_info.fd);
+                return -1;
+        }
+
+        if (strcmp(ae->info.protocol, conn->conn_info.protocol) ||
+            ae->info.pref_version != conn->conn_info.pref_version ||
+            ae->info.pref_syntax != conn->conn_info.pref_syntax) {
+                flow_dealloc(conn->flow_info.fd);
+                return -1;
+        }
+
+        return 0;
+}
+
+int connmgr_wait(struct ae *   ae,
+                 struct conn * conn)
+{
+        struct ae_conn * ae_conn;
+
+        assert(ae);
+        assert(conn);
+
+        pthread_mutex_lock(&ae->conn_lock);
+
+        pthread_cleanup_push((void(*)(void *))pthread_mutex_unlock,
+                             (void *) &ae->conn_lock);
+
+        while (list_is_empty(&ae->conn_list))
+                pthread_cond_wait(&ae->conn_cond, &ae->conn_lock);
+
+        ae_conn = list_first_entry((&ae->conn_list), struct ae_conn, next);
+        if (ae_conn == NULL) {
+                pthread_mutex_unlock(&ae->conn_lock);
+                return -1;
+        }
+
+        *conn = ae_conn->conn;
+
+        list_del(&ae_conn->next);
+        free(ae_conn);
+
+        pthread_cleanup_pop(true);
+
+        return 0;
+}
diff --git a/src/ipcpd/normal/connmgr.h b/src/ipcpd/normal/connmgr.h
new file mode 100644
index 00000000..bfb3d762
--- /dev/null
+++ b/src/ipcpd/normal/connmgr.h
@@ -0,0 +1,57 @@
+/*
+ * Ouroboros - Copyright (C) 2016 - 2017
+ *
+ * Handles the different AP connections
+ *
+ *    Sander Vrijders   <sander.vrijders@ugent.be>
+ *    Dimitri Staessens <dimitri.staessens@ugent.be>
+ *
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2 as
+ * published by the Free Software Foundation.
+ *
+ * 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.
+ */
+
+#ifndef OUROBOROS_IPCPD_NORMAL_CONNMGR_H
+#define OUROBOROS_IPCPD_NORMAL_CONNMGR_H
+
+#include <ouroboros/cacep.h>
+#include <ouroboros/qos.h>
+
+struct conn {
+        struct conn_info conn_info;
+        struct flow_info {
+                int              fd;
+                qosspec_t        qs;
+        } flow_info;
+};
+
+int         connmgr_init(void);
+
+void        connmgr_fini(void);
+
+int         connmgr_start(void);
+
+void        connmgr_stop(void);
+
+struct ae * connmgr_ae_create(struct conn_info info);
+
+void        connmgr_ae_destroy(struct ae * ae);
+
+int         connmgr_alloc(struct ae *   ae,
+                          char *        dst_name,
+                          qosspec_t     qs,
+                          struct conn * conn);
+
+int         connmgr_wait(struct ae *   ae,
+                         struct conn * conn);
+
+#endif /* OUROBOROS_IPCPD_NORMAL_CONNMGR_H */
diff --git a/src/ipcpd/normal/enroll.c b/src/ipcpd/normal/enroll.c
index 5c7ebd7e..9c3b9973 100644
--- a/src/ipcpd/normal/enroll.c
+++ b/src/ipcpd/normal/enroll.c
@@ -197,7 +197,7 @@ int enroll_boot(char * dst_name)
 
         delta_t = ts_diff_ms(&t0, &rtt);
 
-        assert (len == 2 * sizeof (uint64_t));
+        assert(len == 2 * sizeof (uint64_t));
 
         rtt.tv_sec  = ntoh64(((uint64_t *) data)[0]);
         rtt.tv_nsec = ntoh64(((uint64_t *) data)[1]);
diff --git a/src/ipcpd/normal/fmgr.c b/src/ipcpd/normal/fmgr.c
index 0c927fc7..34724ddd 100644
--- a/src/ipcpd/normal/fmgr.c
+++ b/src/ipcpd/normal/fmgr.c
@@ -162,7 +162,7 @@ void * fmgr_nm1_sdu_reader(void * o)
 
                         shm_pci_des(sdb, &pci);
 
-                        if (pci.dst_addr != ipcpi.address) {
+                        if (pci.dst_addr != ipcpi.dt_addr) {
                                 log_dbg("PDU needs to be forwarded.");
 
                                 if (pci.ttl == 0) {
diff --git a/src/ipcpd/normal/frct.c b/src/ipcpd/normal/frct.c
index 915feaf8..c9b23060 100644
--- a/src/ipcpd/normal/frct.c
+++ b/src/ipcpd/normal/frct.c
@@ -285,7 +285,7 @@ cep_id_t frct_i_create(uint64_t   address,
 
         pci.pdu_type = PDU_TYPE_MGMT;
         pci.dst_addr = address;
-        pci.src_addr = ipcpi.address;
+        pci.src_addr = ipcpi.dt_addr;
         pci.dst_cep_id = 0;
         pci.src_cep_id = id;
         pci.seqno = 0;
@@ -330,7 +330,7 @@ int frct_i_accept(cep_id_t   id,
 
         pci.pdu_type = PDU_TYPE_MGMT;
         pci.dst_addr = instance->r_address;
-        pci.src_addr = ipcpi.address;
+        pci.src_addr = ipcpi.dt_addr;
         pci.dst_cep_id = instance->r_cep_id;
         pci.src_cep_id = instance->cep_id;
         pci.seqno = 0;
@@ -367,7 +367,7 @@ int frct_i_destroy(cep_id_t   id,
 
         pci.pdu_type = PDU_TYPE_MGMT;
         pci.dst_addr = instance->r_address;
-        pci.src_addr = ipcpi.address;
+        pci.src_addr = ipcpi.dt_addr;
         pci.dst_cep_id = instance->r_cep_id;
         pci.src_cep_id = instance->cep_id;
         pci.seqno = 0;
@@ -413,7 +413,7 @@ int frct_i_write_sdu(cep_id_t             id,
 
         pci.pdu_type = PDU_TYPE_DTP;
         pci.dst_addr = instance->r_address;
-        pci.src_addr = ipcpi.address;
+        pci.src_addr = ipcpi.dt_addr;
         pci.dst_cep_id = instance->r_cep_id;
         pci.src_cep_id = instance->cep_id;
         pci.seqno = (instance->seqno)++;
diff --git a/src/ipcpd/normal/gam.c b/src/ipcpd/normal/gam.c
index f98c0d4f..212cfd83 100644
--- a/src/ipcpd/normal/gam.c
+++ b/src/ipcpd/normal/gam.c
@@ -185,7 +185,7 @@ int gam_flow_arr(struct gam * instance,
         strcpy(snd_info.protocol, CDAP_PROTO);
         snd_info.pref_version = 1;
         snd_info.pref_syntax = PROTO_GPB;
-        snd_info.ae.addr = ipcpi.address;
+        snd_info.addr = ipcpi.dt_addr;
 
         if (cacep_rcv(fd, rcv_info)) {
                 log_err("Error establishing application connection.");
@@ -266,7 +266,7 @@ int gam_flow_alloc(struct gam * instance,
         strcpy(snd_info.protocol, CDAP_PROTO);
         snd_info.pref_version = 1;
         snd_info.pref_syntax = PROTO_GPB;
-        snd_info.ae.addr = ipcpi.address;
+        snd_info.addr = ipcpi.dt_addr;
 
         if (cacep_snd(fd, &snd_info)) {
                 log_err("Failed to create application connection.");
diff --git a/src/ipcpd/normal/main.c b/src/ipcpd/normal/main.c
index c75a74d6..3e5907a8 100644
--- a/src/ipcpd/normal/main.c
+++ b/src/ipcpd/normal/main.c
@@ -24,7 +24,6 @@
 
 #include <ouroboros/config.h>
 #include <ouroboros/logs.h>
-#include <ouroboros/dev.h>
 #include <ouroboros/ipcp-dev.h>
 #include <ouroboros/time_utils.h>
 #include <ouroboros/irm.h>
@@ -33,11 +32,10 @@
 #include <ouroboros/errno.h>
 
 #include "addr_auth.h"
-#include "ae.h"
+#include "connmgr.h"
 #include "dir.h"
 #include "enroll.h"
 #include "fmgr.h"
-#include "frct.h"
 #include "ipcp.h"
 #include "ribconfig.h"
 #include "ribmgr.h"
@@ -45,16 +43,11 @@
 #include <stdbool.h>
 #include <signal.h>
 #include <stdlib.h>
-#include <pthread.h>
 #include <string.h>
 #include <assert.h>
 #include <inttypes.h>
 
-#define THIS_TYPE    IPCP_NORMAL
-
-struct {
-        pthread_t acceptor;
-} normal;
+#define THIS_TYPE IPCP_NORMAL
 
 void ipcp_sig_handler(int         sig,
                       siginfo_t * info,
@@ -82,53 +75,6 @@ void ipcp_sig_handler(int         sig,
         }
 }
 
-static void * flow_acceptor(void * o)
-{
-        int       fd;
-        qosspec_t qs;
-        /* FIXME: Remove once correct AE is known. */
-        char *    ae_name = ENROLL_AE;
-
-        (void) o;
-
-        while (true) {
-                pthread_rwlock_rdlock(&ipcpi.state_lock);
-
-                if (ipcp_get_state() != IPCP_OPERATIONAL) {
-                        pthread_rwlock_unlock(&ipcpi.state_lock);
-                        log_info("Shutting down flow acceptor.");
-                        return 0;
-                }
-
-                pthread_rwlock_unlock(&ipcpi.state_lock);
-
-                fd = flow_accept(&qs);
-                if (fd < 0) {
-                        if (fd != -EIRMD)
-                                log_warn("Flow accept failed: %d", fd);
-                        continue;
-                }
-
-                /* FIXME: Perform CACEP at this point */
-
-                if (strcmp(ae_name, ENROLL_AE) == 0) {
-                        enroll_handle(fd);
-                } else if (strcmp(ae_name, MGMT_AE) == 0) {
-                        ribmgr_flow_arr(fd, qs);
-                } else if (strcmp(ae_name, DT_AE) == 0) {
-                        fmgr_nm1_flow_arr(fd, qs);
-                } else {
-                        log_dbg("Flow allocation request for unknown AE %s.",
-                                ae_name);
-                        if (flow_alloc_resp(fd, -1))
-                                log_warn("Failed to reply to flow allocation.");
-                        flow_dealloc(fd);
-                }
-        }
-
-        return (void *) 0;
-}
-
 /*
  * Boots the IPCP off information in the rib.
  * Common function after bootstrap or enroll.
@@ -153,15 +99,21 @@ static int boot_components(void)
         }
 
         if (rib_add(MEMBERS_PATH, ipcpi.name)) {
-                log_warn("Failed to add name to " MEMBERS_PATH);
+                log_err("Failed to add name to " MEMBERS_PATH);
                 return -1;
         }
 
         log_dbg("Starting components.");
 
+        if (connmgr_init()) {
+                log_err("Failed to init ap connection manager");
+                return -1;
+        }
+
         if (rib_read(BOOT_PATH "/addr_auth/type", &pa, sizeof(pa))
             != sizeof(pa)) {
                 log_err("Failed to read policy for address authority.");
+                connmgr_fini();
                 return -1;
         }
 
@@ -170,20 +122,22 @@ static int boot_components(void)
                 return -1;
         }
 
-        ipcpi.address = addr_auth_address();
-        if (ipcpi.address == 0) {
+        ipcpi.dt_addr = addr_auth_address();
+        if (ipcpi.dt_addr == 0) {
                 log_err("Failed to get a valid address.");
                 addr_auth_fini();
+                connmgr_fini();
                 return -1;
         }
 
-        log_dbg("IPCP got address %" PRIu64 ".", ipcpi.address);
+        log_dbg("IPCP got address %" PRIu64 ".", ipcpi.dt_addr);
 
         log_dbg("Starting ribmgr.");
 
         if (ribmgr_init()) {
                 log_err("Failed to initialize RIB manager.");
                 addr_auth_fini();
+                connmgr_fini();
                 return -1;
         }
 
@@ -191,6 +145,7 @@ static int boot_components(void)
                 log_err("Failed to initialize directory.");
                 ribmgr_fini();
                 addr_auth_fini();
+                connmgr_fini();
                 return -1;
         }
 
@@ -200,6 +155,7 @@ static int boot_components(void)
                 dir_fini();
                 ribmgr_fini();
                 addr_auth_fini();
+                connmgr_fini();
                 log_err("Failed to start flow manager.");
                 return -1;
         }
@@ -209,19 +165,21 @@ static int boot_components(void)
                 dir_fini();
                 ribmgr_fini();
                 addr_auth_fini();
+                connmgr_fini();
                 log_err("Failed to initialize FRCT.");
                 return -1;
         }
 
         ipcp_set_state(IPCP_OPERATIONAL);
 
-        if (pthread_create(&normal.acceptor, NULL, flow_acceptor, NULL)) {
+        if (connmgr_start()) {
                 ipcp_set_state(IPCP_INIT);
                 fmgr_fini();
                 dir_fini();
                 ribmgr_fini();
                 addr_auth_fini();
-                log_err("Failed to create acceptor thread.");
+                connmgr_fini();
+                log_err("Failed to start AP connection manager.");
                 return -1;
         }
 
@@ -230,8 +188,7 @@ static int boot_components(void)
 
 void shutdown_components(void)
 {
-        pthread_cancel(normal.acceptor);
-        pthread_join(normal.acceptor, NULL);
+        connmgr_stop();
 
         frct_fini();
 
@@ -242,6 +199,8 @@ void shutdown_components(void)
         ribmgr_fini();
 
         addr_auth_fini();
+
+        connmgr_fini();
 }
 
 static int normal_ipcp_enroll(char * dst_name)
@@ -410,9 +369,9 @@ static struct ipcp_ops normal_ops = {
         .ipcp_name_reg        = dir_name_reg,
         .ipcp_name_unreg      = dir_name_unreg,
         .ipcp_name_query      = dir_name_query,
-        .ipcp_flow_alloc      = NULL, /* fmgr_np1_alloc, */
-        .ipcp_flow_alloc_resp = NULL, /* fmgr_np1_alloc_resp, */
-        .ipcp_flow_dealloc    = NULL, /* fmgr_np1_dealloc */
+        .ipcp_flow_alloc      = fmgr_np1_alloc,
+        .ipcp_flow_alloc_resp = fmgr_np1_alloc_resp,
+        .ipcp_flow_dealloc    = fmgr_np1_dealloc
 };
 
 int main(int    argc,
diff --git a/src/ipcpd/normal/pol/complete.c b/src/ipcpd/normal/pol/complete.c
index 68f43e81..daf8c9bf 100644
--- a/src/ipcpd/normal/pol/complete.c
+++ b/src/ipcpd/normal/pol/complete.c
@@ -38,7 +38,7 @@
 
 struct neighbor {
         struct list_head next;
-        char *           neighbor;
+        uint64_t         neighbor;
 };
 
 struct complete {
@@ -135,7 +135,6 @@ void complete_destroy(void * o)
         list_for_each_safe(p, n, &complete->neighbors) {
                 struct neighbor * e = list_entry(p, struct neighbor, next);
                 list_del(&e->next);
-                free(e->neighbor);
                 free(e);
         }
 
@@ -168,7 +167,7 @@ int complete_accept_flow(void *                    o,
         list_for_each(pos, &complete->neighbors) {
                 struct neighbor * e = list_entry(pos, struct neighbor, next);
                 /* FIXME: figure out union type and check name or address */
-                if (strcmp(e->neighbor, info->ae.name) == 0) {
+                if (e->neighbor == info->addr) {
                         pthread_mutex_unlock(&complete->neighbors_lock);
                         return -1;
                 }
@@ -186,13 +185,7 @@ int complete_accept_flow(void *                    o,
 
         list_head_init(&n->next);
 
-        /* FIXME: figure out union type and check name or address */
-        n->neighbor = strdup(info->ae.name);
-        if (n->neighbor == NULL) {
-                pthread_mutex_unlock(&complete->neighbors_lock);
-                free(n);
-                return -1;
-        }
+        n->neighbor = info->addr;
 
         list_add(&n->next, &complete->neighbors);
 
diff --git a/src/lib/cacep.c b/src/lib/cacep.c
index abff0aaa..a2c5c3d2 100644
--- a/src/lib/cacep.c
+++ b/src/lib/cacep.c
@@ -49,10 +49,12 @@ int read_msg(int                fd,
         if (msg == NULL)
                 return -1;
 
+        strcpy(info->ae_name, msg->ae_name);
         strcpy(info->protocol, msg->protocol);
 
         info->pref_version = msg->pref_version;
         info->pref_syntax  = msg->pref_syntax;
+        info->addr         = msg->address;
 
         cacep_msg__free_unpacked(msg, NULL);
 
@@ -68,6 +70,7 @@ static int send_msg(int                      fd,
 
         msg.ae_name      = (char *) info->ae_name;
         msg.protocol     = (char *) info->protocol;
+        msg.address      = info->addr;
         msg.pref_version = info->pref_version;
         msg.pref_syntax  = info->pref_syntax;
         if (msg.pref_syntax < 0)
diff --git a/src/lib/cacep.proto b/src/lib/cacep.proto
index 3e1291f6..cdeaa0b7 100644
--- a/src/lib/cacep.proto
+++ b/src/lib/cacep.proto
@@ -30,4 +30,5 @@ message cacep_msg {
         repeated int32 supp_version = 4;
         required int32 pref_syntax  = 5;
         repeated int32 supp_syntax  = 6;
+        required uint64 address     = 7;
 }
\ No newline at end of file