path: root/src/ipcpd/normal/ribmgr.c

/*
 * Ouroboros - Copyright (C) 2016
 *
 * RIB manager of the 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 "rib-manager"

#include <ouroboros/config.h>
#include <ouroboros/logs.h>
#include <ouroboros/cdap.h>
#include <ouroboros/list.h>
#include <ouroboros/time_utils.h>
#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 "addr_auth.h"
#include "ribmgr.h"
#include "dt_const.h"
#include "frct.h"
#include "ipcp.h"
#include "cdap_request.h"
#include "ro.h"

#include "static_info.pb-c.h"
typedef StaticInfoMsg static_info_msg_t;

#include "ro.pb-c.h"
typedef RoMsg ro_msg_t;

#define SUBS_SIZE 25

#define ENROLLMENT     "enrollment"

#define RIBMGR_PREFIX  "/ribmgr"
#define STAT_INFO      "/statinfo"
#define PATH_DELIMITER "/"

/* RIB objects */
struct rnode {
        char *            name;
        char *            full_name;
        uint64_t          seqno;

        /*
         * NOTE: Naive implementation for now, could be replaced by
         * for instance taking a hash of the pathname and using that
         * as an index in a B-tree
         */

        /* If there are no children, this is a leaf */
        struct rnode *    child;
        struct rnode *    sibling;

        struct ro_props * props;
        uint8_t *         data;
        size_t            len;
};

struct mgmt_flow {
        struct cdap *    instance;
        int              fd;
        struct list_head next;
};

struct ro_sub {
        int                 sid;
        char *              name;
        struct ro_sub_ops * ops;
        struct list_head    next;
};

struct {
        struct rnode *     root;
        pthread_mutex_t    ro_lock;

        struct list_head   subs;
        struct bmp *       sids;
        pthread_mutex_t    subs_lock;
        int                ribmgr_sid;

        struct dt_const    dtc;

        uint64_t           address;

        struct list_head   flows;
        pthread_rwlock_t   flows_lock;

        struct list_head   cdap_reqs;
        pthread_mutex_t    cdap_reqs_lock;

        struct addr_auth * addr_auth;
} rib;

int ribmgr_ro_created(const char * name,
                      uint8_t *    data,
                      size_t       len)
{
        static_info_msg_t * stat_msg;

        pthread_rwlock_wrlock(&ipcpi.state_lock);
        if (ipcp_get_state() == IPCP_PENDING_ENROLL &&
            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 -1;
                }

                rib.dtc.addr_size = stat_msg->addr_size;
                rib.dtc.cep_id_size = stat_msg->cep_id_size;
                rib.dtc.pdu_length_size = stat_msg->pdu_length_size;
                rib.dtc.seqno_size = stat_msg->seqno_size;
                rib.dtc.has_ttl = stat_msg->has_ttl;
                rib.dtc.has_chk = stat_msg->has_chk;
                rib.dtc.min_pdu_size = stat_msg->min_pdu_size;
                rib.dtc.max_pdu_size = stat_msg->max_pdu_size;

                rib.addr_auth = addr_auth_create(stat_msg->addr_auth_type);
                if (rib.addr_auth == NULL) {
                        ipcp_set_state(IPCP_INIT);
                        pthread_rwlock_unlock(&ipcpi.state_lock);
                        static_info_msg__free_unpacked(stat_msg, NULL);
                        LOG_ERR("Failed to create address authority");
                        return -1;
                }

                rib.address = rib.addr_auth->address();
                LOG_DBG("IPCP has address %lu", rib.address);

                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 -1;
                }

                static_info_msg__free_unpacked(stat_msg, NULL);
        }
        pthread_rwlock_unlock(&ipcpi.state_lock);

        return 0;
}

/* We only have a create operation for now */
static struct ro_sub_ops ribmgr_sub_ops = {
        .ro_created = ribmgr_ro_created,
        .ro_updated = NULL,
        .ro_deleted = NULL
};

static struct rnode * find_rnode_by_name(const char * name)
{
        char * str;
        char * str1;
        char * saveptr;
        char * token;
        struct rnode * node;

        str = strdup(name);
        if (str == NULL)
                return NULL;

        node = rib.root;

        for (str1 = str; ; str1 = NULL) {
                token = strtok_r(str1, PATH_DELIMITER, &saveptr);
                if (token == NULL)
                        break;

                node = node->child;

                while (node != NULL)
                        if (strcmp(node->name, token) == 0)
                                break;
                        else
                                node = node->sibling;

                if (node == NULL) {
                        free(str);
                        return NULL;
                }
        }

        free(str);
        return node;
}

/* Call under RIB object lock */
static int ro_msg_create(struct rnode * node,
                         ro_msg_t *     msg)
{
        msg->address = rib.address;
        msg->seqno = node->seqno;
        msg->recv_set = node->props->recv_set;
        msg->enrol_sync = node->props->enrol_sync;
        msg->sec = node->props->expiry.tv_sec;
        msg->nsec = node->props->expiry.tv_nsec;
        msg->value.data = node->data;
        msg->value.len = node->len;

        return 0;
}

static struct rnode * ribmgr_ro_create(const char *      name,
                                       struct ro_props * props,
                                       uint8_t *         data,
                                       size_t            len)
{
        char * str;
        char * str1;
        char * saveptr;
        char * token;
        char * token2;
        struct rnode * node;
        struct rnode * new;
        struct rnode * prev;
        bool sibling;

        str = strdup(name);
        if (str == NULL)
                return NULL;

        node = rib.root;

        for (str1 = str; ; str1 = NULL) {
                token = strtok_r(str1, PATH_DELIMITER, &saveptr);
                if (token == NULL) {
                        LOG_ERR("RO already exists.");
                        free(str);
                        return NULL;
                }

                prev = node;
                node = node->child;
                sibling = false;

                /* Search horizontally */
                while (node != NULL) {
                        if (strcmp(node->name, token) == 0) {
                                break;
                        } else {
                                prev = node;
                                node = node->sibling;
                                sibling = true;
                        }
                }

                if (node == NULL)
                        break;
        }

        token2 = strtok_r(NULL, PATH_DELIMITER, &saveptr);
        if (token2 != NULL) {
                LOG_ERR("Part of the pathname does not exist.");
                free(str);
                return NULL;
        }

        new = malloc(sizeof(*new));
        if (new == NULL) {
                free(str);
                return NULL;
        }

        new->name = strdup(token);
        if (new->name == NULL) {
                free(str);
                free(new);
                return NULL;
        }

        free(str);

        new->full_name = strdup(name);
        if (new->full_name == NULL) {
                free(new);
                return NULL;
        }

        new->seqno = 0;
        new->props = props;

        if (sibling)
                prev->sibling = new;
        else
                prev->child = new;

        new->data = data;
        new->len = len;
        new->child = NULL;
        new->sibling = NULL;

        return new;
}

static int ribmgr_ro_delete(const char * name)
{
        char * str;
        char * str1;
        char * saveptr;
        char * token;
        struct rnode * node;
        struct rnode * prev;
        bool sibling = false;

        str = strdup(name);
        if (str == NULL)
                return -1;

        node = rib.root;
        prev = NULL;

        for (str1 = str; ; str1 = NULL) {
                token = strtok_r(str1, PATH_DELIMITER, &saveptr);
                if (token == NULL)
                        break;

                prev = node;
                node = node->child;
                sibling = false;

                while (node != NULL) {
                        if (strcmp(node->name, token) == 0) {
                                break;
                        } else {
                                prev = node;
                                node = node->sibling;
                                sibling = true;
                        }
                }

                if (node == NULL) {
                        free(str);
                        return -1;
                }
        }

        if (node == rib.root) {
                LOG_ERR("Won't remove root.");
                free(str);
                return -1;
        }

        free(node->name);
        free(node->full_name);
        if (node->data != NULL)
                free(node->data);

        if (sibling)
                prev->sibling = node->sibling;
        else
                prev->child = node->sibling;

        free(node);
        free(str);

        return 0;
}

static struct rnode * ribmgr_ro_write(const char * name,
                                      uint8_t *    data,
                                      size_t       len)
{
        struct rnode * node;

        node = find_rnode_by_name(name);
        if (node == NULL)
                return NULL;

        free(node->data);

        node->data = data;
        node->len = len;
        node->seqno++;

        return node;
}

/* Call while holding cdap_reqs_lock */
/* FIXME: better not to call blocking functions under any lock */
int cdap_result_wait(struct cdap * instance,
                     enum cdap_opcode code,
                     char * name,
                     int invoke_id)
{
        struct cdap_request * req;
        int ret;
        char * name_dup = strdup(name);
        if (name_dup == NULL)
                return -1;

        req = cdap_request_create(code, name_dup, invoke_id, instance);
        if (req == NULL) {
                free(name_dup);
                return -1;
        }

        list_add(&req->next, &rib.cdap_reqs);

        pthread_mutex_unlock(&rib.cdap_reqs_lock);

        ret = cdap_request_wait(req);

        pthread_mutex_lock(&rib.cdap_reqs_lock);

        if (ret == -1)  /* should only be on ipcp shutdown */
                LOG_DBG("Waiting CDAP request destroyed.");

        if (ret == -ETIMEDOUT)
                LOG_ERR("CDAP Request timed out.");

        if (ret)
                LOG_DBG("Unknown error code: %d.", ret);

        if (!ret)
                ret = req->result;

        list_del(&req->next);
        cdap_request_destroy(req);

        return ret;
}

static int write_ro_msg(struct cdap *    neighbor,
                        ro_msg_t *       msg,
                        char *           name,
                        enum cdap_opcode code)
{
        uint8_t * data;
        size_t len;
        int iid = 0;

        len = ro_msg__get_packed_size(msg);
        if (len == 0)
                return -1;

        data = malloc(len);
        if (data == NULL)
                return -ENOMEM;

        ro_msg__pack(msg, data);

        pthread_mutex_lock(&rib.cdap_reqs_lock);
        iid = cdap_send_request(neighbor, code,
                                name, data, len, 0);
        if (iid < 0) {
                pthread_mutex_unlock(&rib.cdap_reqs_lock);
                free(data);
                return -1;
        }

        if (cdap_result_wait(neighbor, code, name, iid)) {
                pthread_mutex_unlock(&rib.cdap_reqs_lock);
                free(data);
                LOG_ERR("Remote did not receive RIB object.");
                return -1;
        }
        pthread_mutex_unlock(&rib.cdap_reqs_lock);
        free(data);

        return 0;
}

int ribmgr_init()
{
        INIT_LIST_HEAD(&rib.flows);
        INIT_LIST_HEAD(&rib.cdap_reqs);
        INIT_LIST_HEAD(&rib.subs);

        rib.root = malloc(sizeof(*(rib.root)));
        if (rib.root == NULL)
                return -1;

        rib.root->name = "root";
        rib.root->child = NULL;
        rib.root->sibling = NULL;

        if (pthread_rwlock_init(&rib.flows_lock, NULL)) {
                LOG_ERR("Failed to initialize rwlock.");
                free(rib.root);
                return -1;
        }

        if (pthread_mutex_init(&rib.cdap_reqs_lock, NULL)) {
                LOG_ERR("Failed to initialize mutex.");
                pthread_rwlock_destroy(&rib.flows_lock);
                free(rib.root);
                return -1;
        }

        if (pthread_mutex_init(&rib.ro_lock, NULL)) {
                LOG_ERR("Failed to initialize mutex.");
                pthread_rwlock_destroy(&rib.flows_lock);
                pthread_mutex_destroy(&rib.cdap_reqs_lock);
                free(rib.root);
                return -1;
        }

        if (pthread_mutex_init(&rib.subs_lock, NULL)) {
                LOG_ERR("Failed to initialize mutex.");
                pthread_rwlock_destroy(&rib.flows_lock);
                pthread_mutex_destroy(&rib.cdap_reqs_lock);
                pthread_mutex_destroy(&rib.ro_lock);
                free(rib.root);
                return -1;
        }

        rib.sids = bmp_create(SUBS_SIZE, 0);
        if (rib.sids == NULL) {
                LOG_ERR("Failed to create bitmap.");
                pthread_rwlock_destroy(&rib.flows_lock);
                pthread_mutex_destroy(&rib.cdap_reqs_lock);
                pthread_mutex_destroy(&rib.ro_lock);
                pthread_mutex_destroy(&rib.subs_lock);
                free(rib.root);
                return -1;
        }

        rib.ribmgr_sid = ro_subscribe(RIBMGR_PREFIX, &ribmgr_sub_ops);
        if (rib.ribmgr_sid < 0) {
                LOG_ERR("Failed to subscribe.");
                bmp_destroy(rib.sids);
                pthread_rwlock_destroy(&rib.flows_lock);
                pthread_mutex_destroy(&rib.cdap_reqs_lock);
                pthread_mutex_destroy(&rib.ro_lock);
                pthread_mutex_destroy(&rib.subs_lock);
                free(rib.root);
                return -1;
        }

        return 0;
}

static void rtree_destroy(struct rnode * node)
{
        if (node != NULL) {
                rtree_destroy(node->child);
                rtree_destroy(node->sibling);
                free(node->name);
                if (node->data != NULL)
                        free(node->data);
                free(node);
        }
}

int ribmgr_fini()
{
        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);

        ro_unsubscribe(rib.ribmgr_sid);

        if (rib.addr_auth != NULL)
                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);

        pthread_mutex_destroy(&rib.subs_lock);
        pthread_mutex_destroy(&rib.cdap_reqs_lock);
        pthread_mutex_destroy(&rib.ro_lock);
        pthread_rwlock_destroy(&rib.flows_lock);

        return 0;
}

static int ribmgr_cdap_reply(struct cdap * instance,
                             int           invoke_id,
                             int           result,
                             uint8_t *     data,
                             size_t        len)
{
        struct list_head * pos, * n = NULL;

        /* We never perform reads on other RIBs */
        (void) data;
        (void) len;

        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);
                if (req->instance == instance &&
                    req->invoke_id == invoke_id &&
                    req->state == REQ_PENDING) {
                        if (result != 0)
                                LOG_ERR("CDAP command with code %d and name %s "
                                        "failed with error %d",
                                        req->code, req->name, result);
                        else
                                LOG_DBG("CDAP command with code %d and name %s "
                                        "executed succesfully",
                                        req->code, req->name);

                        pthread_mutex_unlock(&rib.cdap_reqs_lock);

                        cdap_request_respond(req, result);

                        pthread_mutex_lock(&rib.cdap_reqs_lock);
                }
        }
        pthread_mutex_unlock(&rib.cdap_reqs_lock);

        return 0;
}

static int ribmgr_cdap_create(struct cdap * instance,
                              int           invoke_id,
                              char *        name,
                              ro_msg_t *    msg)
{
        int ret = 0;
        struct list_head * p = NULL;
        size_t len_s, len_n;
        uint8_t * ro_data;
        struct ro_props * props;
        struct rnode * node;

        props = malloc(sizeof(*props));
        if (props == NULL) {
                cdap_send_reply(instance, invoke_id, -1, NULL, 0);
                return -ENOMEM;
        }

        props->expiry.tv_sec = msg->sec;
        props->expiry.tv_nsec = msg->nsec;
        props->enrol_sync = msg->enrol_sync;

        pthread_mutex_lock(&rib.ro_lock);

        ro_data = malloc(msg->value.len);
        if (ro_data == NULL) {
                pthread_mutex_unlock(&rib.ro_lock);
                cdap_send_reply(instance, invoke_id, -1, NULL, 0);
                free(props);
                return -1;
        }
        memcpy(ro_data, msg->value.data, msg->value.len);

        node = ribmgr_ro_create(name, props, ro_data, msg->value.len);
        if (node == NULL) {
                pthread_mutex_unlock(&rib.ro_lock);
                cdap_send_reply(instance, invoke_id, -1, NULL, 0);
                free(props);
                free(ro_data);
                return -1;
        }

        pthread_mutex_lock(&rib.subs_lock);
        list_for_each(p, &rib.subs) {
                struct ro_sub * e = list_entry(p, struct ro_sub, next);
                len_s = strlen(e->name);
                len_n = strlen(name);

                if (len_n < len_s)
                        continue;

                if (memcmp(name, e->name, len_s) == 0) {
                        if (e->ops->ro_created == NULL)
                                continue;

                        ro_data = malloc(node->len);
                        if (ro_data == NULL)
                                continue;

                        memcpy(ro_data, node->data, node->len);
                        e->ops->ro_created(name, ro_data, node->len);
                }
        }

        pthread_mutex_unlock(&rib.subs_lock);
        pthread_mutex_unlock(&rib.ro_lock);

        if (cdap_send_reply(instance, invoke_id, ret, NULL, 0)) {
                LOG_ERR("Failed to send reply to create request.");
                return -1;
        }

        return 0;
}

static int ribmgr_cdap_delete(struct cdap * instance,
                              int           invoke_id,
                              char *        name)
{
        struct list_head * p = NULL;
        size_t len_s;
        size_t len_n;

        pthread_mutex_lock(&rib.ro_lock);

        if (ribmgr_ro_delete(name)) {
                pthread_mutex_unlock(&rib.ro_lock);
                cdap_send_reply(instance, invoke_id, -1, NULL, 0);
                return -1;
        }

        pthread_mutex_lock(&rib.subs_lock);

        list_for_each(p, &rib.subs) {
                struct ro_sub * e = list_entry(p, struct ro_sub, next);
                len_s = strlen(e->name);
                len_n = strlen(name);

                if (len_n < len_s)
                        continue;

                if (memcmp(name, e->name, len_s) == 0) {
                        if (e->ops->ro_deleted == NULL)
                                continue;

                        e->ops->ro_deleted(name);
                }
        }

        pthread_mutex_unlock(&rib.subs_lock);
        pthread_mutex_unlock(&rib.ro_lock);

        if (cdap_send_reply(instance, invoke_id, 0, NULL, 0)) {
                LOG_ERR("Failed to send reply to create request.");
                return -1;
        }

        return 0;
}

static int ribmgr_cdap_write(struct cdap * instance,
                             int           invoke_id,
                             char *        name,
                             ro_msg_t *    msg,
                             uint32_t      flags)
{
        int ret = 0;
        struct list_head * p = NULL;
        size_t len_s;
        size_t len_n;
        uint8_t * ro_data;
        struct rnode * node;

        (void) flags;

        pthread_mutex_lock(&rib.ro_lock);

        ro_data = malloc(msg->value.len);
        if (ro_data == NULL) {
                pthread_mutex_unlock(&rib.ro_lock);
                cdap_send_reply(instance, invoke_id, -1, NULL, 0);
                return -1;
        }
        memcpy(ro_data, msg->value.data, msg->value.len);

        node = ribmgr_ro_write(name, msg->value.data, msg->value.len);
        if (node == NULL) {
                pthread_mutex_unlock(&rib.ro_lock);
                free(ro_data);
                cdap_send_reply(instance, invoke_id, -1, NULL, 0);
                return -1;
        }

        pthread_mutex_lock(&rib.subs_lock);

        list_for_each(p, &rib.subs) {
                struct ro_sub * e = list_entry(p, struct ro_sub, next);
                len_s = strlen(e->name);
                len_n = strlen(name);

                if (len_n < len_s)
                        continue;

                if (memcmp(name, e->name, len_s) == 0) {
                        if (e->ops->ro_updated == NULL)
                                continue;

                        ro_data = malloc(node->len);
                        if (ro_data == NULL)
                                continue;

                        memcpy(ro_data, node->data, node->len);
                        e->ops->ro_updated(name, ro_data, node->len);
                }
        }

        pthread_mutex_unlock(&rib.subs_lock);
        pthread_mutex_unlock(&rib.ro_lock);

        if (cdap_send_reply(instance, invoke_id, ret, NULL, 0)) {
                LOG_ERR("Failed to send reply to write request.");
                return -1;
        }

        return 0;
}

static int ribmgr_enrol_sync(struct cdap * instance,
                             struct rnode * node)
{
        int ret = 0;

        if (node != NULL) {
                if (node->props->enrol_sync == true) {
                        ro_msg_t msg = RO_MSG__INIT;

                        if (ro_msg_create(node, &msg)) {
                                LOG_ERR("Failed to create RO msg.");
                                return -1;
                        }

                        if (write_ro_msg(instance, &msg,
                                         node->full_name, CDAP_CREATE)) {
                                LOG_ERR("Failed to send RO msg.");
                                return -1;
                        }
                }

                ret = ribmgr_enrol_sync(instance, node->child);
                if (ret == 0)
                        ret = ribmgr_enrol_sync(instance, node->sibling);
        }

        return ret;
}

static int ribmgr_cdap_start(struct cdap * instance,
                             int           invoke_id,
                             char *        name)
{
        int iid = 0;

        pthread_rwlock_wrlock(&ipcpi.state_lock);
        if (ipcp_get_state() == IPCP_ENROLLED &&
            strcmp(name, ENROLLMENT) == 0) {
                LOG_DBG("New enrollment request.");

                if (cdap_send_reply(instance, invoke_id, 0, NULL, 0)) {
                        pthread_rwlock_unlock(&ipcpi.state_lock);
                        LOG_ERR("Failed to send reply to enrollment request.");
                        return -1;
                }

                /* Loop through rtree and send correct objects */
                LOG_DBGF("Sending ROs that need to be sent on enrolment...");

                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;
                }
                pthread_mutex_unlock(&rib.ro_lock);

                LOG_DBGF("Sending stop enrollment...");

                pthread_mutex_lock(&rib.cdap_reqs_lock);

                iid = cdap_send_request(instance, CDAP_STOP, ENROLLMENT,
                                        NULL, 0, 0);
                if (iid < 0) {
                        pthread_mutex_unlock(&rib.cdap_reqs_lock);
                        pthread_rwlock_unlock(&ipcpi.state_lock);
                        LOG_ERR("Failed to send stop of enrollment.");
                        return -1;
                }

                if (cdap_result_wait(instance, CDAP_STOP,
                                     ENROLLMENT, iid)) {
                        pthread_mutex_unlock(&rib.cdap_reqs_lock);
                        pthread_rwlock_unlock(&ipcpi.state_lock);
                        LOG_ERR("Remote failed to complete enrollment.");
                        return -1;
                }
                pthread_mutex_unlock(&rib.cdap_reqs_lock);
        } else {
                if (cdap_send_reply(instance, invoke_id, -1, NULL, 0)) {
                        pthread_rwlock_unlock(&ipcpi.state_lock);
                        LOG_ERR("Failed to send reply to start request.");
                        return -1;
                }
        }
        pthread_rwlock_unlock(&ipcpi.state_lock);

        return 0;
}

static int ribmgr_cdap_stop(struct cdap * instance,
                            int           invoke_id,
                            char *        name)
{
        int ret = 0;

        pthread_rwlock_wrlock(&ipcpi.state_lock);
        if (ipcp_get_state() == IPCP_PENDING_ENROLL &&
            strcmp(name, ENROLLMENT) == 0) {
                LOG_DBG("Stop enrollment received.");

                ipcp_set_state(IPCP_ENROLLED);
        } else
                ret = -1;

        if (cdap_send_reply(instance, invoke_id, 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;
}

static int ribmgr_cdap_request(struct cdap *    instance,
                               int              invoke_id,
                               enum cdap_opcode opcode,
                               char *           name,
                               uint8_t *        data,
                               size_t           len,
                               uint32_t         flags)
{
        ro_msg_t * msg;
        int ret = -1;
        struct list_head * p = NULL;

        if (opcode == CDAP_START)
                return ribmgr_cdap_start(instance,
                                         invoke_id,
                                         name);
        else if (opcode == CDAP_STOP)
                return ribmgr_cdap_stop(instance,
                                        invoke_id,
                                        name);

        msg = ro_msg__unpack(NULL, len, data);
        if (msg == NULL) {
                cdap_send_reply(instance, invoke_id, -1, NULL, 0);
                LOG_ERR("Failed to unpack RO message");
                return -1;
        }

        /* FIXME: Check if we already received this */

        if (opcode == CDAP_CREATE) {
                ret = ribmgr_cdap_create(instance,
                                         invoke_id,
                                         name,
                                         msg);
        } else if (opcode == CDAP_WRITE) {
                ret = ribmgr_cdap_write(instance,
                                        invoke_id,
                                        name, msg,
                                        flags);

        } else if (opcode == CDAP_DELETE) {
                ret = ribmgr_cdap_delete(instance,
                                         invoke_id,
                                         name);
        } else {
                LOG_INFO("Unsupported opcode received.");
                cdap_send_reply(instance, invoke_id, -1, NULL, 0);
                return -1;
        }

        if (msg->recv_set == ALL_MEMBERS) {
                pthread_rwlock_rdlock(&rib.flows_lock);
                list_for_each(p, &rib.flows) {
                        struct mgmt_flow * e =
                                list_entry(p, struct mgmt_flow, next);

                        /* Don't send it back */
                        if (e->instance == instance)
                                continue;

                        if (write_ro_msg(e->instance, msg, name, opcode)) {
                                LOG_ERR("Failed to send to a neighbor.");
                                pthread_rwlock_unlock(&rib.flows_lock);
                                ro_msg__free_unpacked(msg, NULL);
                                return -1;
                        }
                }
                pthread_rwlock_unlock(&rib.flows_lock);
        }

        ro_msg__free_unpacked(msg, NULL);

        return ret;
}

static struct cdap_ops ribmgr_cdap_ops = {
        .cdap_reply   = ribmgr_cdap_reply,
        .cdap_request = ribmgr_cdap_request
};

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_cdap_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_wrlock(&ipcpi.state_lock);
        pthread_rwlock_wrlock(&rib.flows_lock);
        if (list_empty(&rib.flows) && ipcp_get_state() == IPCP_INIT) {
                ipcp_set_state(IPCP_PENDING_ENROLL);

                pthread_mutex_lock(&rib.cdap_reqs_lock);
                iid = cdap_send_request(instance,
                                        CDAP_START,
                                        ENROLLMENT,
                                        NULL, 0, 0);
                if (iid < 0) {
                        pthread_mutex_unlock(&rib.cdap_reqs_lock);
                        pthread_rwlock_unlock(&rib.flows_lock);
                        pthread_rwlock_unlock(&ipcpi.state_lock);
                        LOG_ERR("Failed to start enrollment.");
                        cdap_destroy(instance);
                        free(flow);
                        return -1;
                }

                if (cdap_result_wait(instance, CDAP_START,
                                     ENROLLMENT, iid)) {
                        pthread_mutex_unlock(&rib.cdap_reqs_lock);
                        pthread_rwlock_unlock(&rib.flows_lock);
                        pthread_rwlock_unlock(&ipcpi.state_lock);
                        LOG_ERR("Failed to start enrollment.");
                        cdap_destroy(instance);
                        free(flow);
                        return -1;
                }
                pthread_mutex_unlock(&rib.cdap_reqs_lock);
        }

        list_add(&flow->next, &rib.flows);
        pthread_rwlock_unlock(&rib.flows_lock);
        pthread_rwlock_unlock(&ipcpi.state_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);
                        pthread_rwlock_unlock(&rib.flows_lock);
                        free(flow);
                        return 0;
                }
        }
        pthread_rwlock_unlock(&rib.flows_lock);

        return -1;
}

int ribmgr_bootstrap(struct dif_config * conf)
{
        static_info_msg_t stat_info = STATIC_INFO_MSG__INIT;
        uint8_t * data = NULL;
        size_t len = 0;
        struct ro_props * props;

        if (conf == NULL ||
            conf->type != IPCP_NORMAL) {
                LOG_ERR("Bad DIF configuration.");
                return -1;
        }

        props = malloc(sizeof(*props));
        if (props == NULL) {
                LOG_ERR("Failed to allocate memory.");
                return -1;
        }

        props->enrol_sync = true;
        props->recv_set = NEIGHBORS;
        props->expiry.tv_sec = 0;
        props->expiry.tv_nsec = 0;

        if (ribmgr_ro_create(RIBMGR_PREFIX, props, NULL, 0) == NULL) {
                LOG_ERR("Failed to create RIBMGR RO.");
                free(props);
                return -1;
        }

        stat_info.addr_size = rib.dtc.addr_size = conf->addr_size;
        stat_info.cep_id_size = rib.dtc.cep_id_size  = conf->cep_id_size;
        stat_info.pdu_length_size = rib.dtc.pdu_length_size
                = conf->pdu_length_size;
        stat_info.seqno_size = rib.dtc.seqno_size = conf->seqno_size;
        stat_info.has_ttl = rib.dtc.has_ttl = conf->has_ttl;
        stat_info.has_chk = rib.dtc.has_chk = conf->has_chk;
        stat_info.min_pdu_size = rib.dtc.min_pdu_size = conf->min_pdu_size;
        stat_info.max_pdu_size = rib.dtc.max_pdu_size = conf->max_pdu_size;

        rib.addr_auth = addr_auth_create(conf->addr_auth_type);
        if (rib.addr_auth == NULL) {
                LOG_ERR("Failed to create address authority.");
                ro_delete(RIBMGR_PREFIX);
                return -1;
        }

        stat_info.addr_auth_type = rib.addr_auth->type;

        len = static_info_msg__get_packed_size(&stat_info);
        if (len == 0) {
                LOG_ERR("Failed to get size of static information.");
                addr_auth_destroy(rib.addr_auth);
                ribmgr_ro_delete(RIBMGR_PREFIX);
                return -1;
        }

        data = malloc(len);
        if (data == NULL) {
                LOG_ERR("Failed to allocate memory.");
                addr_auth_destroy(rib.addr_auth);
                ribmgr_ro_delete(RIBMGR_PREFIX);
                return -1;
        }

        static_info_msg__pack(&stat_info, data);

        props = malloc(sizeof(*props));
        if (props == NULL) {
                LOG_ERR("Failed to allocate memory.");
                free(data);
                addr_auth_destroy(rib.addr_auth);
                ribmgr_ro_delete(RIBMGR_PREFIX);
                return -1;
        }

        props->enrol_sync = true;
        props->recv_set = NEIGHBORS;
        props->expiry.tv_sec = 0;
        props->expiry.tv_nsec = 0;

        if (ribmgr_ro_create(RIBMGR_PREFIX STAT_INFO,
                             props, data, len) == NULL) {
                LOG_ERR("Failed to create static info RO.");
                free(data);
                free(props);
                addr_auth_destroy(rib.addr_auth);
                ribmgr_ro_delete(RIBMGR_PREFIX);
                return -1;
        }

        rib.address = rib.addr_auth->address();
        LOG_DBG("IPCP has address %lu", rib.address);

        if (frct_init()) {
                LOG_ERR("Failed to initialize FRCT.");
                ribmgr_ro_delete(RIBMGR_PREFIX STAT_INFO);
                addr_auth_destroy(rib.addr_auth);
                ribmgr_ro_delete(RIBMGR_PREFIX);
                return -1;
        }

        LOG_DBG("Bootstrapped RIB Manager.");

        return 0;
}

struct dt_const * ribmgr_dt_const()
{
        return &(rib.dtc);
}

uint64_t ribmgr_address()
{
        return rib.address;
}

static int send_neighbors_ro(char *           name,
                             ro_msg_t *       msg,
                             enum cdap_opcode code)
{
        struct list_head * p = NULL;

        pthread_rwlock_rdlock(&rib.flows_lock);
        list_for_each(p, &rib.flows) {
                struct mgmt_flow * e = list_entry(p, struct mgmt_flow, next);

                if (write_ro_msg(e->instance, msg, name, code)) {
                        LOG_ERR("Failed to send to a neighbor.");
                        pthread_rwlock_unlock(&rib.flows_lock);
                        return -1;
                }
        }
        pthread_rwlock_unlock(&rib.flows_lock);

        return 0;
}

int ro_create(const char *      name,
              struct ro_props * props,
              uint8_t *         data,
              size_t            len)
{
        struct rnode * node;
        ro_msg_t msg = RO_MSG__INIT;

        if (name == NULL || props == NULL)
                return -EINVAL;

        pthread_mutex_lock(&rib.ro_lock);

        node = ribmgr_ro_create(name, props, data, len);
        if (node == NULL) {
                pthread_mutex_unlock(&rib.ro_lock);
                LOG_ERR("Failed to create RO.");
                return -1;
        }

        if (node->props->recv_set == NO_SYNC) {
                pthread_mutex_unlock(&rib.ro_lock);
                return 0;
        }

        if (ro_msg_create(node, &msg)) {
                pthread_mutex_unlock(&rib.ro_lock);
                LOG_ERR("Failed to create RO msg.");
                return -1;
        }

        if (send_neighbors_ro(node->full_name, &msg, CDAP_CREATE)) {
                pthread_mutex_unlock(&rib.ro_lock);
                LOG_ERR("Failed to send to neighbors.");
                return -1;
        }

        pthread_mutex_unlock(&rib.ro_lock);

        return 0;
}

int ro_delete(const char * name)
{
        struct rnode * node;
        ro_msg_t msg = RO_MSG__INIT;

        if (name == NULL)
                return -EINVAL;

        pthread_mutex_lock(&rib.ro_lock);

        node = find_rnode_by_name(name);
        if (node == NULL) {
                pthread_mutex_unlock(&rib.ro_lock);
                LOG_ERR("Failed to sync RO.");
                return -1;
        }

        if (node->props->recv_set != NO_SYNC) {
                if (ro_msg_create(node, &msg)) {
                        pthread_mutex_unlock(&rib.ro_lock);
                        LOG_ERR("Failed to create RO msg.");
                        return -1;
                }

                if (send_neighbors_ro(node->full_name, &msg, CDAP_DELETE)) {
                        pthread_mutex_unlock(&rib.ro_lock);
                        LOG_ERR("Failed to send to neighbors.");
                        return -1;
                }
        }

        if (ribmgr_ro_delete(name)) {
                pthread_mutex_unlock(&rib.ro_lock);
                return -1;
        }

        pthread_mutex_unlock(&rib.ro_lock);

        return 0;
}

int ro_write(const char * name,
             uint8_t *    data,
             size_t       len)
{
        struct rnode * node;
        ro_msg_t msg = RO_MSG__INIT;

        if (name == NULL || data == NULL)
                return -EINVAL;

        pthread_mutex_lock(&rib.ro_lock);

        node = ribmgr_ro_write(name, data, len);
        if (node == NULL) {
                pthread_mutex_unlock(&rib.ro_lock);
                LOG_ERR("Failed to create RO.");
                return -1;
        }

        if (node->props->recv_set == NO_SYNC) {
                pthread_mutex_unlock(&rib.ro_lock);
                return 0;
        }

        if (ro_msg_create(node, &msg)) {
                pthread_mutex_unlock(&rib.ro_lock);
                LOG_ERR("Failed to create RO msg.");
                return -1;
        }

        if (send_neighbors_ro(node->full_name, &msg, CDAP_WRITE)) {
                pthread_mutex_unlock(&rib.ro_lock);
                LOG_ERR("Failed to send to neighbors.");
                return -1;
        }

        pthread_mutex_unlock(&rib.ro_lock);

        return 0;
}

ssize_t ro_read(const char * name,
                uint8_t **   data)
{
        struct rnode * node;
        ssize_t        len;

        if (name == NULL || data == NULL)
                return -EINVAL;

        pthread_mutex_lock(&rib.ro_lock);

        node = find_rnode_by_name(name);
        if (node == NULL) {
                pthread_mutex_unlock(&rib.ro_lock);
                return -1;
        }

        *data = malloc(node->len);
        if (*data == NULL) {
                pthread_mutex_unlock(&rib.ro_lock);
                return -1;
        }

        memcpy(*data, node->data, node->len);
        len = node->len;

        pthread_mutex_unlock(&rib.ro_lock);

        return len;
}

int ro_subscribe(const char *        name,
                 struct ro_sub_ops * ops)
{
        struct ro_sub * sub;

        if (name == NULL || ops == NULL)
                return -EINVAL;

        sub = malloc(sizeof(*sub));
        if (sub == NULL)
                return -ENOMEM;

        INIT_LIST_HEAD(&sub->next);

        sub->name = strdup(name);
        if (sub->name == NULL) {
                free(sub);
                return -1;
        }

        sub->ops = ops;

        pthread_mutex_lock(&rib.subs_lock);

        sub->sid = bmp_allocate(rib.sids);
        if (sub->sid < 0) {
                pthread_mutex_unlock(&rib.subs_lock);
                free(sub->name);
                free(sub);
                LOG_ERR("Failed to get sub id.");
        }

        list_add(&sub->next, &rib.subs);

        pthread_mutex_unlock(&rib.subs_lock);

        return 0;
}

int ro_unsubscribe(int sid)
{
        struct list_head * pos = NULL;
        struct list_head * n   = NULL;

        pthread_mutex_lock(&rib.subs_lock);

        list_for_each_safe(pos, n, &(rib.subs)) {
                struct ro_sub * e = list_entry(pos, struct ro_sub, next);
                if (sid == e->sid) {
                        bmp_release(rib.sids, sid);
                        list_del(&e->next);
                        free(e->name);
                        free(e);
                        pthread_mutex_unlock(&rib.subs_lock);
                        return 0;
                }
        }

        pthread_mutex_unlock(&rib.subs_lock);

        LOG_ERR("No such subscription found.");

        return -1;
}