ipcpd: normal: Refactor application entities and add neighbors struct

This refactors the different Application Entities of the normal
IPCP. They all listen to and use the connection manager to establish
new application connections.

This commit also adds a neighbors struct to the normal IPCP. It
contains neighbor structs that contain application
connection. Notifiers can be registered in case a neighbor changes
(added, removed, QoS changed).

The flow manager has an instance of this neighbors struct and listens
to these events to update its flow set. The routing component also
listens to these events so that it can update the FSDB if needed. The
flow manager now also creates the PFF instances and the routing
instances per QoS cube.

The RIB manager also uses this an instance of the neighbors struct and
listens to neighbor events as well.

1 files changed, 178 insertions, 114 deletions

diff --git a/src/ipcpd/normal/enroll.c b/src/ipcpd/normal/enroll.c
index 9c3b9973..25460161 100644
--- a/src/ipcpd/normal/enroll.c
+++ b/src/ipcpd/normal/enroll.c
@@ -22,6 +22,8 @@
 
 #include <ouroboros/config.h>
 #include <ouroboros/endian.h>
+#include <ouroboros/errno.h>
+#include <ouroboros/cdap.h>
 #include <ouroboros/time_utils.h>
 #include <ouroboros/dev.h>
 #include <ouroboros/logs.h>
@@ -29,29 +31,35 @@
 #include <ouroboros/errno.h>
 
 #include "ae.h"
-#include "cdap_flow.h"
+#include "connmgr.h"
 #include "ribconfig.h"
 
 #include <assert.h>
 #include <stdlib.h>
 #include <string.h>
+#include <pthread.h>
 
 /* Symbolic, will return current time */
 #define TIME_NAME               "localtime"
 #define TIME_PATH               DLR TIME_NAME
 #define ENROLL_WARN_TIME_OFFSET 20
 
-int enroll_handle(int fd)
+struct {
+        struct ae * ae;
+        pthread_t   listener;
+} enroll;
+
+static void * enroll_handle(void * o)
 {
-        struct cdap_flow * flow;
-        struct conn_info   info;
-        cdap_key_t         key;
-        enum cdap_opcode   oc;
-        char *             name;
-        uint8_t *          buf;
-        uint8_t *          data;
-        ssize_t            len;
-        uint32_t           flags;
+        struct cdap *    cdap;
+        struct conn      conn;
+        cdap_key_t       key;
+        enum cdap_opcode oc;
+        char *           name;
+        uint8_t *        buf;
+        uint8_t *        data;
+        ssize_t          len;
+        uint32_t         flags;
 
         bool boot_r     = false;
         bool members_r  = false;
@@ -61,98 +69,107 @@ int enroll_handle(int fd)
         char * members_ro = MEMBERS_PATH;
         char * dif_ro     = DIF_PATH;
 
-        memset(&info, 0, sizeof(info));
-
-        strcpy(info.ae_name, ENROLL_AE);
-        strcpy(info.protocol, CDAP_PROTO);
-        info.pref_version = 1;
-        info.pref_syntax = PROTO_GPB;
+        (void) o;
 
-        flow = cdap_flow_arr(fd, 0, &info);
-        if (flow == NULL) {
-                log_err("Failed to auth enrollment request.");
-                flow_dealloc(fd);
-                return -1;
-        }
-
-        while (!(boot_r && members_r && dif_name_r)) {
-                key = cdap_request_wait(flow->ci, &oc, &name, &data,
-                                        (size_t *) &len , &flags);
-                assert(key >= 0);
-                assert(name);
-
-                if (data != NULL) {
-                        free(data);
-                        log_warn("Received data with enrollment request.");
-                }
-
-                if (oc != CDAP_READ) {
-                        log_warn("Invalid request.");
-                        cdap_reply_send(flow->ci, key, -1, NULL, 0);
-                        cdap_flow_dealloc(flow);
-                        free(name);
-                        return -1;
+        while (true) {
+                if (connmgr_wait(enroll.ae, &conn)) {
+                        log_err("Failed to get next connection.");
+                        continue;
                 }
 
-                if (strcmp(name, boot_ro) == 0) {
-                        boot_r = true;
-                } else if (strcmp(name, members_ro) == 0) {
-                        members_r = true;
-                } else if (strcmp(name, dif_ro) == 0) {
-                        dif_name_r = true;
-                } else if (strcmp(name, TIME_PATH) == 0) {
-                        struct timespec t;
-                        uint64_t buf[2];
-                        clock_gettime(CLOCK_REALTIME, &t);
-                        buf[0] = hton64(t.tv_sec);
-                        buf[1] = hton64(t.tv_nsec);
-                        cdap_reply_send(flow->ci, key, 0, buf, sizeof(buf));
-                        free(name);
+                cdap = cdap_create(conn.flow_info.fd);
+                if (cdap == NULL) {
+                        log_err("Failed to instantiate CDAP.");
+                        flow_dealloc(conn.flow_info.fd);
                         continue;
-                } else {
-                        log_warn("Illegal read: %s.", name);
-                        cdap_reply_send(flow->ci, key, -1, NULL, 0);
-                        cdap_flow_dealloc(flow);
-                        free(name);
-                        return -1;
                 }
 
-                len = rib_pack(name, &buf, PACK_HASH_ROOT);
-                if (len < 0) {
-                        log_err("Failed to pack %s.", name);
-                        cdap_reply_send(flow->ci, key, -1, NULL, 0);
-                        cdap_flow_dealloc(flow);
-                        free(name);
-                        return -1;
-                }
+                while (!(boot_r && members_r && dif_name_r)) {
+                        key = cdap_request_wait(cdap, &oc, &name, &data,
+                                                (size_t *) &len , &flags);
+                        assert(key >= 0);
+                        assert(name);
+
+                        if (data != NULL) {
+                                free(data);
+                                log_warn("Received data with enroll request.");
+                        }
+
+                        if (oc != CDAP_READ) {
+                                log_warn("Invalid request.");
+                                cdap_reply_send(cdap, key, -1, NULL, 0);
+                                cdap_destroy(cdap);
+                                flow_dealloc(conn.flow_info.fd);
+                                free(name);
+                                continue;
+                        }
+
+                        if (strcmp(name, boot_ro) == 0) {
+                                boot_r = true;
+                        } else if (strcmp(name, members_ro) == 0) {
+                                members_r = true;
+                        } else if (strcmp(name, dif_ro) == 0) {
+                                dif_name_r = true;
+                        } else if (strcmp(name, TIME_PATH) == 0) {
+                                struct timespec t;
+                                uint64_t buf[2];
+                                clock_gettime(CLOCK_REALTIME, &t);
+                                buf[0] = hton64(t.tv_sec);
+                                buf[1] = hton64(t.tv_nsec);
+                                cdap_reply_send(cdap, key, 0, buf, sizeof(buf));
+                                free(name);
+                                cdap_destroy(cdap);
+                                flow_dealloc(conn.flow_info.fd);
+                                continue;
+                        } else {
+                                log_warn("Illegal read: %s.", name);
+                                cdap_reply_send(cdap, key, -1, NULL, 0);
+                                cdap_destroy(cdap);
+                                flow_dealloc(conn.flow_info.fd);
+                                free(name);
+                                continue;
+                        }
+
+                        len = rib_pack(name, &buf, PACK_HASH_ROOT);
+                        if (len < 0) {
+                                log_err("Failed to pack %s.", name);
+                                cdap_reply_send(cdap, key, -1, NULL, 0);
+                                cdap_destroy(cdap);
+                                flow_dealloc(conn.flow_info.fd);
+                                free(name);
+                                continue;
+                        }
+
+                        log_dbg("Packed %s (%zu bytes).", name, len);
 
-                log_dbg("Packed %s (%zu bytes).", name, len);
+                        free(name);
 
-                free(name);
+                        if (cdap_reply_send(cdap, key, 0, buf, len)) {
+                                log_err("Failed to send CDAP reply.");
+                                cdap_destroy(cdap);
+                                flow_dealloc(conn.flow_info.fd);
+                                continue;
+                        }
 
-                if (cdap_reply_send(flow->ci, key, 0, buf, len)) {
-                        log_err("Failed to send CDAP reply.");
-                        cdap_flow_dealloc(flow);
-                        return -1;
+                        free(buf);
                 }
 
-                free(buf);
-        }
-
-        log_dbg("Sent boot info to new member.");
+                log_dbg("Sent boot info to new member.");
 
-        cdap_flow_dealloc(flow);
+                cdap_destroy(cdap);
+                flow_dealloc(conn.flow_info.fd);
+        }
 
         return 0;
 }
 
 int enroll_boot(char * dst_name)
 {
-        struct cdap_flow * flow;
-        struct conn_info   info;
-        cdap_key_t         key;
-        uint8_t *          data;
-        size_t             len;
+        struct cdap * cdap;
+        cdap_key_t    key;
+        uint8_t *     data;
+        size_t        len;
+        struct conn   conn;
 
         struct timespec t0;
         struct timespec rtt;
@@ -163,16 +180,14 @@ int enroll_boot(char * dst_name)
         char * members_ro = MEMBERS_PATH;
         char * dif_ro     = DIF_PATH;
 
-        memset(&info, 0, sizeof(info));
-
-        strcpy(info.ae_name, ENROLL_AE);
-        strcpy(info.protocol, CDAP_PROTO);
-        info.pref_version = 1;
-        info.pref_syntax  = PROTO_GPB;
+        if (connmgr_alloc(enroll.ae, dst_name, NULL, &conn)) {
+                log_err("Failed to get connection.");
+                return -1;
+        }
 
-        flow = cdap_flow_alloc(dst_name, NULL, &info);
-        if (flow == NULL) {
-                log_err("Failed to allocate flow for enrollment request.");
+        cdap = cdap_create(conn.flow_info.fd);
+        if (cdap == NULL) {
+                log_err("Failed to instantiate CDAP.");
                 return -1;
         }
 
@@ -180,16 +195,18 @@ int enroll_boot(char * dst_name)
 
         clock_gettime(CLOCK_REALTIME, &t0);
 
-        key = cdap_request_send(flow->ci, CDAP_READ, TIME_PATH, NULL, 0, 0);
+        key = cdap_request_send(cdap, CDAP_READ, TIME_PATH, NULL, 0, 0);
         if (key < 0) {
                 log_err("Failed to send CDAP request.");
-                cdap_flow_dealloc(flow);
+                cdap_destroy(cdap);
+                flow_dealloc(conn.flow_info.fd);
                 return -1;
         }
 
-        if (cdap_reply_wait(flow->ci, key, &data, &len)) {
+        if (cdap_reply_wait(cdap, key, &data, &len)) {
                 log_err("Failed to get CDAP reply.");
-                cdap_flow_dealloc(flow);
+                cdap_destroy(cdap);
+                flow_dealloc(conn.flow_info.fd);
                 return -1;
         }
 
@@ -207,16 +224,18 @@ int enroll_boot(char * dst_name)
 
         free(data);
 
-        key = cdap_request_send(flow->ci, CDAP_READ, boot_ro, NULL, 0, 0);
+        key = cdap_request_send(cdap, CDAP_READ, boot_ro, NULL, 0, 0);
         if (key < 0) {
                 log_err("Failed to send CDAP request.");
-                cdap_flow_dealloc(flow);
+                cdap_destroy(cdap);
+                flow_dealloc(conn.flow_info.fd);
                 return -1;
         }
 
-        if (cdap_reply_wait(flow->ci, key, &data, &len)) {
+        if (cdap_reply_wait(cdap, key, &data, &len)) {
                 log_err("Failed to get CDAP reply.");
-                cdap_flow_dealloc(flow);
+                cdap_destroy(cdap);
+                flow_dealloc(conn.flow_info.fd);
                 return -1;
         }
 
@@ -226,22 +245,25 @@ int enroll_boot(char * dst_name)
                 log_warn("Error unpacking RIB data.");
                 rib_del(boot_ro);
                 free(data);
-                cdap_flow_dealloc(flow);
+                cdap_destroy(cdap);
+                flow_dealloc(conn.flow_info.fd);
                 return -1;
         }
 
         log_dbg("Packed information inserted into RIB.");
 
-        key = cdap_request_send(flow->ci, CDAP_READ, members_ro, NULL, 0, 0);
+        key = cdap_request_send(cdap, CDAP_READ, members_ro, NULL, 0, 0);
         if (key < 0) {
                 log_err("Failed to send CDAP request.");
-                cdap_flow_dealloc(flow);
+                cdap_destroy(cdap);
+                flow_dealloc(conn.flow_info.fd);
                 return -1;
         }
 
-        if (cdap_reply_wait(flow->ci, key, &data, &len)) {
+        if (cdap_reply_wait(cdap, key, &data, &len)) {
                 log_err("Failed to get CDAP reply.");
-                cdap_flow_dealloc(flow);
+                cdap_destroy(cdap);
+                flow_dealloc(conn.flow_info.fd);
                 return -1;
         }
 
@@ -251,22 +273,25 @@ int enroll_boot(char * dst_name)
                 log_warn("Error unpacking RIB data.");
                 rib_del(boot_ro);
                 free(data);
-                cdap_flow_dealloc(flow);
+                cdap_destroy(cdap);
+                flow_dealloc(conn.flow_info.fd);
                 return -1;
         }
 
         log_dbg("Packed information inserted into RIB.");
 
-        key = cdap_request_send(flow->ci, CDAP_READ, dif_ro, NULL, 0, 0);
+        key = cdap_request_send(cdap, CDAP_READ, dif_ro, NULL, 0, 0);
         if (key < 0) {
                 log_err("Failed to send CDAP request.");
-                cdap_flow_dealloc(flow);
+                cdap_destroy(cdap);
+                flow_dealloc(conn.flow_info.fd);
                 return -1;
         }
 
-        if (cdap_reply_wait(flow->ci, key, &data, &len)) {
+        if (cdap_reply_wait(cdap, key, &data, &len)) {
                 log_err("Failed to get CDAP reply.");
-                cdap_flow_dealloc(flow);
+                cdap_destroy(cdap);
+                flow_dealloc(conn.flow_info.fd);
                 return -1;
         }
 
@@ -276,13 +301,52 @@ int enroll_boot(char * dst_name)
                 log_warn("Error unpacking RIB data.");
                 rib_del(boot_ro);
                 free(data);
-                cdap_flow_dealloc(flow);
+                cdap_destroy(cdap);
+                flow_dealloc(conn.flow_info.fd);
                 return -1;
         }
 
         log_dbg("Packed information inserted into RIB.");
 
-        cdap_flow_dealloc(flow);
+        cdap_destroy(cdap);
+        flow_dealloc(conn.flow_info.fd);
+
+        return 0;
+}
+
+int enroll_init(void)
+{
+        struct conn_info info;
+
+        memset(&info, 0, sizeof(info));
+
+        strcpy(info.ae_name, ENROLL_AE);
+        strcpy(info.protocol, CDAP_PROTO);
+        info.pref_version = 1;
+        info.pref_syntax  = PROTO_GPB;
+
+        enroll.ae = connmgr_ae_create(info);
+        if (enroll.ae == NULL)
+                return -1;
+
+        return 0;
+}
+
+void enroll_fini(void)
+{
+        connmgr_ae_destroy(enroll.ae);
+}
+
+int enroll_start(void)
+{
+        if (pthread_create(&enroll.listener, NULL, enroll_handle, NULL))
+                return -1;
 
         return 0;
 }
+
+void enroll_stop(void)
+{
+        pthread_cancel(enroll.listener);
+        pthread_join(enroll.listener, NULL);
+}