diff options
Diffstat (limited to 'src/ipcpd/local')
| -rw-r--r-- | src/ipcpd/local/CMakeLists.txt | 40 | ||||
| -rw-r--r-- | src/ipcpd/local/main.c | 320 | ||||
| -rw-r--r-- | src/ipcpd/local/reg.c | 217 | ||||
| -rw-r--r-- | src/ipcpd/local/reg.h | 45 |
4 files changed, 436 insertions, 186 deletions
diff --git a/src/ipcpd/local/CMakeLists.txt b/src/ipcpd/local/CMakeLists.txt index a84f4f1b..af433d01 100644 --- a/src/ipcpd/local/CMakeLists.txt +++ b/src/ipcpd/local/CMakeLists.txt @@ -1,32 +1,18 @@ -get_filename_component(CURRENT_SOURCE_PARENT_DIR - ${CMAKE_CURRENT_SOURCE_DIR} DIRECTORY) -get_filename_component(CURRENT_BINARY_PARENT_DIR - ${CMAKE_CURRENT_BINARY_DIR} DIRECTORY) +# Local IPCP build configuration -include_directories(${CMAKE_CURRENT_SOURCE_DIR}) -include_directories(${CMAKE_CURRENT_BINARY_DIR}) +add_executable(${IPCP_LOCAL_TARGET} + main.c + reg.c + ${IPCP_SOURCES} +) -include_directories(${CURRENT_SOURCE_PARENT_DIR}) -include_directories(${CURRENT_BINARY_PARENT_DIR}) +target_include_directories(${IPCP_LOCAL_TARGET} PRIVATE ${IPCP_INCLUDE_DIRS}) +target_link_libraries(${IPCP_LOCAL_TARGET} PRIVATE ouroboros-dev) -include_directories(${CMAKE_SOURCE_DIR}/include) -include_directories(${CMAKE_BINARY_DIR}/include) +ouroboros_target_debug_definitions(${IPCP_LOCAL_TARGET}) -set(IPCP_LOCAL_TARGET ipcpd-local CACHE INTERNAL "") +if(IPCP_LOCAL_POLLING) + target_compile_definitions(${IPCP_LOCAL_TARGET} PRIVATE CONFIG_IPCP_LOCAL_POLLING) +endif() -set(LOCAL_SOURCES - # Add source files here - ${CMAKE_CURRENT_SOURCE_DIR}/main.c) - -add_executable(ipcpd-local ${LOCAL_SOURCES} ${IPCP_SOURCES}) -target_link_libraries(ipcpd-local LINK_PUBLIC ouroboros-common ouroboros-dev) - -include(AddCompileFlags) -if (CMAKE_BUILD_TYPE MATCHES "Debug*") - add_compile_flags(ipcpd-local -DCONFIG_OUROBOROS_DEBUG) -endif () - -install(TARGETS ipcpd-local RUNTIME DESTINATION ${CMAKE_INSTALL_SBINDIR}) - -# Enable once ipcp-local has tests -# add_subdirectory(tests) +install(TARGETS ${IPCP_LOCAL_TARGET} RUNTIME DESTINATION ${CMAKE_INSTALL_SBINDIR}) diff --git a/src/ipcpd/local/main.c b/src/ipcpd/local/main.c index 48c2f67b..69eac8a6 100644 --- a/src/ipcpd/local/main.c +++ b/src/ipcpd/local/main.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2026 * * Local IPC process * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This program is free software; you can redistribute it and/or modify * it under the terms of the GNU General Public License version 2 as @@ -38,9 +38,11 @@ #include <ouroboros/ipcp.h> #include <ouroboros/ipcp-dev.h> #include <ouroboros/local-dev.h> +#include <ouroboros/np1_flow.h> #include "ipcp.h" -#include "shim-data.h" +#include "np1.h" +#include "reg.h" #include <string.h> #include <stdlib.h> @@ -48,20 +50,17 @@ #include <sys/wait.h> #include <assert.h> -#define THIS_TYPE IPCP_LOCAL -#define ALLOC_TIMEOUT 10 /* ms */ - -struct ipcp ipcpi; +#define THIS_TYPE IPCP_LOCAL struct { - struct shim_data * shim_data; + struct reg * reg; - int in_out[SYS_MAX_FLOWS]; - fset_t * flows; - fqueue_t * fq; + int in_out[SYS_MAX_FLOWS]; + fset_t * flows; + fqueue_t * fq; - pthread_rwlock_t lock; - pthread_t packet_loop; + pthread_rwlock_t lock; + pthread_t packet_loop; } local_data; static int local_data_init(void) @@ -72,155 +71,142 @@ static int local_data_init(void) local_data.flows = fset_create(); if (local_data.flows == NULL) - return -ENFILE; + goto fail_fset; local_data.fq = fqueue_create(); - if (local_data.fq == NULL) { - fset_destroy(local_data.flows); - return -ENOMEM; - } + if (local_data.fq == NULL) + goto fail_fqueue; - local_data.shim_data = shim_data_create(); - if (local_data.shim_data == NULL) { - fqueue_destroy(local_data.fq); - fset_destroy(local_data.flows); - return -ENOMEM; - } + local_data.reg = reg_create(); + if (local_data.reg == NULL) + goto fail_reg; - pthread_rwlock_init(&local_data.lock, NULL); + if (pthread_rwlock_init(&local_data.lock, NULL) < 0) + goto fail_rwlock_init; return 0; + + fail_rwlock_init: + reg_destroy(local_data.reg); + fail_reg: + fqueue_destroy(local_data.fq); + fail_fqueue: + fset_destroy(local_data.flows); + fail_fset: + return -ENOMEM; } static void local_data_fini(void){ - shim_data_destroy(local_data.shim_data); - fset_destroy(local_data.flows); - fqueue_destroy(local_data.fq); pthread_rwlock_destroy(&local_data.lock); + reg_destroy(local_data.reg); + fqueue_destroy(local_data.fq); + fset_destroy(local_data.flows); } -static void * ipcp_local_packet_loop(void * o) +static void * local_ipcp_packet_loop(void * o) { + int src_fd; + int dst_fd; + struct timespec * timeout; +#ifdef CONFIG_IPCP_LOCAL_POLLING + struct timespec ts_poll = {0, 0}; +#endif (void) o; ipcp_lock_to_core(); - while (true) { - int fd; - ssize_t idx; +#ifdef CONFIG_IPCP_LOCAL_POLLING + timeout = &ts_poll; /* Spin poll with zero timeout */ +#else + timeout = NULL; /* Block until event */ +#endif - fevent(local_data.flows, local_data.fq, NULL); + while (true) { + fevent(local_data.flows, local_data.fq, timeout); - while ((fd = fqueue_next(local_data.fq)) >= 0) { + while ((src_fd = fqueue_next(local_data.fq)) >= 0) { if (fqueue_type(local_data.fq) != FLOW_PKT) continue; - idx = local_flow_read(fd); - if (idx < 0) - continue; - - assert(idx < (SHM_BUFFER_SIZE)); - pthread_rwlock_rdlock(&local_data.lock); - fd = local_data.in_out[fd]; + dst_fd = local_data.in_out[src_fd]; pthread_rwlock_unlock(&local_data.lock); - if (fd != -1) - local_flow_write(fd, idx); + if (dst_fd == -1) + continue; + + local_flow_transfer(src_fd, dst_fd, + NP1_GET_POOL(src_fd), + NP1_GET_POOL(dst_fd)); } } return (void *) 0; } -static int ipcp_local_bootstrap(const struct ipcp_config * conf) +static int local_ipcp_bootstrap(struct ipcp_config * conf) { + assert(conf); assert(conf->type == THIS_TYPE); (void) conf; - ipcp_set_state(IPCP_OPERATIONAL); - if (pthread_create(&local_data.packet_loop, NULL, - ipcp_local_packet_loop, NULL)) { + local_ipcp_packet_loop, NULL)) { + log_err("Failed to create pthread: %s", strerror(errno)); ipcp_set_state(IPCP_INIT); return -1; } - log_info("Bootstrapped local IPCP with pid %d.", getpid()); - return 0; } -static int ipcp_local_reg(const uint8_t * hash) +static int local_ipcp_reg(const uint8_t * hash) { - if (shim_data_reg_add_entry(local_data.shim_data, hash)) { - log_dbg("Failed to add " HASH_FMT " to local registry.", - HASH_VAL(hash)); + if (reg_add(local_data.reg, hash) < 0) { + log_err("Failed to add " HASH_FMT32 " to local registry.", + HASH_VAL32(hash)); return -1; } - log_info("Registered " HASH_FMT ".", HASH_VAL(hash)); - return 0; } -static int ipcp_local_unreg(const uint8_t * hash) +static int local_ipcp_unreg(const uint8_t * hash) { - shim_data_reg_del_entry(local_data.shim_data, hash); + reg_del(local_data.reg, hash); - log_info("Unregistered " HASH_FMT ".", HASH_VAL(hash)); + log_info("Unregistered " HASH_FMT32 ".", HASH_VAL32(hash)); return 0; } -static int ipcp_local_query(const uint8_t * hash) +static int local_ipcp_query(const uint8_t * hash) { int ret; - ret = (shim_data_reg_has(local_data.shim_data, hash) ? 0 : -1); + ret = (reg_has(local_data.reg, hash) ? 0 : -1); return ret; } -static int ipcp_local_flow_alloc(int fd, - const uint8_t * dst, - qosspec_t qs, - const void * data, - size_t len) +static int local_ipcp_flow_alloc(int fd, + const uint8_t * dst, + qosspec_t qs, + const buffer_t * data) { - struct timespec ts = {0, ALLOC_TIMEOUT * MILLION}; - struct timespec abstime; - int out_fd = -1; + int out_fd = -1; - log_dbg("Allocating flow to " HASH_FMT " on fd %d.", HASH_VAL(dst), fd); + log_dbg("Allocating flow to " HASH_FMT32 " on fd %d.", + HASH_VAL32(dst), fd); assert(dst); - clock_gettime(PTHREAD_COND_CLOCK, &abstime); - - pthread_mutex_lock(&ipcpi.alloc_lock); - - while (ipcpi.alloc_id != -1 && ipcp_get_state() == IPCP_OPERATIONAL) { - ts_add(&abstime, &ts, &abstime); - pthread_cond_timedwait(&ipcpi.alloc_cond, - &ipcpi.alloc_lock, - &abstime); - } - - if (ipcp_get_state() != IPCP_OPERATIONAL) { - log_dbg("Won't allocate over non-operational IPCP."); - pthread_mutex_unlock(&ipcpi.alloc_lock); - return -1; - } - - assert(ipcpi.alloc_id == -1); - - out_fd = ipcp_flow_req_arr(dst, ipcp_dir_hash_len(), qs, data, len); + out_fd = ipcp_wait_flow_req_arr(dst, qs, IPCP_LOCAL_MPL, + IPCP_LOCAL_MTU, data); if (out_fd < 0) { - pthread_mutex_unlock(&ipcpi.alloc_lock); log_dbg("Flow allocation failed: %d", out_fd); return -1; } @@ -232,11 +218,6 @@ static int ipcp_local_flow_alloc(int fd, pthread_rwlock_unlock(&local_data.lock); - ipcpi.alloc_id = out_fd; - pthread_cond_broadcast(&ipcpi.alloc_cond); - - pthread_mutex_unlock(&ipcpi.alloc_lock); - fset_add(local_data.flows, fd); log_info("Pending local allocation request on fd %d.", fd); @@ -244,65 +225,59 @@ static int ipcp_local_flow_alloc(int fd, return 0; } -static int ipcp_local_flow_alloc_resp(int fd, - int response, - const void * data, - size_t len) +static int local_ipcp_flow_alloc_resp(int fd, + int response, + const buffer_t * data) { - struct timespec ts = {0, ALLOC_TIMEOUT * MILLION}; - struct timespec abstime; - int out_fd = -1; - - clock_gettime(PTHREAD_COND_CLOCK, &abstime); - - pthread_mutex_lock(&ipcpi.alloc_lock); - - while (ipcpi.alloc_id != fd && ipcp_get_state() == IPCP_OPERATIONAL) { - ts_add(&abstime, &ts, &abstime); - pthread_cond_timedwait(&ipcpi.alloc_cond, - &ipcpi.alloc_lock, - &abstime); - } + struct timespec wait = TIMESPEC_INIT_MS(1); + time_t mpl = IPCP_LOCAL_MPL; + int out_fd; - if (ipcp_get_state() != IPCP_OPERATIONAL) { - pthread_mutex_unlock(&ipcpi.alloc_lock); + if (ipcp_wait_flow_resp(fd) < 0) { + log_err("Failed waiting for IRMd response."); return -1; } - ipcpi.alloc_id = -1; - pthread_cond_broadcast(&ipcpi.alloc_cond); - - pthread_mutex_unlock(&ipcpi.alloc_lock); + pthread_rwlock_rdlock(&local_data.lock); - pthread_rwlock_wrlock(&local_data.lock); - - if (response) { - if (local_data.in_out[fd] != -1) - local_data.in_out[local_data.in_out[fd]] = fd; - local_data.in_out[fd] = -1; + out_fd = local_data.in_out[fd]; + if (out_fd == -1) { pthread_rwlock_unlock(&local_data.lock); - return 0; + log_dbg("Potential race detected"); + nanosleep(&wait, NULL); + pthread_rwlock_rdlock(&local_data.lock); + out_fd = local_data.in_out[fd]; } - out_fd = local_data.in_out[fd]; + pthread_rwlock_unlock(&local_data.lock); + if (out_fd == -1) { - pthread_rwlock_unlock(&local_data.lock); + log_err("Invalid out_fd."); return -1; } - pthread_rwlock_unlock(&local_data.lock); + if (response < 0) { + ipcp_flow_alloc_reply(out_fd, response, mpl, + IPCP_LOCAL_MTU, data); + log_info("Flow allocation rejected, fds (%d, %d).", out_fd, fd); + return 0; + } fset_add(local_data.flows, fd); - if (ipcp_flow_alloc_reply(out_fd, response, data, len) < 0) + if (ipcp_flow_alloc_reply(out_fd, response, mpl, + IPCP_LOCAL_MTU, data) < 0) { + log_err("Failed to reply to allocation"); + fset_del(local_data.flows, fd); return -1; + } log_info("Flow allocation completed, fds (%d, %d).", out_fd, fd); return 0; } -static int ipcp_local_flow_dealloc(int fd) +static int local_ipcp_flow_dealloc(int fd) { assert(!(fd < 0)); @@ -316,68 +291,95 @@ static int ipcp_local_flow_dealloc(int fd) pthread_rwlock_unlock(&local_data.lock); - flow_dealloc(fd); + ipcp_flow_dealloc(fd); log_info("Flow with fd %d deallocated.", fd); return 0; } +/* Loopback relay: deliver the update back to the peer end (same IRMd). */ +static int local_ipcp_flow_update(int fd, + const buffer_t * data) +{ + int out_fd; + int out_flow_id; + + pthread_rwlock_rdlock(&local_data.lock); + + out_fd = local_data.in_out[fd]; + + pthread_rwlock_unlock(&local_data.lock); + + if (out_fd == -1) { + log_err("Flow update on fd %d with no peer.", fd); + return -1; + } + + out_flow_id = np1_flow_id(out_fd); + if (out_flow_id < 0) { + log_err("No flow_id for peer fd %d.", out_fd); + return -1; + } + + if (ipcp_flow_update_arr(out_flow_id, data) < 0) { + log_err("Failed to relay flow update to fd %d.", out_fd); + return -1; + } + + return 0; +} + static struct ipcp_ops local_ops = { - .ipcp_bootstrap = ipcp_local_bootstrap, + .ipcp_bootstrap = local_ipcp_bootstrap, .ipcp_enroll = NULL, .ipcp_connect = NULL, .ipcp_disconnect = NULL, - .ipcp_reg = ipcp_local_reg, - .ipcp_unreg = ipcp_local_unreg, - .ipcp_query = ipcp_local_query, - .ipcp_flow_alloc = ipcp_local_flow_alloc, + .ipcp_reg = local_ipcp_reg, + .ipcp_unreg = local_ipcp_unreg, + .ipcp_query = local_ipcp_query, + .ipcp_flow_alloc = local_ipcp_flow_alloc, .ipcp_flow_join = NULL, - .ipcp_flow_alloc_resp = ipcp_local_flow_alloc_resp, - .ipcp_flow_dealloc = ipcp_local_flow_dealloc + .ipcp_flow_alloc_resp = local_ipcp_flow_alloc_resp, + .ipcp_flow_dealloc = local_ipcp_flow_dealloc, + .ipcp_flow_update = local_ipcp_flow_update }; int main(int argc, char * argv[]) { - if (ipcp_init(argc, argv, &local_ops) < 0) - goto fail_init; - if (local_data_init() < 0) { log_err("Failed to init local data."); goto fail_data_init; } - if (ipcp_boot() < 0) { - log_err("Failed to boot IPCP."); - goto fail_boot; - } + if (ipcp_init(argc, argv, &local_ops, THIS_TYPE) < 0) + goto fail_init; - if (ipcp_create_r(0)) { - log_err("Failed to notify IRMd we are initialized."); - goto fail_create_r; + if (ipcp_start() < 0) { + log_err("Failed to start IPCP."); + goto fail_start; } - ipcp_shutdown(); + ipcp_sigwait(); if (ipcp_get_state() == IPCP_SHUTDOWN) { pthread_cancel(local_data.packet_loop); pthread_join(local_data.packet_loop, NULL); } - local_data_fini(); + ipcp_stop(); ipcp_fini(); - exit(EXIT_SUCCESS); - fail_create_r: - ipcp_set_state(IPCP_NULL); - ipcp_shutdown(); - fail_boot: local_data_fini(); - fail_data_init: + + exit(EXIT_SUCCESS); + + fail_start: ipcp_fini(); fail_init: - ipcp_create_r(-1); + local_data_fini(); + fail_data_init: exit(EXIT_FAILURE); } diff --git a/src/ipcpd/local/reg.c b/src/ipcpd/local/reg.c new file mode 100644 index 00000000..36f19b16 --- /dev/null +++ b/src/ipcpd/local/reg.c @@ -0,0 +1,217 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2026 + * + * Names registered with the local IPCP + * + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License version 2 as + * published by the Free Software Foundation. + * + * 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., http://www.fsf.org/about/contact/. + */ + +#if defined(__linux__) || defined(__CYGWIN__) +#define _DEFAULT_SOURCE +#else +#define _POSIX_C_SOURCE 200112L +#endif + +#define OUROBOROS_PREFIX "local-reg" + +#include <ouroboros/hash.h> +#include <ouroboros/list.h> +#include <ouroboros/logs.h> + +#include "reg.h" +#include "ipcp.h" + +#include <assert.h> +#include <pthread.h> +#include <stdlib.h> +#include <string.h> + +struct reg_entry { + struct list_head list; + uint8_t * hash; +}; + +struct reg { + struct list_head names; + pthread_rwlock_t lock; +}; + +static struct reg_entry * reg_entry_create(uint8_t * hash) +{ + struct reg_entry * entry; + + entry = malloc(sizeof(*entry)); + if (entry == NULL) + return NULL; + + list_head_init(&entry->list); + + entry->hash = hash; + + return entry; +} + +static void reg_entry_destroy(struct reg_entry * entry) +{ + assert(entry); + + free(entry->hash); + free(entry); +} + +/* Call with the lock held. */ +static struct reg_entry * reg_find(struct reg * reg, + const uint8_t * hash) +{ + struct list_head * p; + + list_for_each(p, ®->names) { + struct reg_entry * e; + + e = list_entry(p, struct reg_entry, list); + if (memcmp(e->hash, hash, ipcp_dir_hash_len()) == 0) + return e; + } + + return NULL; +} + +struct reg * reg_create(void) +{ + struct reg * reg; + + reg = malloc(sizeof(*reg)); + if (reg == NULL) + goto fail_malloc; + + list_head_init(®->names); + + if (pthread_rwlock_init(®->lock, NULL) < 0) + goto fail_lock; + + return reg; + + fail_lock: + free(reg); + fail_malloc: + return NULL; +} + +void reg_destroy(struct reg * reg) +{ + if (reg == NULL) + return; + + pthread_rwlock_wrlock(®->lock); + + while (!list_is_empty(®->names)) { + struct reg_entry * e; + + e = list_first_entry(®->names, struct reg_entry, list); + + list_del(&e->list); + + reg_entry_destroy(e); + } + + pthread_rwlock_unlock(®->lock); + + pthread_rwlock_destroy(®->lock); + + free(reg); +} + +int reg_add(struct reg * reg, + const uint8_t * hash) +{ + struct reg_entry * entry; + uint8_t * dup; + + assert(reg); + assert(hash); + + pthread_rwlock_wrlock(®->lock); + + if (reg_find(reg, hash) != NULL) { + pthread_rwlock_unlock(®->lock); + log_dbg(HASH_FMT32 " was already registered.", + HASH_VAL32(hash)); + return 0; + } + + dup = ipcp_hash_dup(hash); + if (dup == NULL) + goto fail; + + entry = reg_entry_create(dup); + if (entry == NULL) { + free(dup); + goto fail; + } + + list_add(&entry->list, ®->names); + + pthread_rwlock_unlock(®->lock); + + return 0; + + fail: + pthread_rwlock_unlock(®->lock); + return -1; +} + +int reg_del(struct reg * reg, + const uint8_t * hash) +{ + struct reg_entry * e; + + if (reg == NULL) + return -1; + + pthread_rwlock_wrlock(®->lock); + + e = reg_find(reg, hash); + if (e == NULL) { + pthread_rwlock_unlock(®->lock); + return 0; /* nothing to do */ + } + + list_del(&e->list); + + pthread_rwlock_unlock(®->lock); + + reg_entry_destroy(e); + + return 0; +} + +bool reg_has(struct reg * reg, + const uint8_t * hash) +{ + bool ret; + + assert(reg); + assert(hash); + + pthread_rwlock_rdlock(®->lock); + + ret = reg_find(reg, hash) != NULL; + + pthread_rwlock_unlock(®->lock); + + return ret; +} diff --git a/src/ipcpd/local/reg.h b/src/ipcpd/local/reg.h new file mode 100644 index 00000000..2c6142bb --- /dev/null +++ b/src/ipcpd/local/reg.h @@ -0,0 +1,45 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2026 + * + * Names registered with the local IPCP + * + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License version 2 as + * published by the Free Software Foundation. + * + * 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., http://www.fsf.org/about/contact/. + */ + +#ifndef OUROBOROS_IPCPD_LOCAL_REG_H +#define OUROBOROS_IPCPD_LOCAL_REG_H + +#include <stdbool.h> +#include <stdint.h> + +/* The hashes of the names registered with this IPCP. */ +struct reg; + +struct reg * reg_create(void); + +void reg_destroy(struct reg * reg); + +int reg_add(struct reg * reg, + const uint8_t * hash); + +int reg_del(struct reg * reg, + const uint8_t * hash); + +bool reg_has(struct reg * reg, + const uint8_t * hash); + +#endif /* OUROBOROS_IPCPD_LOCAL_REG_H */ |