From 45c6615484ffe347654c34decb72ff1ef9bde0f3 Mon Sep 17 00:00:00 2001 From: dimitri staessens Date: Sat, 9 Sep 2017 13:50:47 +0200 Subject: ipcpd: Revise internals of normal IPCP This removes the RIB as a datastructure and CDAP as the protocol between IPCPs. CDAP, the rib and related sources are deprecated. The link-state protocol policy is udpated to use its own protocol based on a simple broadcast strategy along a tree. The neighbors struct is deprecated and moved to the library as a generic notifier component. --- src/lib/cdap.c | 868 --------------------------------------------------------- 1 file changed, 868 deletions(-) delete mode 100644 src/lib/cdap.c (limited to 'src/lib/cdap.c') diff --git a/src/lib/cdap.c b/src/lib/cdap.c deleted file mode 100644 index d9cb2036..00000000 --- a/src/lib/cdap.c +++ /dev/null @@ -1,868 +0,0 @@ -/* - * Ouroboros - Copyright (C) 2016 - 2017 - * - * The Common Distributed Application Protocol - * - * Dimitri Staessens - * Sander Vrijders - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public License - * version 2.1 as published by the Free Software Foundation. - * - * This library 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 - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., http://www.fsf.org/about/contact/. - */ - -#define _POSIX_C_SOURCE 200809L - -#include -#include -#include -#include -#include - -#include "cdap_req.h" - -#include -#include -#include -#include - -#include "cdap.pb-c.h" -typedef Cdap cdap_t; - -#define CDAP_REPLY (CDAP_DELETE + 1) - -#define INVALID_ID -1 -#define IDS_SIZE 2048 -#define BUF_SIZE 2048 - -struct fd_el { - struct list_head next; - - int fd; -}; - -struct cdap { - fset_t * set; - fqueue_t * fq; - - bool proc; - pthread_mutex_t mtx; - pthread_cond_t cond; - - size_t n_flows; - struct list_head flows; - pthread_rwlock_t flows_lock; - - struct bmp * ids; - pthread_mutex_t ids_lock; - - struct list_head sent; - pthread_rwlock_t sent_lock; - - struct list_head rcvd; - pthread_cond_t rcvd_cond; - pthread_mutex_t rcvd_lock; - - pthread_t reader; -}; - -struct cdap_rcvd { - struct list_head next; - - int fd; - bool proc; - - invoke_id_t iid; - cdap_key_t key; - - enum cdap_opcode opcode; - char * name; - void * data; - size_t len; - uint32_t flags; -}; - -static int next_id(struct cdap * instance) -{ - int ret; - - assert(instance); - - pthread_mutex_lock(&instance->ids_lock); - - ret = bmp_allocate(instance->ids); - if (!bmp_is_id_valid(instance->ids, ret)) - ret = INVALID_ID; - - pthread_mutex_unlock(&instance->ids_lock); - - return ret; -} - -static int release_id(struct cdap * instance, - int32_t id) -{ - int ret; - - assert(instance); - - pthread_mutex_lock(&instance->ids_lock); - - ret = bmp_release(instance->ids, id); - - pthread_mutex_unlock(&instance->ids_lock); - - return ret; -} - -#define cdap_sent_has_key(i, key) (cdap_sent_get_by_key(i, key) != NULL) - -static struct cdap_req * cdap_sent_get_by_key(struct cdap * instance, - cdap_key_t key) -{ - struct list_head * p = NULL; - struct cdap_req * req = NULL; - - assert(instance); - - pthread_rwlock_rdlock(&instance->sent_lock); - - list_for_each(p, &instance->sent) { - req = list_entry(p, struct cdap_req, next); - if (req->key == key) { - pthread_rwlock_unlock(&instance->sent_lock); - return req; - } - } - - pthread_rwlock_unlock(&instance->sent_lock); - - return NULL; -} - -static struct cdap_req * cdap_sent_get_by_iid(struct cdap * instance, - invoke_id_t iid) -{ - struct list_head * p = NULL; - struct cdap_req * req = NULL; - - assert(instance); - - pthread_rwlock_rdlock(&instance->sent_lock); - - list_for_each(p, &instance->sent) { - req = list_entry(p, struct cdap_req, next); - if (req->iid == iid) { - pthread_rwlock_unlock(&instance->sent_lock); - return req; - } - } - - pthread_rwlock_unlock(&instance->sent_lock); - - return NULL; -} - -static struct cdap_rcvd * cdap_rcvd_get_by_key(struct cdap * instance, - cdap_key_t key) -{ - struct list_head * p = NULL; - struct list_head * h = NULL; - struct cdap_rcvd * rcvd = NULL; - - assert(instance); - - pthread_mutex_lock(&instance->rcvd_lock); - - list_for_each_safe(p, h, &instance->rcvd) { - rcvd = list_entry(p, struct cdap_rcvd, next); - if (rcvd->key == key) { - list_del(&rcvd->next); - pthread_mutex_unlock(&instance->rcvd_lock); - return rcvd; - } - } - - pthread_mutex_unlock(&instance->rcvd_lock); - - assert(false); - - return NULL; -} - -static struct cdap_req * cdap_sent_add(struct cdap * instance, - int fd, - invoke_id_t iid, - cdap_key_t key) -{ - struct cdap_req * req; - - assert(instance); - assert(!cdap_sent_has_key(instance, key)); - - req = cdap_req_create(fd, iid, key); - if (req == NULL) - return NULL; - - pthread_rwlock_wrlock(&instance->sent_lock); - - list_add(&req->next, &instance->sent); - - pthread_rwlock_unlock(&instance->sent_lock); - - return req; -} - -static void cdap_sent_del(struct cdap * instance, - struct cdap_req * req) -{ - assert(instance); - assert(req); - - assert(cdap_sent_has_key(instance, req->key)); - - pthread_rwlock_wrlock(&instance->sent_lock); - - list_del(&req->next); - - pthread_rwlock_unlock(&instance->sent_lock); - - cdap_req_destroy(req); -} - -static void cdap_sent_destroy(struct cdap * instance) -{ - struct list_head * p = NULL; - struct list_head * h = NULL; - - assert(instance); - - pthread_rwlock_wrlock(&instance->sent_lock); - - list_for_each_safe(p, h, &instance->sent) { - struct cdap_req * req = list_entry(p, struct cdap_req, next); - list_del(&req->next); - cdap_req_cancel(req); - cdap_req_destroy(req); - } - - pthread_rwlock_unlock(&instance->sent_lock); -} - -static void cdap_rcvd_destroy(struct cdap * instance) -{ - struct list_head * p = NULL; - struct list_head * h = NULL; - - assert(instance); - - pthread_mutex_lock(&instance->rcvd_lock); - - list_for_each_safe(p, h, &instance->rcvd) { - struct cdap_rcvd * r = list_entry(p, struct cdap_rcvd, next); - list_del(&r->next); - if (r->data != NULL) - free(r->data); - if (r->name != NULL) - free(r->name); - free(r); - } - - pthread_cond_broadcast(&instance->rcvd_cond); - - pthread_mutex_unlock(&instance->rcvd_lock); -} - -static void set_proc(struct cdap * instance, - bool status) -{ - pthread_mutex_lock(&instance->mtx); - - instance->proc = status; - pthread_cond_signal(&instance->cond); - - pthread_mutex_unlock(&instance->mtx); -} - -static void * sdu_reader(void * o) -{ - struct cdap * instance = (struct cdap *) o; - struct cdap_req * req; - struct cdap_rcvd * rcvd; - cdap_t * msg; - uint8_t buf[BUF_SIZE]; - ssize_t len; - buffer_t data; - - while (fevent(instance->set, instance->fq, NULL)) { - int fd; - set_proc(instance, true); - fd = fqueue_next(instance->fq); - len = flow_read(fd, buf, BUF_SIZE); - if (len < 0) { - set_proc(instance, false); - continue; - } - - msg = cdap__unpack(NULL, len, buf); - if (msg == NULL) { - set_proc(instance, false); - continue; - } - - if (msg->opcode != CDAP_REPLY) { - rcvd = malloc(sizeof(*rcvd)); - if (rcvd == NULL) { - cdap__free_unpacked(msg, NULL); - set_proc(instance, false); - continue; - } - - assert(msg->name); - - rcvd->opcode = msg->opcode; - rcvd->fd = fd; - rcvd->iid = msg->invoke_id; - rcvd->key = next_id(instance); - if (rcvd->key == INVALID_ID) { - cdap__free_unpacked(msg, NULL); - set_proc(instance, false); - free(rcvd); - continue; - } - - rcvd->flags = msg->flags; - rcvd->proc = false; - rcvd->name = strdup(msg->name); - if (rcvd->name == NULL) { - release_id(instance, rcvd->key); - cdap__free_unpacked(msg, NULL); - set_proc(instance, false); - free(rcvd); - continue; - } - - if (msg->has_value) { - rcvd->len = msg->value.len; - rcvd->data = malloc(rcvd->len); - if (rcvd->data == NULL) { - release_id(instance, rcvd->key); - cdap__free_unpacked(msg, NULL); - set_proc(instance, false); - free(rcvd->name); - free(rcvd); - continue; - } - memcpy(rcvd->data, msg->value.data, rcvd->len); - } else { - rcvd->len = 0; - rcvd->data = NULL; - } - - pthread_mutex_lock(&instance->rcvd_lock); - - list_add(&rcvd->next, &instance->rcvd); - - pthread_cond_signal(&instance->rcvd_cond); - pthread_mutex_unlock(&instance->rcvd_lock); - } else { - req = cdap_sent_get_by_iid(instance, msg->invoke_id); - if (req == NULL) { - cdap__free_unpacked(msg, NULL); - set_proc(instance, false); - continue; - } - - if (msg->has_value) { - data.len = msg->value.len; - data.data = malloc(data.len); - if (data.data == NULL) { - cdap__free_unpacked(msg, NULL); - set_proc(instance, false); - continue; - } - memcpy(data.data, msg->value.data, data.len); - } else { - data.len = 0; - data.data = NULL; - } - - cdap_req_respond(req, msg->result, data); - } - - cdap__free_unpacked(msg, NULL); - set_proc(instance, false); - } - - return (void *) 0; -} - -struct cdap * cdap_create() -{ - struct cdap * instance = NULL; - - instance = malloc(sizeof(*instance)); - if (instance == NULL) - goto fail_malloc; - - if (pthread_rwlock_init(&instance->flows_lock, NULL)) - goto fail_flows_lock; - - if (pthread_mutex_init(&instance->ids_lock, NULL)) - goto fail_ids_lock; - - if (pthread_mutex_init(&instance->rcvd_lock, NULL)) - goto fail_rcvd_lock; - - if (pthread_rwlock_init(&instance->sent_lock, NULL)) - goto fail_sent_lock; - - if (pthread_cond_init(&instance->rcvd_cond, NULL)) - goto fail_rcvd_cond; - - if (pthread_mutex_init(&instance->mtx, NULL)) - goto fail_mtx; - - if (pthread_cond_init(&instance->cond, NULL)) - goto fail_cond; - - instance->ids = bmp_create(IDS_SIZE, 0); - if (instance->ids == NULL) - goto fail_bmp_create; - - instance->set = fset_create(); - if (instance->set == NULL) - goto fail_set_create; - - instance->fq = fqueue_create(); - if (instance->fq == NULL) - goto fail_fqueue_create; - - instance->n_flows = 0; - instance->proc = false; - - list_head_init(&instance->flows); - list_head_init(&instance->sent); - list_head_init(&instance->rcvd); - - if (pthread_create(&instance->reader, NULL, sdu_reader, instance)) - goto fail_pthread_create; - - return instance; - - fail_pthread_create: - fqueue_destroy(instance->fq); - fail_fqueue_create: - fset_destroy(instance->set); - fail_set_create: - bmp_destroy(instance->ids); - fail_bmp_create: - pthread_cond_destroy(&instance->cond); - fail_cond: - pthread_mutex_destroy(&instance->mtx); - fail_mtx: - pthread_cond_destroy(&instance->rcvd_cond); - fail_rcvd_cond: - pthread_rwlock_destroy(&instance->sent_lock); - fail_sent_lock: - pthread_mutex_destroy(&instance->rcvd_lock); - fail_rcvd_lock: - pthread_mutex_destroy(&instance->ids_lock); - fail_ids_lock: - pthread_rwlock_destroy(&instance->flows_lock); - fail_flows_lock: - free(instance); - fail_malloc: - return NULL; -} - -int cdap_destroy(struct cdap * instance) -{ - struct list_head * p; - struct list_head * h; - - if (instance == NULL) - return 0; - - pthread_cancel(instance->reader); - pthread_join(instance->reader, NULL); - - fqueue_destroy(instance->fq); - - fset_destroy(instance->set); - - pthread_cond_destroy(&instance->cond); - pthread_mutex_destroy(&instance->mtx); - - pthread_rwlock_wrlock(&instance->flows_lock); - - list_for_each_safe(p,h, &instance->flows) { - struct fd_el * e = list_entry(p, struct fd_el, next); - list_del(&e->next); - free(e); - } - - pthread_rwlock_unlock(&instance->flows_lock); - - pthread_rwlock_destroy(&instance->flows_lock); - - pthread_mutex_lock(&instance->ids_lock); - - bmp_destroy(instance->ids); - - pthread_mutex_unlock(&instance->ids_lock); - - pthread_mutex_destroy(&instance->ids_lock); - - cdap_sent_destroy(instance); - - pthread_rwlock_destroy(&instance->sent_lock); - - cdap_rcvd_destroy(instance); - - pthread_mutex_destroy(&instance->rcvd_lock); - - free(instance); - - return 0; -} - -int cdap_add_flow(struct cdap * instance, - int fd) -{ - struct fd_el * e; - - if (fd < 0) - return -EINVAL; - - e = malloc(sizeof(*e)); - if (e == NULL) - return -ENOMEM; - - e->fd = fd; - - pthread_rwlock_wrlock(&instance->flows_lock); - - if (fset_add(instance->set, fd)) { - pthread_rwlock_unlock(&instance->flows_lock); - free(e); - return -1; - } - - list_add(&e->next, &instance->flows); - - ++instance->n_flows; - - pthread_rwlock_unlock(&instance->flows_lock); - - return 0; -} - -int cdap_del_flow(struct cdap * instance, - int fd) -{ - struct list_head * p; - struct list_head * h; - - if (fd < 0) - return -EINVAL; - - pthread_rwlock_wrlock(&instance->flows_lock); - - fset_del(instance->set, fd); - - list_for_each_safe(p, h, &instance->flows) { - struct fd_el * e = list_entry(p, struct fd_el, next); - if (e->fd == fd) { - list_del(&e->next); - free(e); - break; - } - } - - --instance->n_flows; - - pthread_rwlock_unlock(&instance->flows_lock); - - pthread_mutex_lock(&instance->mtx); - - pthread_cleanup_push((void(*)(void *))pthread_mutex_unlock, - (void *) &instance->mtx); - - while (instance->proc) - pthread_cond_wait(&instance->cond, &instance->mtx); - - pthread_cleanup_pop(true); - - return 0; -} - -static int write_msg(int fd, - cdap_t * msg) -{ - uint8_t * data; - size_t len; - - assert(msg); - - len = cdap__get_packed_size(msg); - if (len == 0) - return -1; - - data = malloc(len); - if (data == NULL) - return -ENOMEM; - - cdap__pack(msg, data); - - if (flow_write(fd, data, len)) { - free(data); - return -1; - } - - free(data); - - return 0; -} - -cdap_key_t * cdap_request_send(struct cdap * instance, - enum cdap_opcode code, - const char * name, - const void * data, - size_t len, - uint32_t flags) -{ - cdap_key_t * keys; - cdap_key_t * key; - cdap_t msg = CDAP__INIT; - struct list_head * p; - int ret; - - if (instance == NULL || name == NULL || code > CDAP_DELETE) - return NULL; - - pthread_rwlock_rdlock(&instance->flows_lock); - - keys = malloc(sizeof(*keys) * (instance->n_flows + 1)); - if (keys == NULL) { - pthread_rwlock_unlock(&instance->flows_lock); - return NULL; - } - - memset(keys, INVALID_CDAP_KEY, sizeof(*keys) * (instance->n_flows + 1)); - - key = keys; - - cdap__init(&msg); - - msg.opcode = code; - msg.name = (char *) name; - msg.has_flags = true; - msg.flags = flags; - - if (data != NULL) { - msg.has_value = true; - msg.value.data = (uint8_t *) data; - msg.value.len = len; - } - - list_for_each(p, &instance->flows) { - struct cdap_req * req; - invoke_id_t iid; - struct fd_el * e; - - iid = next_id(instance); - if (iid == INVALID_ID) { - pthread_rwlock_unlock(&instance->flows_lock); - return keys; - } - - msg.invoke_id = iid; - - e = list_entry(p, struct fd_el, next); - - *key = next_id(instance); - if (*key == INVALID_ID) { - release_id(instance, iid); - pthread_rwlock_unlock(&instance->flows_lock); - return keys; - } - - req = cdap_sent_add(instance, e->fd, iid, *key); - if (req == NULL) { - release_id(instance, *key); - release_id(instance, iid); - pthread_rwlock_unlock(&instance->flows_lock); - *key = INVALID_CDAP_KEY; - return keys; - } - - ret = write_msg(e->fd, &msg); - if (ret == -ENOMEM) { - cdap_sent_del(instance, req); - release_id(instance, *key); - release_id(instance, iid); - pthread_rwlock_unlock(&instance->flows_lock); - *key = INVALID_CDAP_KEY; - return keys; - } - - if (ret < 0) { - cdap_sent_del(instance, req); - release_id(instance, *key); - release_id(instance, iid); - pthread_rwlock_unlock(&instance->flows_lock); - *key = INVALID_CDAP_KEY; - return keys; - } - - ++key; - } - - pthread_rwlock_unlock(&instance->flows_lock); - - return keys; -} - -int cdap_reply_wait(struct cdap * instance, - cdap_key_t key, - uint8_t ** data, - size_t * len) -{ - int ret; - struct cdap_req * r; - invoke_id_t iid; - - if (instance == NULL || (data != NULL && len == NULL)) - return -EINVAL; - - r = cdap_sent_get_by_key(instance, key); - if (r == NULL) - return -EINVAL; - - iid = r->iid; - - ret = cdap_req_wait(r); - if (ret < 0) { - cdap_sent_del(instance, r); - release_id(instance, iid); - release_id(instance, key); - return ret; - } - - assert(ret == 0); - - if (data != NULL) { - *data = r->data.data; - *len = r->data.len; - } - - ret = r->response; - - cdap_sent_del(instance, r); - release_id(instance, iid); - release_id(instance, key); - - return ret; -} - -cdap_key_t cdap_request_wait(struct cdap * instance, - enum cdap_opcode * opcode, - char ** name, - uint8_t ** data, - size_t * len, - uint32_t * flags) -{ - struct cdap_rcvd * rcv = NULL; - - if (instance == NULL || opcode == NULL || name == NULL || data == NULL - || len == NULL || flags == NULL) - return -EINVAL; - - pthread_mutex_lock(&instance->rcvd_lock); - - pthread_cleanup_push((void(*)(void *))pthread_mutex_unlock, - (void *) &instance->rcvd_lock); - - while (rcv == NULL) { - while (list_is_empty(&instance->rcvd)) - pthread_cond_wait(&instance->rcvd_cond, - &instance->rcvd_lock); - - rcv = list_first_entry(&instance->rcvd, struct cdap_rcvd, next); - if (rcv->proc) { - rcv = NULL; - pthread_cond_wait(&instance->rcvd_cond, - &instance->rcvd_lock); - } - } - - assert(rcv->proc == false); - - rcv->proc = true; - list_del(&rcv->next); - list_add_tail(&rcv->next, &instance->rcvd); - - pthread_cleanup_pop(true); - - *opcode = rcv->opcode; - *name = rcv->name; - *data = rcv->data; - *len = rcv->len; - *flags = rcv->flags; - - rcv->name = NULL; - rcv->data = NULL; - - return rcv->key; -} - -int cdap_reply_send(struct cdap * instance, - cdap_key_t key, - int result, - const void * data, - size_t len) -{ - int fd; - cdap_t msg = CDAP__INIT; - struct cdap_rcvd * rcvd; - - if (instance == NULL) - return -EINVAL; - - rcvd = cdap_rcvd_get_by_key(instance, key); - if (rcvd == NULL) - return -1; - - msg.opcode = CDAP_REPLY; - msg.invoke_id = rcvd->iid; - msg.has_result = true; - msg.result = result; - - if (data != NULL) { - msg.has_value = true; - msg.value.data = (uint8_t *) data; - msg.value.len = len; - } - - fd = rcvd->fd; - - release_id(instance, rcvd->key); - - assert(rcvd->data == NULL); - assert(rcvd->name == NULL); - assert(rcvd->proc); - - free(rcvd); - - return write_msg(fd, &msg); -} -- cgit v1.2.3