ipcpd: Refactor of normal IPCP

Reorganizes the normal IPCP a bit to make sure internal components do
not need to access the state of the IPCP. The IPCP has now a thread
calling accept and delegating it to the correct component based on the
AE name (this used to be in the fmgr).

Internal components are initialized upon enrollment or bootstrap of
the IPCP. If a step fails, the IPCP goes back to the INIT state, if
all components boot correctly, it goes to the operational state.

RIB synchronization is still done by sending a CDAP start/stop and
syncing with a ribmgr state, but needs revision later on.

1 files changed, 174 insertions, 103 deletions

diff --git a/src/ipcpd/normal/ribmgr.c b/src/ipcpd/normal/ribmgr.c
index 3b4a5784..ab492e7f 100644
--- a/src/ipcpd/normal/ribmgr.c
+++ b/src/ipcpd/normal/ribmgr.c
@@ -1,5 +1,5 @@
 /*
- * Ouroboros - Copyright (C) 2016
+ * Ouroboros - Copyright (C) 2016 - 2017
  *
  * RIB manager of the IPC Process
  *
@@ -29,21 +29,21 @@
 #include <ouroboros/ipcp-dev.h>
 #include <ouroboros/bitmap.h>
 #include <ouroboros/errno.h>
-
-#include <stdlib.h>
-#include <pthread.h>
-#include <string.h>
-#include <errno.h>
+#include <ouroboros/dev.h>
 
 #include "timerwheel.h"
 #include "addr_auth.h"
 #include "ribmgr.h"
 #include "dt_const.h"
-#include "frct.h"
-#include "ipcp.h"
 #include "ro.h"
 #include "pathname.h"
 #include "dir.h"
+#include "ae.h"
+
+#include <stdlib.h>
+#include <pthread.h>
+#include <string.h>
+#include <errno.h>
 
 #include "static_info.pb-c.h"
 typedef StaticInfoMsg static_info_msg_t;
@@ -73,7 +73,7 @@ struct rnode {
          * as an index in a B-tree
          */
 
-        /* If there are no children, this is a leaf */
+        /* If there are no children, this is a leaf. */
         struct rnode * child;
         struct rnode * sibling;
 
@@ -107,6 +107,14 @@ struct ro_id {
         char *           full_name;
 };
 
+enum ribmgr_state {
+        RIBMGR_NULL,
+        RIBMGR_INIT,
+        RIBMGR_OPERATIONAL,
+        RIBMGR_SHUTDOWN
+};
+
+/* FIXME: Extract rib from ribmgr. */
 struct {
         struct rnode *      root;
         pthread_mutex_t     ro_lock;
@@ -130,6 +138,10 @@ struct {
 
         struct addr_auth *  addr_auth;
         enum pol_addr_auth  addr_auth_type;
+
+        enum ribmgr_state   state;
+        pthread_cond_t      state_cond;
+        pthread_mutex_t     state_lock;
 } rib;
 
 void ribmgr_ro_created(const char * name,
@@ -138,15 +150,11 @@ void ribmgr_ro_created(const char * name,
 {
         static_info_msg_t * stat_msg;
 
-        pthread_rwlock_wrlock(&ipcpi.state_lock);
-        if (ipcp_get_state() == IPCP_CONFIG &&
-            strcmp(name, RIBMGR_PREFIX STAT_INFO) == 0) {
+        if (strcmp(name, RIBMGR_PREFIX STAT_INFO) == 0) {
                 LOG_DBG("Received static DIF information.");
 
                 stat_msg = static_info_msg__unpack(NULL, len, data);
                 if (stat_msg == NULL) {
-                        ipcp_set_state(IPCP_INIT);
-                        pthread_rwlock_unlock(&ipcpi.state_lock);
                         LOG_ERR("Failed to unpack static info message.");
                         return;
                 }
@@ -161,17 +169,8 @@ void ribmgr_ro_created(const char * name,
                 rib.dtc.max_pdu_size = stat_msg->max_pdu_size;
                 rib.addr_auth_type = stat_msg->addr_auth_type;
 
-                if (frct_init()) {
-                        ipcp_set_state(IPCP_INIT);
-                        pthread_rwlock_unlock(&ipcpi.state_lock);
-                        static_info_msg__free_unpacked(stat_msg, NULL);
-                        LOG_ERR("Failed to init FRCT");
-                        return;
-                }
-
                 static_info_msg__free_unpacked(stat_msg, NULL);
         }
-        pthread_rwlock_unlock(&ipcpi.state_lock);
 }
 
 /* We only have a create operation for now. */
@@ -185,7 +184,6 @@ static struct rnode * find_rnode_by_name(const char * name)
 {
         char * str;
         char * str1;
-        char * saveptr;
         char * token;
         struct rnode * node;
 
@@ -195,8 +193,8 @@ static struct rnode * find_rnode_by_name(const char * name)
 
         node = rib.root;
 
-        for (str1 = str; ; str1 = NULL) {
-                token = strtok_r(str1, PATH_DELIMITER, &saveptr);
+        for (str1 = str; node != NULL; str1 = NULL) {
+                token = strtok(str1, PATH_DELIMITER);
                 if (token == NULL)
                         break;
 
@@ -207,18 +205,13 @@ static struct rnode * find_rnode_by_name(const char * name)
                                 break;
                         else
                                 node = node->sibling;
-
-                if (node == NULL) {
-                        free(str);
-                        return NULL;
-                }
         }
 
         free(str);
         return node;
 }
 
-/* Call under RIB object lock */
+/* Call under RIB object lock. */
 static int ro_msg_create(struct rnode * node,
                          ro_msg_t *     msg)
 {
@@ -334,7 +327,7 @@ static struct rnode * ribmgr_ro_create(const char *   name,
 
         node = rib.root;
 
-        for (str1 = str; ; str1 = NULL) {
+        for (str1 = str; node != NULL; str1 = NULL) {
                 token = strtok_r(str1, PATH_DELIMITER, &saveptr);
                 if (token == NULL) {
                         LOG_ERR("RO already exists.");
@@ -356,9 +349,6 @@ static struct rnode * ribmgr_ro_create(const char *   name,
                                 sibling = true;
                         }
                 }
-
-                if (node == NULL)
-                        break;
         }
 
         token2 = strtok_r(NULL, PATH_DELIMITER, &saveptr);
@@ -554,9 +544,88 @@ int ribmgr_init()
                 return -1;
         }
 
+        if (pthread_cond_init(&rib.state_cond, NULL)) {
+                LOG_ERR("Failed to init condvar.");
+                timerwheel_destroy(rib.wheel);
+                bmp_destroy(rib.sids);
+                pthread_rwlock_destroy(&rib.flows_lock);
+                pthread_mutex_destroy(&rib.ro_lock);
+                pthread_mutex_destroy(&rib.subs_lock);
+                pthread_mutex_destroy(&rib.ro_ids_lock);
+                free(rib.root);
+                return -1;
+        }
+
+        if (pthread_mutex_init(&rib.state_lock, NULL)) {
+                LOG_ERR("Failed to init mutex.");
+                pthread_cond_destroy(&rib.state_cond);
+                timerwheel_destroy(rib.wheel);
+                bmp_destroy(rib.sids);
+                pthread_rwlock_destroy(&rib.flows_lock);
+                pthread_mutex_destroy(&rib.ro_lock);
+                pthread_mutex_destroy(&rib.subs_lock);
+                pthread_mutex_destroy(&rib.ro_ids_lock);
+                free(rib.root);
+                return -1;
+        }
+
+        rib.state = RIBMGR_INIT;
+
         return 0;
 }
 
+static enum ribmgr_state ribmgr_get_state(void)
+{
+        enum ribmgr_state state;
+
+        pthread_mutex_lock(&rib.state_lock);
+
+        state = rib.state;
+
+        pthread_mutex_unlock(&rib.state_lock);
+
+        return state;
+}
+
+static void ribmgr_set_state(enum ribmgr_state state)
+{
+        pthread_mutex_lock(&rib.state_lock);
+
+        rib.state = state;
+
+        pthread_cond_broadcast(&rib.state_cond);
+
+        pthread_mutex_unlock(&rib.state_lock);
+}
+
+static int ribmgr_wait_state(enum ribmgr_state       state,
+                             const struct timespec * timeout)
+{
+        struct timespec abstime;
+        int ret = 0;
+
+        clock_gettime(PTHREAD_COND_CLOCK, &abstime);
+        ts_add(&abstime, timeout, &abstime);
+
+        pthread_mutex_lock(&rib.state_lock);
+
+        while (rib.state != state
+               && rib.state != RIBMGR_SHUTDOWN
+               && rib.state != RIBMGR_NULL) {
+                if (timeout == NULL)
+                        ret = -pthread_cond_wait(&rib.state_cond,
+                                                 &rib.state_lock);
+                else
+                        ret = -pthread_cond_timedwait(&rib.state_cond,
+                                                      &rib.state_lock,
+                                                      &abstime);
+        }
+
+        pthread_mutex_unlock(&rib.state_lock);
+
+        return ret;
+}
+
 static void rtree_destroy(struct rnode * node)
 {
         if (node != NULL) {
@@ -574,7 +643,13 @@ int ribmgr_fini()
         struct list_head * pos = NULL;
         struct list_head * n = NULL;
 
+        pthread_mutex_lock(&rib.state_lock);
+        rib.state = RIBMGR_SHUTDOWN;
+        pthread_cond_broadcast(&rib.state_cond);
+        pthread_mutex_unlock(&rib.state_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);
@@ -583,6 +658,7 @@ int ribmgr_fini()
                 list_del(&flow->next);
                 free(flow);
         }
+
         pthread_rwlock_unlock(&rib.flows_lock);
 
         ro_unsubscribe(rib.ribmgr_sid);
@@ -591,8 +667,10 @@ int ribmgr_fini()
                 addr_auth_destroy(rib.addr_auth);
 
         pthread_mutex_lock(&rib.ro_lock);
+
         rtree_destroy(rib.root->child);
         free(rib.root);
+
         pthread_mutex_unlock(&rib.ro_lock);
 
         bmp_destroy(rib.sids);
@@ -603,6 +681,9 @@ int ribmgr_fini()
         pthread_rwlock_destroy(&rib.flows_lock);
         pthread_mutex_destroy(&rib.ro_ids_lock);
 
+        pthread_cond_destroy(&rib.state_cond);
+        pthread_mutex_destroy(&rib.state_lock);
+
         return 0;
 }
 
@@ -827,16 +908,7 @@ static int ribmgr_cdap_start(struct cdap * instance,
         if (strcmp(name, ENROLLMENT) == 0) {
                 LOG_DBG("New enrollment request.");
 
-                pthread_rwlock_wrlock(&ipcpi.state_lock);
-
-                if (ipcp_get_state() != IPCP_OPERATIONAL) {
-                        pthread_rwlock_unlock(&ipcpi.state_lock);
-                        LOG_ERR("IPCP in wrong state.");
-                        return -1;
-                }
-
                 if (cdap_reply_send(instance, key, 0, NULL, 0)) {
-                        pthread_rwlock_unlock(&ipcpi.state_lock);
                         LOG_ERR("Failed to send reply to enrollment request.");
                         return -1;
                 }
@@ -847,7 +919,6 @@ static int ribmgr_cdap_start(struct cdap * instance,
                 pthread_mutex_lock(&rib.ro_lock);
                 if (ribmgr_enrol_sync(instance, rib.root->child)) {
                         pthread_mutex_unlock(&rib.ro_lock);
-                        pthread_rwlock_unlock(&ipcpi.state_lock);
                         LOG_ERR("Failed to sync part of the RIB.");
                         return -1;
                 }
@@ -859,18 +930,14 @@ static int ribmgr_cdap_start(struct cdap * instance,
                 key = cdap_request_send(instance, CDAP_STOP, ENROLLMENT,
                                         NULL, 0, 0);
                 if (key < 0) {
-                        pthread_rwlock_unlock(&ipcpi.state_lock);
                         LOG_ERR("Failed to send stop of enrollment.");
                         return -1;
                 }
 
                 if (cdap_reply_wait(instance, key, NULL, NULL)) {
-                        pthread_rwlock_unlock(&ipcpi.state_lock);
                         LOG_ERR("Remote failed to complete enrollment.");
                         return -1;
                 }
-
-                pthread_rwlock_unlock(&ipcpi.state_lock);
         } else {
                 LOG_WARN("Request to start unknown operation.");
                 if (cdap_reply_send(instance, key, -1, NULL, 0))
@@ -884,20 +951,18 @@ static int ribmgr_cdap_stop(struct cdap * instance, cdap_key_t key, char * name)
 {
         int ret = 0;
 
-        pthread_rwlock_wrlock(&ipcpi.state_lock);
-        if (ipcp_get_state() == IPCP_CONFIG && strcmp(name, ENROLLMENT) == 0) {
+        if (strcmp(name, ENROLLMENT) == 0) {
                 LOG_DBG("Stop enrollment received.");
-
-                ipcp_set_state(IPCP_BOOTING);
-        } else
+                /* FIXME: don't use states to match start to stop. */
+                ribmgr_set_state(RIBMGR_OPERATIONAL);
+        } else {
                 ret = -1;
+        }
 
         if (cdap_reply_send(instance, key, ret, NULL, 0)) {
-                pthread_rwlock_unlock(&ipcpi.state_lock);
                 LOG_ERR("Failed to send reply to stop request.");
                 return -1;
         }
-        pthread_rwlock_unlock(&ipcpi.state_lock);
 
         return 0;
 }
@@ -1068,7 +1133,7 @@ static void * cdap_req_handler(void * o)
         return (void *) 0;
 }
 
-int ribmgr_add_flow(int fd)
+static int ribmgr_add_flow(int fd)
 {
         struct cdap * instance = NULL;
         struct mgmt_flow * flow;
@@ -1127,6 +1192,46 @@ int ribmgr_remove_flow(int fd)
         return -1;
 }
 
+/* FIXME: do this in a topologymanager instance */
+int ribmgr_add_nm1_flow(int fd)
+{
+        if (flow_alloc_resp(fd, 0) < 0) {
+                LOG_ERR("Could not respond to new flow.");
+                return -1;
+        }
+
+        return ribmgr_add_flow(fd);
+}
+
+int ribmgr_nm1_mgt_flow(char * dst_name)
+{
+        int fd;
+        int result;
+
+        /* FIXME: Request retransmission. */
+        fd = flow_alloc(dst_name, MGMT_AE, NULL);
+        if (fd < 0) {
+                LOG_ERR("Failed to allocate flow to %s.", dst_name);
+                return -1;
+        }
+
+        result = flow_alloc_res(fd);
+        if (result < 0) {
+                LOG_ERR("Result of flow allocation to %s is %d.",
+                        dst_name, result);
+                flow_dealloc(fd);
+                return -1;
+        }
+
+        if (ribmgr_add_flow(fd)) {
+                LOG_ERR("Failed to add file descriptor.");
+                flow_dealloc(fd);
+                return -1;
+        }
+
+        return fd;
+}
+
 int ribmgr_bootstrap(struct dif_config * conf)
 {
         static_info_msg_t stat_info = STATIC_INFO_MSG__INIT;
@@ -1134,11 +1239,6 @@ int ribmgr_bootstrap(struct dif_config * conf)
         size_t len = 0;
         struct ro_attr attr;
 
-        if (conf == NULL || conf->type != IPCP_NORMAL) {
-                LOG_ERR("Bad DIF configuration.");
-                return -EINVAL;
-        }
-
         ro_attr_init(&attr);
         attr.enrol_sync = true;
 
@@ -1189,83 +1289,54 @@ int ribmgr_bootstrap(struct dif_config * conf)
                 return -1;
         }
 
-        if (frct_init()) {
-                LOG_ERR("Failed to initialize FRCT.");
-                dir_fini();
-                ribmgr_ro_delete(RIBMGR_PREFIX STAT_INFO);
-                ribmgr_ro_delete(RIBMGR_PREFIX);
-                return -1;
-        }
-
         LOG_DBG("Bootstrapped RIB Manager.");
 
         return 0;
 }
 
-int ribmgr_enrol(void)
+int ribmgr_enrol()
 {
         struct cdap * instance = NULL;
         struct mgmt_flow * flow;
         cdap_key_t key;
         int ret;
-
-        pthread_rwlock_wrlock(&ipcpi.state_lock);
-
-        if (ipcp_get_state() != IPCP_INIT) {
-                pthread_rwlock_unlock(&ipcpi.state_lock);
-                LOG_ERR("IPCP in wrong state.");
-                return -1;
-        }
-
-        ipcp_set_state(IPCP_CONFIG);
+        struct timespec timeout = {(ENROLL_TIMEOUT / 1000),
+                                   (ENROLL_TIMEOUT % 1000) * MILLION};
 
         pthread_rwlock_wrlock(&rib.flows_lock);
-        if (list_empty(&rib.flows)) {
-                ipcp_set_state(IPCP_INIT);
-                pthread_rwlock_unlock(&rib.flows_lock);
-                pthread_rwlock_unlock(&ipcpi.state_lock);
-                LOG_ERR("No flows in RIB.");
-                return -1;
-        }
+
+        assert(!list_empty(&rib.flows));
 
         flow = list_first_entry((&rib.flows), struct mgmt_flow, next);
         instance = flow->instance;
 
         key = cdap_request_send(instance, CDAP_START, ENROLLMENT, NULL, 0, 0);
         if (key < 0) {
-                ipcp_set_state(IPCP_INIT);
                 pthread_rwlock_unlock(&rib.flows_lock);
-                pthread_rwlock_unlock(&ipcpi.state_lock);
                 LOG_ERR("Failed to start enrollment.");
                 return -1;
         }
 
         ret = cdap_reply_wait(instance, key, NULL, NULL);
         if (ret) {
-                ipcp_set_state(IPCP_INIT);
                 pthread_rwlock_unlock(&rib.flows_lock);
-                pthread_rwlock_unlock(&ipcpi.state_lock);
                 LOG_ERR("Failed to enroll: %d.", ret);
                 return -1;
         }
 
         pthread_rwlock_unlock(&rib.flows_lock);
-        pthread_rwlock_unlock(&ipcpi.state_lock);
+
+        if (ribmgr_wait_state(RIBMGR_OPERATIONAL, &timeout) == -ETIMEDOUT)
+                LOG_ERR("Enrollment of RIB timed out.");
+
+        if (ribmgr_get_state() != RIBMGR_OPERATIONAL)
+                return -1;
 
         return 0;
 }
 
 int ribmgr_start_policies(void)
 {
-        pthread_rwlock_rdlock(&ipcpi.state_lock);
-
-        if (ipcp_get_state() != IPCP_BOOTING) {
-                pthread_rwlock_unlock(&ipcpi.state_lock);
-                LOG_ERR("Cannot start policies in wrong state.");
-                return -1;
-        }
-        pthread_rwlock_unlock(&ipcpi.state_lock);
-
         rib.addr_auth = addr_auth_create(rib.addr_auth_type);
         if (rib.addr_auth == NULL) {
                 LOG_ERR("Failed to create address authority.");