From 3d99be39d1c803ff0faa049fe31b9c7b3302890a Mon Sep 17 00:00:00 2001 From: dimitri staessens Date: Fri, 24 Mar 2017 14:38:09 +0100 Subject: ipcpd: Initial synchronisation of the RIB --- src/ipcpd/normal/main.c | 4 +- src/ipcpd/normal/ribmgr.c | 284 ++++++++++++++++++++++++++++++++++++++++++---- 2 files changed, 264 insertions(+), 24 deletions(-) (limited to 'src/ipcpd') diff --git a/src/ipcpd/normal/main.c b/src/ipcpd/normal/main.c index 00ecaae9..41e0544d 100644 --- a/src/ipcpd/normal/main.c +++ b/src/ipcpd/normal/main.c @@ -202,6 +202,8 @@ static int boot_components(void) void shutdown_components(void) { + ribmgr_fini(); + connmgr_stop(); enroll_stop(); @@ -214,8 +216,6 @@ void shutdown_components(void) dir_fini(); - ribmgr_fini(); - addr_auth_fini(); free(ipcpi.dif_name); diff --git a/src/ipcpd/normal/ribmgr.c b/src/ipcpd/normal/ribmgr.c index ef79f627..2696026a 100644 --- a/src/ipcpd/normal/ribmgr.c +++ b/src/ipcpd/normal/ribmgr.c @@ -44,35 +44,272 @@ #include #include +#define RIB_SYNC_TIMEOUT 1 + +enum ribmgr_state { + RIBMGR_NULL = 0, + RIBMGR_OPERATIONAL, + RIBMGR_SHUTDOWN +}; + struct { - flow_set_t * fs; - fqueue_t * fq; + struct cdap * cdap; + + pthread_t reader; + pthread_t sync; struct gam * gam; struct nbs * nbs; struct ae * ae; struct nb_notifier nb_notifier; + + pthread_rwlock_t state_lock; + enum ribmgr_state state; } ribmgr; static int ribmgr_neighbor_event(enum nb_event event, struct conn conn) { - /* We are only interested in neighbors being added and removed. */ switch (event) { case NEIGHBOR_ADDED: - flow_set_add(ribmgr.fs, conn.flow_info.fd); + cdap_add_flow(ribmgr.cdap, conn.flow_info.fd); break; case NEIGHBOR_REMOVED: - flow_set_del(ribmgr.fs, conn.flow_info.fd); + cdap_del_flow(ribmgr.cdap, conn.flow_info.fd); break; default: + /* Don't care about other events */ break; } return 0; } +static enum ribmgr_state ribmgr_get_state(void) +{ + enum ribmgr_state state; + + pthread_rwlock_rdlock(&ribmgr.state_lock); + + state = ribmgr.state; + + pthread_rwlock_unlock(&ribmgr.state_lock); + + return state; +} + +static void ribmgr_set_state(enum ribmgr_state state) +{ + pthread_rwlock_wrlock(&ribmgr.state_lock); + + ribmgr.state = state; + + pthread_rwlock_unlock(&ribmgr.state_lock); +} + +static void * reader(void * o) +{ + cdap_key_t key; + enum cdap_opcode oc; + char * name; + uint8_t * data; + size_t len; + ssize_t slen; + uint32_t flags; + uint8_t * buf; + int rval; + + (void) o; + + while (ribmgr_get_state() == RIBMGR_OPERATIONAL) { + key = cdap_request_wait(ribmgr.cdap, &oc, &name, &data, + (size_t *) &len , &flags); + assert(key != -EINVAL); + + if (key == INVALID_CDAP_KEY) { + log_warn("Bad CDAP request."); + continue; + } + + assert(name); + assert(strlen(name)); + + switch (oc) { + case CDAP_READ: + assert(len == 0); + slen = rib_pack(name, &buf, PACK_HASH_ROOT); + if (slen < 0) { + log_err("Failed to pack %s.", name); + cdap_reply_send(ribmgr.cdap, key, -1, NULL, 0); + free(name); + continue; + } + + log_dbg("Packed %s (%zu bytes).", name, slen); + + free(name); + + if (cdap_reply_send(ribmgr.cdap, key, 0, buf, slen)) { + log_err("Failed to send CDAP reply."); + free(buf); + continue; + } + + free(buf); + break; + case CDAP_WRITE: + assert(len); + assert(data); + + rval = rib_unpack(data, len, 0); + switch(rval) { + case 0: + break; + case -EFAULT: + log_warn("Hash mismatch, not in sync."); + break; + default: + log_warn("Error unpacking %s.", name); + cdap_reply_send(ribmgr.cdap, key, -1, NULL, 0); + free(name); + free(data); + continue; + } + + free(name); + + if (cdap_reply_send(ribmgr.cdap, key, 0, NULL, 0)) { + log_err("Failed to send CDAP reply."); + continue; + } + break; + case CDAP_CREATE: + assert(len); + assert(data); + + rval = rib_unpack(data, len, UNPACK_CREATE); + switch(rval) { + case 0: + break; + case -EFAULT: + log_warn("Hash mismatch, not yet in sync."); + break; + default: + log_warn("Error unpacking %s.", name); + cdap_reply_send(ribmgr.cdap, key, -1, NULL, 0); + free(name); + free(data); + continue; + } + + free(name); + + if (cdap_reply_send(ribmgr.cdap, key, 0, NULL, 0)) { + log_err("Failed to send CDAP reply."); + continue; + } + break; + case CDAP_DELETE: + assert(len == 0); + if (rib_del(name)) { + log_warn("Failed deleting %s.", name); + cdap_reply_send(ribmgr.cdap, key, -1, NULL, 0); + } + + free(name); + + if (cdap_reply_send(ribmgr.cdap, key, 0, NULL, 0)) { + log_err("Failed to send CDAP reply."); + continue; + } + + break; + case CDAP_START: + case CDAP_STOP: + log_warn("Unsupported CDAP command."); + if (len) + free(data); + break; + default: + log_err("Bad CDAP command."); + if (len) + free(data); + break; + } + } + + return (void *) 0; +} + +char path[RIB_MAX_PATH_LEN + 1]; + +static void path_reset(void) { + path[strlen(RIB_ROOT)] = '\0'; + assert(strcmp(path, RIB_ROOT) == 0); +} + +static int ribmgr_sync(const char * path) +{ + uint8_t * buf; + ssize_t len; + cdap_key_t * keys; + + len = rib_pack(path, &buf, PACK_HASH_ALL); + if (len < 0) { + log_warn("Failed to pack %s.", path); + return -1; + } + + keys = cdap_request_send(ribmgr.cdap, CDAP_CREATE, path, buf, len, 0); + if (keys != NULL) { + cdap_key_t * key = keys; + while (*key != INVALID_CDAP_KEY) + cdap_reply_wait(ribmgr.cdap, *(key++), NULL, NULL); + free(keys); + } + + free(buf); + + return 0; +} + +/* FIXME: Temporary thread, syncs rib with neighbors every second */ +static void * sync_rib(void *o) +{ + char ** children; + ssize_t ch; + + (void) o; + + strcpy(path, RIB_ROOT); + + while (ribmgr_get_state() == RIBMGR_OPERATIONAL) { + sleep(RIB_SYNC_TIMEOUT); + + ch = rib_children(RIB_ROOT, &children); + if (ch <= 0) + continue; + + while (ch > 0) { + path_reset(); + + rib_path_append(path, children[--ch]); + free(children[ch]); + + /* Only sync fsdb, members and directory */ + if (strcmp(path, MEMBERS_PATH) == 0 + || strcmp(path, DIR_PATH) == 0 + || strcmp(path, ROUTING_PATH) == 0) + ribmgr_sync(path); + } + + free(children); + } + + return (void *) 0; +} + int ribmgr_init(void) { enum pol_gam pg; @@ -114,19 +351,9 @@ int ribmgr_init(void) return -1; } - ribmgr.fs = flow_set_create(); - if (ribmgr.fs == NULL) { - log_err("Failed to create flow set."); - gam_destroy(ribmgr.gam); - connmgr_ae_destroy(ribmgr.ae); - nbs_destroy(ribmgr.nbs); - return -1; - } - - ribmgr.fq = fqueue_create(); - if (ribmgr.fq == NULL) { - log_err("Failed to create fq."); - flow_set_destroy(ribmgr.fs); + ribmgr.cdap = cdap_create(); + if (ribmgr.cdap == NULL) { + log_err("Failed to create CDAP instance."); gam_destroy(ribmgr.gam); connmgr_ae_destroy(ribmgr.ae); nbs_destroy(ribmgr.nbs); @@ -136,22 +363,35 @@ int ribmgr_init(void) ribmgr.nb_notifier.notify_call = ribmgr_neighbor_event; if (nbs_reg_notifier(ribmgr.nbs, &ribmgr.nb_notifier)) { log_err("Failed to register notifier."); - fqueue_destroy(ribmgr.fq); - flow_set_destroy(ribmgr.fs); + cdap_destroy(ribmgr.cdap); gam_destroy(ribmgr.gam); connmgr_ae_destroy(ribmgr.ae); nbs_destroy(ribmgr.nbs); return -1; } + pthread_rwlock_init(&ribmgr.state_lock, NULL); + + ribmgr.state = RIBMGR_OPERATIONAL; + + pthread_create(&ribmgr.sync, NULL, sync_rib, NULL); + + pthread_create(&ribmgr.reader, NULL, reader, NULL); + return 0; } void ribmgr_fini(void) { + ribmgr_set_state(RIBMGR_SHUTDOWN); + + pthread_cancel(ribmgr.reader); + + pthread_join(ribmgr.reader, NULL); + pthread_join(ribmgr.sync, NULL); + nbs_unreg_notifier(ribmgr.nbs, &ribmgr.nb_notifier); - flow_set_destroy(ribmgr.fs); - fqueue_destroy(ribmgr.fq); + cdap_destroy(ribmgr.cdap); gam_destroy(ribmgr.gam); connmgr_ae_destroy(ribmgr.ae); nbs_destroy(ribmgr.nbs); -- cgit v1.2.3