diff options
Diffstat (limited to 'src/lib')
52 files changed, 4731 insertions, 1956 deletions
diff --git a/src/lib/.gitignore b/src/lib/.gitignore index 8704469b..3ecaf66d 100644 --- a/src/lib/.gitignore +++ b/src/lib/.gitignore @@ -1 +1 @@ -*.pb-c.[ch]
\ No newline at end of file +*.pb-c.[ch] diff --git a/src/lib/CMakeLists.txt b/src/lib/CMakeLists.txt index c5be9946..a6d7ac98 100644 --- a/src/lib/CMakeLists.txt +++ b/src/lib/CMakeLists.txt @@ -4,13 +4,18 @@ include_directories(${CMAKE_CURRENT_BINARY_DIR}) include_directories(${CMAKE_SOURCE_DIR}/include) include_directories(${CMAKE_BINARY_DIR}/include) -protobuf_generate_c(IRM_PROTO_SRCS IRM_PROTO_HDRS irmd_messages.proto) -protobuf_generate_c(IPCP_PROTO_SRCS IPCP_PROTO_HDRS ipcpd_messages.proto) -protobuf_generate_c(QOSSPEC_PROTO_SRCS QOSSPEC_PROTO_HDRS - qosspec.proto) -protobuf_generate_c(LAYER_CONFIG_PROTO_SRCS LAYER_CONFIG_PROTO_HDRS - ipcp_config.proto) -protobuf_generate_c(CACEP_PROTO_SRCS CACEP_PROTO_HDRS cacep.proto) +protobuf_generate_c(MODEL_PROTO_SRCS MODEL_PROTO_HDRS + pb/model.proto) +protobuf_generate_c(IPCP_CONFIG_PROTO_SRCS IPCP_CONFIG_PROTO_HDRS + pb/ipcp_config.proto) +protobuf_generate_c(ENROLL_PROTO_SRCS ENROLL_PROTO_HDRS + pb/enroll.proto) +protobuf_generate_c(CEP_PROTO_SRCS CEP_PROTO_HDRS + pb/cep.proto) +protobuf_generate_c(IRM_PROTO_SRCS IRM_PROTO_HDRS + pb/irm.proto) +protobuf_generate_c(IPCP_PROTO_SRCS IPCP_PROTO_HDRS + pb/ipcp.proto) if (NOT APPLE) find_library(LIBRT_LIBRARIES rt) @@ -118,6 +123,7 @@ endif () if (NOT HAVE_OPENSSL_RNG) set(OPENSSL_INCLUDE_DIR "") set(OPENSSL_LIBRARIES "") + set(OPENSSL_CRYPTO_LIBRARY "") endif () if (APPLE OR CMAKE_SYSTEM_NAME STREQUAL "FreeBSD") @@ -140,10 +146,10 @@ if (NOT ((CMAKE_SYSTEM_NAME STREQUAL "FreeBSD") OR APPLE OR endif () mark_as_advanced(LIBRT_LIBRARIES LIBPTHREAD_LIBRARIES - LIBGCRYPT_LIBRARIES OPENSSL_LIBRARIES SYS_RND_INCLUDE_DIR - LIBGCRYPT_INCLUDE_DIR SYS_RND_HDR) + LIBGCRYPT_LIBRARIES OPENSSL_LIBRARIES OPENSSL_CRYPTO_LIBRARY + SYS_RND_INCLUDE_DIR LIBGCRYPT_INCLUDE_DIR SYS_RND_HDR) -set(SHM_BUFFER_SIZE 4096 CACHE STRING +set(SHM_BUFFER_SIZE 16384 CACHE STRING "Number of blocks in packet buffer, must be a power of 2") set(SHM_RBUFF_SIZE 1024 CACHE STRING "Number of blocks in rbuff buffer, must be a power of 2") @@ -166,7 +172,7 @@ else () set(PTHREAD_COND_CLOCK "CLOCK_REALTIME" CACHE INTERNAL "Clock to use for condition variable timing") endif () -set(SOCKET_TIMEOUT 1000 CACHE STRING +set(SOCKET_TIMEOUT 500 CACHE STRING "Default timeout for responses from IPCPs (ms)") set(SHM_PREFIX "ouroboros" CACHE STRING "String to prepend to POSIX shared memory filenames") @@ -180,16 +186,62 @@ set(SHM_RDRB_NAME "/${SHM_PREFIX}.rdrb" CACHE INTERNAL "Name for the main POSIX shared memory buffer") set(SHM_RDRB_BLOCK_SIZE "sysconf(_SC_PAGESIZE)" CACHE STRING "Packet buffer block size, multiple of pagesize for performance") -set(SHM_RDRB_MULTI_BLOCK true CACHE BOOL +set(SHM_RDRB_MULTI_BLOCK TRUE CACHE BOOL "Packet buffer multiblock packet support") -set(SHM_RBUFF_LOCKLESS 0 CACHE BOOL +set(SHM_RBUFF_LOCKLESS FALSE CACHE BOOL "Enable shared memory lockless rbuff support") set(QOS_DISABLE_CRC TRUE CACHE BOOL "Ignores ber setting on all QoS cubes") +set(DELTA_T_MPL 60 CACHE STRING + "Maximum packet lifetime (s)") +set(DELTA_T_ACK 10 CACHE STRING + "Maximum time to acknowledge a packet (s)") +set(DELTA_T_RTX 120 CACHE STRING + "Maximum time to retransmit a packet (s)") +set(FRCT_REORDER_QUEUE_SIZE 256 CACHE STRING + "Size of the reordering queue, must be a power of 2") +set(FRCT_START_WINDOW 64 CACHE STRING + "Start window, must be a power of 2") +set(FRCT_LINUX_RTT_ESTIMATOR TRUE CACHE BOOL + "Use Linux RTT estimator formula instead of the TCP RFC formula") +set(FRCT_RTO_MDEV_MULTIPLIER 2 CACHE STRING + "Multiplier for deviation term in the RTO: RTO = sRTT + (mdev << X)") +set(FRCT_RTO_INC_FACTOR 0 CACHE STRING + "Divisor for RTO increase after timeout: RTO += RTX >> X, 0: Karn/Partridge") +set(FRCT_RTO_MIN 250 CACHE STRING + "Minimum Retransmission Timeout (RTO) for FRCT (us)") +set(FRCT_TICK_TIME 5000 CACHE STRING + "Tick time for FRCT activity (retransmission, acknowledgments) (us)") +set(RXM_BUFFER_ON_HEAP FALSE CACHE BOOL + "Store packets for retransmission on the heap instead of in packet buffer") +set(RXM_BLOCKING TRUE CACHE BOOL + "Use blocking writes for retransmission") +set(RXM_MIN_RESOLUTION 20 CACHE STRING + "Minimum retransmission delay (ns), as a power to 2") +set(RXM_WHEEL_MULTIPLIER 4 CACHE STRING + "Factor for retransmission wheel levels as a power to 2") +set(RXM_WHEEL_LEVELS 3 CACHE STRING + "Number of levels in the retransmission wheel") +set(RXM_WHEEL_SLOTS_PER_LEVEL 256 CACHE STRING + "Number of slots per level in the retransmission wheel, must be a power of 2") +set(ACK_WHEEL_SLOTS 256 CACHE STRING + "Number of slots in the acknowledgment wheel, must be a power of 2") +set(ACK_WHEEL_RESOLUTION 18 CACHE STRING + "Minimum acknowledgment delay (ns), as a power to 2") + +if (HAVE_FUSE) + set(PROC_FLOW_STATS TRUE CACHE BOOL + "Enable flow statistics tracking for application flows") + if (PROC_FLOW_STATS) + message(STATUS "Application flow statistics enabled") + else () + message(STATUS "Application flow statistics disabled") + endif () +endif () set(SOURCE_FILES_DEV # Add source files here - cacep.c + cep.c dev.c ) @@ -201,15 +253,19 @@ set(SOURCE_FILES_COMMON bitmap.c btree.c crc32.c + crypt.c hash.c list.c lockfile.c logs.c md5.c notifier.c + protobuf.c qoscube.c random.c rib.c + serdes-irm.c + serdes-oep.c sha3.c shm_flow_set.c shm_rbuff.c @@ -223,12 +279,23 @@ configure_file("${CMAKE_CURRENT_SOURCE_DIR}/config.h.in" "${CMAKE_CURRENT_BINARY_DIR}/config.h" @ONLY) add_library(ouroboros-common SHARED ${SOURCE_FILES_COMMON} ${IRM_PROTO_SRCS} - ${IPCP_PROTO_SRCS} ${LAYER_CONFIG_PROTO_SRCS} ${QOSSPEC_PROTO_SRCS}) + ${IPCP_PROTO_SRCS} ${IPCP_CONFIG_PROTO_SRCS} ${MODEL_PROTO_SRCS} + ${ENROLL_PROTO_SRCS}) -add_library(ouroboros-dev SHARED ${SOURCE_FILES_DEV} ${CACEP_PROTO_SRCS}) +add_library(ouroboros-dev SHARED ${SOURCE_FILES_DEV} ${CEP_PROTO_SRCS}) add_library(ouroboros-irm SHARED ${SOURCE_FILES_IRM}) +set_target_properties(ouroboros-common PROPERTIES + VERSION ${PACKAGE_VERSION} + SOVERSION ${PACKAGE_VERSION_MAJOR}.${PACKAGE_VERSION_MINOR}) +set_target_properties(ouroboros-dev PROPERTIES + VERSION ${PACKAGE_VERSION} + SOVERSION ${PACKAGE_VERSION_MAJOR}.${PACKAGE_VERSION_MINOR}) +set_target_properties(ouroboros-irm PROPERTIES + VERSION ${PACKAGE_VERSION} + SOVERSION ${PACKAGE_VERSION_MAJOR}.${PACKAGE_VERSION_MINOR}) + include(AddCompileFlags) if (CMAKE_BUILD_TYPE MATCHES "Debug*") add_compile_flags(ouroboros-common -DCONFIG_OUROBOROS_DEBUG) @@ -237,7 +304,7 @@ if (CMAKE_BUILD_TYPE MATCHES "Debug*") endif () target_link_libraries(ouroboros-common ${LIBRT_LIBRARIES} - ${LIBPTHREAD_LIBRARIES} ${PROTOBUF_C_LIBRARY} ${OPENSSL_LIBRARIES} + ${LIBPTHREAD_LIBRARIES} ${PROTOBUF_C_LIBRARY} ${OPENSSL_CRYPTO_LIBRARY} ${LIBGCRYPT_LIBRARIES} ${FUSE_LIBRARIES}) target_link_libraries(ouroboros-dev ouroboros-common) diff --git a/src/lib/bitmap.c b/src/lib/bitmap.c index b76960da..b0840c44 100644 --- a/src/lib/bitmap.c +++ b/src/lib/bitmap.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Bitmap implementation * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License diff --git a/src/lib/btree.c b/src/lib/btree.c index 86d1299b..1af94b73 100644 --- a/src/lib/btree.c +++ b/src/lib/btree.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * B-trees * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License diff --git a/src/lib/cacep.c b/src/lib/cep.c index 9a889530..ba238023 100644 --- a/src/lib/cacep.c +++ b/src/lib/cep.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * - * The Common Application Connection Establishment Protocol + * The Ouroboros Connection Establishment Protocol * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License @@ -22,35 +22,35 @@ #define _POSIX_C_SOURCE 199309L -#include <ouroboros/cacep.h> +#include <ouroboros/cep.h> #include <ouroboros/dev.h> #include <ouroboros/errno.h> #include <stdlib.h> #include <string.h> -#include "cacep.pb-c.h" -typedef CacepMsg cacep_msg_t; +#include "cep.pb-c.h" +typedef CepMsg cep_msg_t; #define BUF_SIZE 128 static int read_msg(int fd, struct conn_info * info) { - uint8_t buf[BUF_SIZE]; - cacep_msg_t * msg; - ssize_t len; + uint8_t buf[BUF_SIZE]; + cep_msg_t * msg; + ssize_t len; len = flow_read(fd, buf, BUF_SIZE); if (len < 0) - return -1; + return (int) len; - msg = cacep_msg__unpack(NULL, len, buf); + msg = cep_msg__unpack(NULL, len, buf); if (msg == NULL) return -1; - if (strlen(msg->comp_name) > CACEP_BUF_STRLEN) { - cacep_msg__free_unpacked(msg, NULL); + if (strlen(msg->comp_name) > OCEP_BUF_STRLEN) { + cep_msg__free_unpacked(msg, NULL); return -1; } @@ -61,7 +61,7 @@ static int read_msg(int fd, info->pref_syntax = msg->pref_syntax; info->addr = msg->address; - cacep_msg__free_unpacked(msg, NULL); + cep_msg__free_unpacked(msg, NULL); return 0; } @@ -69,9 +69,9 @@ static int read_msg(int fd, static int send_msg(int fd, const struct conn_info * info) { - cacep_msg_t msg = CACEP_MSG__INIT; - uint8_t * data = NULL; - size_t len = 0; + cep_msg_t msg = CEP_MSG__INIT; + uint8_t * data = NULL; + size_t len = 0; msg.comp_name = (char *) info->comp_name; msg.protocol = (char *) info->protocol; @@ -81,7 +81,7 @@ static int send_msg(int fd, if (msg.pref_syntax < 0) return -1; - len = cacep_msg__get_packed_size(&msg); + len = cep_msg__get_packed_size(&msg); if (len == 0) return -1; @@ -89,7 +89,7 @@ static int send_msg(int fd, if (data == NULL) return -ENOMEM; - cacep_msg__pack(&msg, data); + cep_msg__pack(&msg, data); if (flow_write(fd, data, len) < 0) { free(data); @@ -101,26 +101,20 @@ static int send_msg(int fd, return 0; } -int cacep_snd(int fd, - const struct conn_info * in) +int cep_snd(int fd, + const struct conn_info * in) { if (in == NULL) return -EINVAL; - if (send_msg(fd, in)) - return -1; - - return 0; + return send_msg(fd, in); } -int cacep_rcv(int fd, - struct conn_info * out) +int cep_rcv(int fd, + struct conn_info * out) { if (out == NULL) return -EINVAL; - if (read_msg(fd, out)) - return -1; - - return 0; + return read_msg(fd, out); } diff --git a/src/lib/config.h.in b/src/lib/config.h.in index b6e49100..604038b4 100644 --- a/src/lib/config.h.in +++ b/src/lib/config.h.in @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Ouroboros library configuration * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License @@ -42,6 +42,7 @@ #define SHM_RDRB_BLOCK_SIZE @SHM_RDRB_BLOCK_SIZE@ #define SHM_BUFFER_SIZE @SHM_BUFFER_SIZE@ #define SHM_RBUFF_SIZE @SHM_RBUFF_SIZE@ +#define FLOW_ALLOC_TIMEOUT @FLOW_ALLOC_TIMEOUT@ #if defined(__linux__) || (defined(__MACH__) && !defined(__APPLE__)) /* Avoid a bug in robust mutex implementation of glibc 2.25 */ @@ -55,7 +56,8 @@ #cmakedefine HAVE_FUSE #ifdef HAVE_FUSE -#define FUSE_PREFIX "@FUSE_PREFIX@" +#define FUSE_PREFIX "@FUSE_PREFIX@" +#cmakedefine PROC_FLOW_STATS #endif #define PTHREAD_COND_CLOCK @PTHREAD_COND_CLOCK@ @@ -66,3 +68,28 @@ #define DU_BUFF_HEADSPACE @DU_BUFF_HEADSPACE@ #define DU_BUFF_TAILSPACE @DU_BUFF_TAILSPACE@ + +/* Default Delta-t parameters */ +#cmakedefine FRCT_LINUX_RTT_ESTIMATOR +#define DELT_A (@DELTA_T_ACK@) /* ns */ +#define DELT_R (@DELTA_T_RTX@) /* ns */ + +#define RQ_SIZE (@FRCT_REORDER_QUEUE_SIZE@) +#define START_WINDOW (@FRCT_START_WINDOW@) +#define RTO_MIN (@FRCT_RTO_MIN@ * 1000) +#define RTO_DIV (@FRCT_RTO_INC_FACTOR@) +#define MDEV_MUL (@FRCT_RTO_MDEV_MULTIPLIER@) + +#define TICTIME (@FRCT_TICK_TIME@ * 1000) /* ns */ + +/* Retransmission tuning */ +#cmakedefine RXM_BUFFER_ON_HEAP +#cmakedefine RXM_BLOCKING + +#define RXMQ_RES (@RXM_MIN_RESOLUTION@) /* 2^N ns */ +#define RXMQ_BUMP (@RXM_WHEEL_MULTIPLIER@) +#define RXMQ_LVLS (@RXM_WHEEL_LEVELS@) +#define RXMQ_SLOTS (@RXM_WHEEL_SLOTS_PER_LEVEL@) + +#define ACKQ_SLOTS (@ACK_WHEEL_SLOTS@) +#define ACKQ_RES (@ACK_WHEEL_RESOLUTION@) /* 2^N ns */ diff --git a/src/lib/crc32.c b/src/lib/crc32.c index 1b8a5bc8..f369ad20 100644 --- a/src/lib/crc32.c +++ b/src/lib/crc32.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * 32-bit Cyclic Redundancy Check * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License diff --git a/src/lib/crypt.c b/src/lib/crypt.c index 1fa12bb8..ad679501 100644 --- a/src/lib/crypt.c +++ b/src/lib/crypt.c @@ -1,5 +1,5 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Elliptic curve Diffie-Hellman key exchange and * AES encryption for flows using OpenSSL @@ -20,9 +20,19 @@ * License along with this library; if not, write to the Free Software * Foundation, Inc., http://www.fsf.org/about/contact/. */ +#include <config.h> + +#include <ouroboros/crypt.h> +#include <ouroboros/errno.h> + +#include <assert.h> +#include <string.h> #ifdef HAVE_OPENSSL +#include <ouroboros/hash.h> +#include <ouroboros/random.h> + #include <openssl/evp.h> #include <openssl/ec.h> #include <openssl/pem.h> @@ -144,8 +154,8 @@ static int __openssl_ecdh_gen_key(void ** kp) static ssize_t openssl_ecdh_pkp_create(void ** pkp, uint8_t * pk) { - uint8_t * pos; - ssize_t len; + uint8_t * pos; + ssize_t len; assert(pkp != NULL); assert(*pkp == NULL); @@ -172,15 +182,14 @@ static void openssl_ecdh_pkp_destroy(void * pkp) } static int openssl_ecdh_derive(void * pkp, - uint8_t * pk, - size_t len, + buffer_t pk, uint8_t * s) { uint8_t * pos; EVP_PKEY * pub; - pos = pk; /* d2i_PUBKEY increments the pointer, don't use key ptr! */ - pub = d2i_PUBKEY(NULL, (const uint8_t **) &pos, (long) len); + pos = pk.data; /* d2i_PUBKEY increments the pointer, don't use key ptr! */ + pub = d2i_PUBKEY(NULL, (const uint8_t **) &pos, (long) pk.len); if (pub == NULL) return -ECRYPT; @@ -202,7 +211,8 @@ static int openssl_ecdh_derive(void * pkp, * encryption context for each packet. */ -static int openssl_encrypt(struct flow * f, +static int openssl_encrypt(void * ctx, + uint8_t * key, struct shm_du_buff * sdb) { uint8_t * out; @@ -217,6 +227,8 @@ static int openssl_encrypt(struct flow * f, in = shm_du_buff_head(sdb); in_sz = shm_du_buff_tail(sdb) - in; + assert(in_sz > 0); + if (random_buffer(iv, IVSZ) < 0) goto fail_iv; @@ -224,28 +236,24 @@ static int openssl_encrypt(struct flow * f, if (out == NULL) goto fail_iv; - EVP_CIPHER_CTX_reset(f->ctx); + EVP_CIPHER_CTX_reset(ctx); - ret = EVP_EncryptInit_ex(f->ctx, - EVP_aes_256_cbc(), - NULL, - f->key, - iv); + ret = EVP_EncryptInit_ex(ctx, EVP_aes_256_cbc(), NULL, key, iv); if (ret != 1) goto fail_encrypt_init; - ret = EVP_EncryptUpdate(f->ctx, out, &tmp_sz, in, in_sz); + ret = EVP_EncryptUpdate(ctx, out, &tmp_sz, in, in_sz); if (ret != 1) goto fail_encrypt; out_sz = tmp_sz; - ret = EVP_EncryptFinal_ex(f->ctx, out + tmp_sz, &tmp_sz); + ret = EVP_EncryptFinal_ex(ctx, out + tmp_sz, &tmp_sz); if (ret != 1) goto fail_encrypt; out_sz += tmp_sz; - EVP_CIPHER_CTX_cleanup(f->ctx); + EVP_CIPHER_CTX_cleanup(ctx); assert(out_sz >= in_sz); @@ -266,14 +274,15 @@ static int openssl_encrypt(struct flow * f, fail_tail_alloc: shm_du_buff_head_release(sdb, IVSZ); fail_encrypt: - EVP_CIPHER_CTX_cleanup(f->ctx); + EVP_CIPHER_CTX_cleanup(ctx); fail_encrypt_init: free(out); fail_iv: return -ECRYPT; } -static int openssl_decrypt(struct flow * f, +static int openssl_decrypt(void * ctx, + uint8_t * key, struct shm_du_buff * sdb) { uint8_t * in; @@ -284,35 +293,34 @@ static int openssl_decrypt(struct flow * f, int in_sz; int tmp_sz; + in_sz = shm_du_buff_len(sdb); + if (in_sz < IVSZ) + return -ECRYPT; + in = shm_du_buff_head_release(sdb, IVSZ); memcpy(iv, in, IVSZ); in = shm_du_buff_head(sdb); - - in_sz = shm_du_buff_tail(sdb) - shm_du_buff_head(sdb); + in_sz = shm_du_buff_tail(sdb) - in; out = malloc(in_sz); if (out == NULL) goto fail_malloc; - EVP_CIPHER_CTX_reset(f->ctx); + EVP_CIPHER_CTX_reset(ctx); - ret = EVP_DecryptInit_ex(f->ctx, - EVP_aes_256_cbc(), - NULL, - f->key, - iv); + ret = EVP_DecryptInit_ex(ctx, EVP_aes_256_cbc(), NULL, key, iv); if (ret != 1) goto fail_decrypt_init; - ret = EVP_DecryptUpdate(f->ctx, out, &tmp_sz, in, in_sz); + ret = EVP_DecryptUpdate(ctx, out, &tmp_sz, in, in_sz); if (ret != 1) goto fail_decrypt; out_sz = tmp_sz; - ret = EVP_DecryptFinal_ex(f->ctx, out + tmp_sz, &tmp_sz); + ret = EVP_DecryptFinal_ex(ctx, out + tmp_sz, &tmp_sz); if (ret != 1) goto fail_decrypt; @@ -329,7 +337,7 @@ static int openssl_decrypt(struct flow * f, return 0; fail_decrypt: - EVP_CIPHER_CTX_cleanup(f->ctx); + EVP_CIPHER_CTX_cleanup(ctx); fail_decrypt_init: free(out); fail_malloc: @@ -353,8 +361,8 @@ static void openssl_crypt_fini(void * ctx) #endif /* HAVE_OPENSSL */ -static int crypt_dh_pkp_create(void ** pkp, - uint8_t * pk) +int crypt_dh_pkp_create(void ** pkp, + uint8_t * pk) { #ifdef HAVE_OPENSSL assert(pkp != NULL); @@ -364,33 +372,31 @@ static int crypt_dh_pkp_create(void ** pkp, (void) pkp; (void) pk; - memset(pk, 0, MSGBUFSZ); + *pkp = NULL; - return -ECRYPT; + return 0; #endif } -static void crypt_dh_pkp_destroy(void * pkp) +void crypt_dh_pkp_destroy(void * pkp) { #ifdef HAVE_OPENSSL openssl_ecdh_pkp_destroy(pkp); #else (void) pkp; - return 0; + return; #endif } -static int crypt_dh_derive(void * pkp, - uint8_t * pk, - size_t len, - uint8_t * s) +int crypt_dh_derive(void * pkp, + buffer_t pk, + uint8_t * s) { #ifdef HAVE_OPENSSL - return openssl_ecdh_derive(pkp, pk, len, s); + return openssl_ecdh_derive(pkp, pk, s); #else (void) pkp; (void) pk; - (void) len; memset(s, 0, SYMMKEYSZ); @@ -398,50 +404,52 @@ static int crypt_dh_derive(void * pkp, #endif } -static int crypt_encrypt(struct flow * f, - struct shm_du_buff * sdb) +int crypt_encrypt(struct crypt_info * info, + struct shm_du_buff * sdb) { + if (info->flags == 0) + return 0; + #ifdef HAVE_OPENSSL - return openssl_encrypt(f, sdb); + return openssl_encrypt(info->ctx, info->key, sdb); #else - (void) f; (void) sdb; return 0; #endif } -static int crypt_decrypt(struct flow * f, - struct shm_du_buff * sdb) +int crypt_decrypt(struct crypt_info * info, + struct shm_du_buff * sdb) { + if (info->flags == 0) + return 0; + #ifdef HAVE_OPENSSL - return openssl_decrypt(f, sdb); + return openssl_decrypt(info->ctx, info->key, sdb); #else - (void) f; (void) sdb; return -ECRYPT; #endif } -static int crypt_init(void ** ctx) +int crypt_init(struct crypt_info * info) { #ifdef HAVE_OPENSSL - return openssl_crypt_init(ctx); + return openssl_crypt_init(&info->ctx); #else - assert(ctx != NULL); - *ctx = NULL; - + info->ctx = NULL; return 0; #endif } -static void crypt_fini(void * ctx) +void crypt_fini(struct crypt_info * info) { #ifdef HAVE_OPENSSL - openssl_crypt_fini(ctx); + openssl_crypt_fini(info->ctx); #else - assert(ctx == NULL); - (void) ctx; + (void) info; + assert(info->ctx == NULL); #endif } diff --git a/src/lib/dev.c b/src/lib/dev.c index 80d7e9ad..92310b9e 100644 --- a/src/lib/dev.c +++ b/src/lib/dev.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * API for applications * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License @@ -28,22 +28,36 @@ #include "config.h" -#include <ouroboros/hash.h> -#include <ouroboros/cacep.h> -#include <ouroboros/errno.h> +#include <ouroboros/bitmap.h> +#include <ouroboros/cep.h> +#include <ouroboros/crypt.h> #include <ouroboros/dev.h> +#include <ouroboros/errno.h> +#include <ouroboros/fccntl.h> +#include <ouroboros/flow.h> +#include <ouroboros/fqueue.h> +#include <ouroboros/hash.h> +#include <ouroboros/ipcp.h> #include <ouroboros/ipcp-dev.h> +#include <ouroboros/list.h> #include <ouroboros/local-dev.h> -#include <ouroboros/sockets.h> -#include <ouroboros/fccntl.h> -#include <ouroboros/bitmap.h> +#include <ouroboros/np1_flow.h> +#include <ouroboros/pthread.h> #include <ouroboros/random.h> +#include <ouroboros/serdes-irm.h> #include <ouroboros/shm_flow_set.h> #include <ouroboros/shm_rdrbuff.h> #include <ouroboros/shm_rbuff.h> +#include <ouroboros/sockets.h> #include <ouroboros/utils.h> -#include <ouroboros/fqueue.h> +#ifdef PROC_FLOW_STATS +#include <ouroboros/rib.h> +#endif +#ifdef HAVE_LIBGCRYPT +#include <gcrypt.h> +#endif +#include <assert.h> #include <stdlib.h> #include <string.h> #include <stdio.h> @@ -61,51 +75,34 @@ #define CRCLEN (sizeof(uint32_t)) #define SECMEMSZ 16384 -#define SYMMKEYSZ 32 #define MSGBUFSZ 2048 -struct flow_set { - size_t idx; -}; - -struct fqueue { - int fqueue[2 * SHM_BUFFER_SIZE]; /* Safe copy from shm. */ - size_t fqsize; - size_t next; -}; - -enum port_state { - PORT_NULL = 0, - PORT_INIT, - PORT_ID_PENDING, - PORT_ID_ASSIGNED, - PORT_DESTROY -}; - -struct port { +/* map flow_ids to flow descriptors; track state of the flow */ +struct fmap { int fd; - - enum port_state state; - pthread_mutex_t state_lock; - pthread_cond_t state_cond; + /* TODO: use actual flow state */ + enum flow_state state; }; #define frcti_to_flow(frcti) \ ((struct flow *)((uint8_t *) frcti - offsetof(struct flow, frcti))) struct flow { + struct list_head next; + + struct flow_info info; + struct shm_rbuff * rx_rb; struct shm_rbuff * tx_rb; struct shm_flow_set * set; - int flow_id; - int oflags; - qosspec_t qs; + + uint16_t oflags; ssize_t part_idx; - void * ctx; - uint8_t key[SYMMKEYSZ]; + struct crypt_info crypt; - pid_t pid; + struct timespec snd_act; + struct timespec rcv_act; bool snd_timesout; bool rcv_timesout; @@ -115,143 +112,297 @@ struct flow { struct frcti * frcti; }; -struct { - char * prog; - pid_t pid; +struct flow_set { + size_t idx; + pthread_rwlock_t lock; +}; + +struct fqueue { + struct flowevent fqueue[SHM_BUFFER_SIZE]; /* Safe copy from shm. */ + size_t fqsize; + size_t next; +}; +struct { struct shm_rdrbuff * rdrb; struct shm_flow_set * fqset; - struct timerwheel * tw; - struct bmp * fds; struct bmp * fqueues; struct flow * flows; - struct port * ports; + struct fmap * id_to_fd; + struct list_head flow_list; + + pthread_mutex_t mtx; + pthread_cond_t cond; + + pthread_t tx; + pthread_t rx; + size_t n_frcti; + fset_t * frct_set; pthread_rwlock_t lock; } ai; -#include "frct.c" - -static void port_destroy(struct port * p) +static void flow_destroy(struct fmap * p) { - pthread_mutex_lock(&p->state_lock); + pthread_mutex_lock(&ai.mtx); - if (p->state == PORT_DESTROY) { - pthread_mutex_unlock(&p->state_lock); + if (p->state == FLOW_DESTROY) { + pthread_mutex_unlock(&ai.mtx); return; } - if (p->state == PORT_ID_PENDING) - p->state = PORT_DESTROY; + if (p->state == FLOW_ALLOC_PENDING) + p->state = FLOW_DESTROY; else - p->state = PORT_NULL; + p->state = FLOW_NULL; + + pthread_cond_signal(&ai.cond); - pthread_cond_signal(&p->state_cond); + pthread_cleanup_push(__cleanup_mutex_unlock, &ai.mtx); - while (p->state != PORT_NULL) - pthread_cond_wait(&p->state_cond, &p->state_lock); + while (p->state != FLOW_NULL) + pthread_cond_wait(&ai.cond, &ai.mtx); - p->fd = -1; - p->state = PORT_INIT; + p->fd = -1; + p->state = FLOW_INIT; - pthread_mutex_unlock(&p->state_lock); + pthread_cleanup_pop(true); } -static void port_set_state(struct port * p, - enum port_state state) +static void flow_set_state(struct fmap * p, + enum flow_state state) { - pthread_mutex_lock(&p->state_lock); + pthread_mutex_lock(&ai.mtx); - if (p->state == PORT_DESTROY) { - pthread_mutex_unlock(&p->state_lock); + if (p->state == FLOW_DESTROY) { + pthread_mutex_unlock(&ai.mtx); return; } p->state = state; - pthread_cond_broadcast(&p->state_cond); + pthread_cond_broadcast(&ai.cond); - pthread_mutex_unlock(&p->state_lock); + pthread_mutex_unlock(&ai.mtx); } -static enum port_state port_wait_assign(int flow_id) +static enum flow_state flow_wait_assign(int flow_id) { - enum port_state state; - struct port * p; + enum flow_state state; + struct fmap * p; - p = &ai.ports[flow_id]; + p = &ai.id_to_fd[flow_id]; - pthread_mutex_lock(&p->state_lock); + pthread_mutex_lock(&ai.mtx); - if (p->state == PORT_ID_ASSIGNED) { - pthread_mutex_unlock(&p->state_lock); - return PORT_ID_ASSIGNED; + if (p->state == FLOW_ALLOCATED) { + pthread_mutex_unlock(&ai.mtx); + return FLOW_ALLOCATED; } - if (p->state == PORT_INIT) - p->state = PORT_ID_PENDING; + if (p->state == FLOW_INIT) + p->state = FLOW_ALLOC_PENDING; - while (p->state == PORT_ID_PENDING) - pthread_cond_wait(&p->state_cond, &p->state_lock); + pthread_cleanup_push(__cleanup_mutex_unlock, &ai.mtx); - if (p->state == PORT_DESTROY) { - p->state = PORT_NULL; - pthread_cond_broadcast(&p->state_cond); + while (p->state == FLOW_ALLOC_PENDING) + pthread_cond_wait(&ai.cond, &ai.mtx); + + if (p->state == FLOW_DESTROY) { + p->state = FLOW_NULL; + pthread_cond_broadcast(&ai.cond); } state = p->state; - assert(state != PORT_INIT); + pthread_cleanup_pop(true); - pthread_mutex_unlock(&p->state_lock); + assert(state != FLOW_INIT); return state; } -static int proc_announce(char * prog) +static int proc_announce(const char * prog) +{ + uint8_t buf[SOCK_BUF_SIZE]; + buffer_t msg = {buf, SOCK_BUF_SIZE}; + int err; + + if (proc_announce__irm_req_ser(&msg, prog) < 0) + return -ENOMEM; + + err = send_recv_msg(&msg); + if (err < 0) + return err; + + return irm__irm_result_des(&msg); +} + +/* IRMd will clean up the mess if this fails */ +static void proc_exit(void) +{ + uint8_t buf[SOCK_BUF_SIZE]; + buffer_t msg = {buf, SOCK_BUF_SIZE}; + + if (proc_exit__irm_req_ser(&msg) < 0) + return; + + send_recv_msg(&msg); +} + +#include "frct.c" + +void * flow_tx(void * o) +{ + struct timespec tic = TIMESPEC_INIT_NS(TICTIME); + + (void) o; + + while (true) { + timerwheel_move(); + + nanosleep(&tic, NULL); + } + + return (void *) 0; +} + +static void flow_send_keepalive(struct flow * flow, + struct timespec now) +{ + struct shm_du_buff * sdb; + ssize_t idx; + uint8_t * ptr; + + idx = shm_rdrbuff_alloc(ai.rdrb, 0, &ptr, &sdb); + if (idx < 0) + return; + + pthread_rwlock_wrlock(&ai.lock); + + flow->snd_act = now; + + if (shm_rbuff_write(flow->tx_rb, idx)) + shm_rdrbuff_remove(ai.rdrb, idx); + else + shm_flow_set_notify(flow->set, flow->info.id, FLOW_PKT); + + pthread_rwlock_unlock(&ai.lock); +} + +/* Needs rdlock on ai. */ +static void _flow_keepalive(struct flow * flow) { - irm_msg_t msg = IRM_MSG__INIT; - irm_msg_t * recv_msg; - int ret = -1; - - msg.code = IRM_MSG_CODE__IRM_PROC_ANNOUNCE; - msg.has_pid = true; - msg.pid = ai.pid; - msg.prog = prog; - - recv_msg = send_recv_irm_msg(&msg); - if (recv_msg == NULL) { - return -EIRMD; + struct timespec now; + struct timespec s_act; + struct timespec r_act; + int flow_id; + time_t timeo; + uint32_t acl; + + s_act = flow->snd_act; + r_act = flow->rcv_act; + + flow_id = flow->info.id; + timeo = flow->info.qs.timeout; + + acl = shm_rbuff_get_acl(flow->rx_rb); + if (timeo == 0 || acl & (ACL_FLOWPEER | ACL_FLOWDOWN)) + return; + + clock_gettime(PTHREAD_COND_CLOCK, &now); + + if (ts_diff_ns(&r_act, &now) > (int64_t) timeo * MILLION) { + shm_rbuff_set_acl(flow->rx_rb, ACL_FLOWPEER); + shm_flow_set_notify(ai.fqset, flow_id, FLOW_PEER); + return; } - if (!recv_msg->has_result || (ret = recv_msg->result)) { - irm_msg__free_unpacked(recv_msg, NULL); - return ret; + if (ts_diff_ns(&s_act, &now) > (int64_t) timeo * (MILLION >> 2)) { + pthread_rwlock_unlock(&ai.lock); + + flow_send_keepalive(flow, now); + + pthread_rwlock_rdlock(&ai.lock); } +} - irm_msg__free_unpacked(recv_msg, NULL); +static void handle_keepalives(void) +{ + struct list_head * p; + struct list_head * h; - return ret; + pthread_rwlock_rdlock(&ai.lock); + + list_for_each_safe(p, h, &ai.flow_list) { + struct flow * flow; + flow = list_entry(p, struct flow, next); + _flow_keepalive(flow); + } + + pthread_rwlock_unlock(&ai.lock); +} + +static void __cleanup_fqueue_destroy(void * fq) +{ + fqueue_destroy((fqueue_t *) fq); +} + +void * flow_rx(void * o) +{ + struct timespec tic = TIMESPEC_INIT_NS(TICTIME); + int ret; + struct fqueue * fq; + + (void) o; + + fq = fqueue_create(); + + pthread_cleanup_push(__cleanup_fqueue_destroy, fq); + + /* fevent will filter all FRCT packets for us */ + while ((ret = fevent(ai.frct_set, fq, &tic)) != 0) { + if (ret == -ETIMEDOUT) { + handle_keepalives(); + continue; + } + + while (fqueue_next(fq) >= 0) + ; /* no need to act */ + } + + pthread_cleanup_pop(true); + + return (void *) 0; } static void flow_clear(int fd) { memset(&ai.flows[fd], 0, sizeof(ai.flows[fd])); - ai.flows[fd].flow_id = -1; - ai.flows[fd].pid = -1; + ai.flows[fd].info.id = -1; } -#include "crypt.c" - -static void flow_fini(int fd) +static void __flow_fini(int fd) { assert(fd >= 0 && fd < SYS_MAX_FLOWS); - if (ai.flows[fd].flow_id != -1) { - port_destroy(&ai.ports[ai.flows[fd].flow_id]); + if (ai.flows[fd].frcti != NULL) { + ai.n_frcti--; + if (ai.n_frcti == 0) { + pthread_cancel(ai.tx); + pthread_join(ai.tx, NULL); + } + + shm_flow_set_del(ai.fqset, 0, ai.flows[fd].info.id); + + frcti_destroy(ai.flows[fd].frcti); + } + + if (ai.flows[fd].info.id != -1) { + flow_destroy(&ai.id_to_fd[ai.flows[fd].info.id]); bmp_release(ai.fds, fd); } @@ -267,27 +418,36 @@ static void flow_fini(int fd) if (ai.flows[fd].set != NULL) { shm_flow_set_notify(ai.flows[fd].set, - ai.flows[fd].flow_id, + ai.flows[fd].info.id, FLOW_DEALLOC); shm_flow_set_close(ai.flows[fd].set); } - if (ai.flows[fd].frcti != NULL) - frcti_destroy(ai.flows[fd].frcti); + crypt_fini(&ai.flows[fd].crypt); - if (ai.flows[fd].ctx != NULL) - crypt_fini(ai.flows[fd].ctx); + list_del(&ai.flows[fd].next); flow_clear(fd); } -static int flow_init(int flow_id, - pid_t pid, - qosspec_t qs, - uint8_t * s) +static void flow_fini(int fd) { - int fd; - int err = -ENOMEM; + pthread_rwlock_wrlock(&ai.lock); + + __flow_fini(fd); + + pthread_rwlock_unlock(&ai.lock); +} + +static int flow_init(struct flow_info * info, + buffer_t * sk) +{ + struct timespec now; + struct flow * flow; + int fd; + int err = -ENOMEM; + + clock_gettime(PTHREAD_COND_CLOCK, &now); pthread_rwlock_wrlock(&ai.lock); @@ -297,46 +457,75 @@ static int flow_init(int flow_id, goto fail_fds; } - ai.flows[fd].rx_rb = shm_rbuff_open(ai.pid, flow_id); - if (ai.flows[fd].rx_rb == NULL) + flow = &ai.flows[fd]; + + flow->info = *info; + + flow->rx_rb = shm_rbuff_open(info->n_pid, info->id); + if (flow->rx_rb == NULL) goto fail_rx_rb; - ai.flows[fd].tx_rb = shm_rbuff_open(pid, flow_id); - if (ai.flows[fd].tx_rb == NULL) + flow->tx_rb = shm_rbuff_open(info->n_1_pid, info->id); + if (flow->tx_rb == NULL) goto fail_tx_rb; - ai.flows[fd].set = shm_flow_set_open(pid); - if (ai.flows[fd].set == NULL) + flow->set = shm_flow_set_open(info->n_1_pid); + if (flow->set == NULL) goto fail_set; - ai.flows[fd].flow_id = flow_id; - ai.flows[fd].oflags = FLOWFDEFAULT; - ai.flows[fd].pid = pid; - ai.flows[fd].part_idx = NO_PART; - ai.flows[fd].qs = qs; + flow->oflags = FLOWFDEFAULT; + flow->part_idx = NO_PART; + flow->snd_act = now; + flow->rcv_act = now; + + flow->crypt.flags = info->qs.cypher_s; /* TODO: move cypher_s */ - if (qs.cypher_s > 0) { - assert(s != NULL); - if (crypt_init(&ai.flows[fd].ctx) < 0) - goto fail_ctx; + memset(flow->crypt.key, 0, SYMMKEYSZ); - memcpy(ai.flows[fd].key, s, SYMMKEYSZ); + if (flow->crypt.flags > 0 && sk!= NULL && sk->data != NULL) + memcpy(flow->crypt.key, sk->data , sk->len); + + if (crypt_init(&flow->crypt) < 0) + goto fail_crypt; + + assert(flow->frcti == NULL); + + if (info->qs.in_order != 0) { + flow->frcti = frcti_create(fd, DELT_A, DELT_R, info->mpl); + if (flow->frcti == NULL) + goto fail_frcti; + + if (shm_flow_set_add(ai.fqset, 0, info->id)) + goto fail_flow_set_add; + + ++ai.n_frcti; + if (ai.n_frcti == 1 && + pthread_create(&ai.tx, NULL, flow_tx, NULL) < 0) + goto fail_tx_thread; } - ai.ports[flow_id].fd = fd; + list_add_tail(&flow->next, &ai.flow_list); + + ai.id_to_fd[info->id].fd = fd; - port_set_state(&ai.ports[flow_id], PORT_ID_ASSIGNED); + flow_set_state(&ai.id_to_fd[info->id], FLOW_ALLOCATED); pthread_rwlock_unlock(&ai.lock); return fd; - fail_ctx: - shm_flow_set_close(ai.flows[fd].set); + fail_tx_thread: + shm_flow_set_del(ai.fqset, 0, info->id); + fail_flow_set_add: + frcti_destroy(flow->frcti); + fail_frcti: + crypt_fini(&flow->crypt); + fail_crypt: + shm_flow_set_close(flow->set); fail_set: - shm_rbuff_close(ai.flows[fd].tx_rb); + shm_rbuff_close(flow->tx_rb); fail_tx_rb: - shm_rbuff_close(ai.flows[fd].rx_rb); + shm_rbuff_close(flow->rx_rb); fail_rx_rb: bmp_release(ai.fds, fd); fail_fds: @@ -358,101 +547,143 @@ static void init(int argc, char ** argv, char ** envp) { - const char * prog = argv[0]; - int i; - + char * prog = argv[0]; + int i; +#ifdef PROC_FLOW_STATS + char procstr[32]; +#endif (void) argc; (void) envp; - assert(ai.prog == NULL); - if (check_python(argv[0])) prog = argv[1]; - ai.pid = getpid(); + prog = path_strip(prog); + if (prog == NULL) { + fprintf(stderr, "FATAL: Could not determine program name.\n"); + goto fail_prog; + } + + if (proc_announce(prog)) { + fprintf(stderr, "FATAL: Could not announce to IRMd.\n"); + goto fail_prog; + } + #ifdef HAVE_LIBGCRYPT if (!gcry_control(GCRYCTL_INITIALIZATION_FINISHED_P)) { - if (!gcry_check_version(GCRYPT_VERSION)) - goto fail_fds; + if (!gcry_check_version(GCRYPT_VERSION)) { + fprintf(stderr, "FATAL: Could not get gcry version.\n"); + goto fail_prog; + } gcry_control(GCRYCTL_DISABLE_SECMEM, 0); gcry_control(GCRYCTL_INITIALIZATION_FINISHED, 0); } #endif ai.fds = bmp_create(PROG_MAX_FLOWS - PROG_RES_FDS, PROG_RES_FDS); - if (ai.fds == NULL) + if (ai.fds == NULL) { + fprintf(stderr, "FATAL: Could not create fd bitmap.\n"); goto fail_fds; + } ai.fqueues = bmp_create(PROG_MAX_FQUEUES, 0); - if (ai.fqueues == NULL) + if (ai.fqueues == NULL) { + fprintf(stderr, "FATAL: Could not create fqueue bitmap.\n"); goto fail_fqueues; + } ai.rdrb = shm_rdrbuff_open(); - if (ai.rdrb == NULL) + if (ai.rdrb == NULL) { + fprintf(stderr, "FATAL: Could not open packet buffer.\n"); goto fail_rdrb; + } ai.flows = malloc(sizeof(*ai.flows) * PROG_MAX_FLOWS); - if (ai.flows == NULL) + if (ai.flows == NULL) { + fprintf(stderr, "FATAL: Could not malloc flows.\n"); goto fail_flows; + } for (i = 0; i < PROG_MAX_FLOWS; ++i) flow_clear(i); - ai.ports = malloc(sizeof(*ai.ports) * SYS_MAX_FLOWS); - if (ai.ports == NULL) - goto fail_ports; + ai.id_to_fd = malloc(sizeof(*ai.id_to_fd) * SYS_MAX_FLOWS); + if (ai.id_to_fd == NULL) { + fprintf(stderr, "FATAL: Could not malloc id_to_fd.\n"); + goto fail_id_to_fd; + } - if (prog != NULL) { - ai.prog = strdup(path_strip((char *) prog)); - if (ai.prog == NULL) - goto fail_prog; + for (i = 0; i < SYS_MAX_FLOWS; ++i) + ai.id_to_fd[i].state = FLOW_INIT; - if (proc_announce((char *) ai.prog)) - goto fail_announce; + if (pthread_mutex_init(&ai.mtx, NULL)) { + fprintf(stderr, "FATAL: Could not init mutex.\n"); + goto fail_mtx; } - for (i = 0; i < SYS_MAX_FLOWS; ++i) { - ai.ports[i].state = PORT_INIT; - if (pthread_mutex_init(&ai.ports[i].state_lock, NULL)) { - int j; - for (j = 0; j < i; ++j) - pthread_mutex_destroy(&ai.ports[j].state_lock); - goto fail_announce; - } - if (pthread_cond_init(&ai.ports[i].state_cond, NULL)) { - int j; - for (j = 0; j < i; ++j) - pthread_cond_destroy(&ai.ports[j].state_cond); - goto fail_state_cond; - } + if (pthread_cond_init(&ai.cond, NULL) < 0) { + fprintf(stderr, "FATAL: Could not init condvar.\n"); + goto fail_cond; } - if (pthread_rwlock_init(&ai.lock, NULL)) - goto fail_lock; - - if (rxmwheel_init()) - goto fail_rxmwheel; + if (pthread_rwlock_init(&ai.lock, NULL) < 0) { + fprintf(stderr, "FATAL: Could not initialize flow lock.\n"); + goto fail_flow_lock; + } ai.fqset = shm_flow_set_open(getpid()); - if (ai.fqset == NULL) + if (ai.fqset == NULL) { + fprintf(stderr, "FATAL: Could not open flow set.\n"); goto fail_fqset; + } + + ai.frct_set = fset_create(); + if (ai.frct_set == NULL || ai.frct_set->idx != 0) { + fprintf(stderr, "FATAL: Could not create FRCT set.\n"); + goto fail_frct_set; + } + + if (timerwheel_init() < 0) { + fprintf(stderr, "FATAL: Could not initialize timerwheel.\n"); + goto fail_timerwheel; + } + +#if defined PROC_FLOW_STATS + if (strstr(argv[0], "ipcpd") == NULL) { + sprintf(procstr, "proc.%d", getpid()); + if (rib_init(procstr) < 0) { + fprintf(stderr, "FATAL: Could not initialize RIB.\n"); + goto fail_rib_init; + } + } +#endif + if (pthread_create(&ai.rx, NULL, flow_rx, NULL) < 0) { + fprintf(stderr, "FATAL: Could not start monitor thread.\n"); + goto fail_monitor; + } + + list_head_init(&ai.flow_list); return; + fail_monitor: +#if defined PROC_FLOW_STATS + rib_fini(); + fail_rib_init: +#endif + timerwheel_fini(); + fail_timerwheel: + fset_destroy(ai.frct_set); + fail_frct_set: + shm_flow_set_close(ai.fqset); fail_fqset: - rxmwheel_fini(); - fail_rxmwheel: pthread_rwlock_destroy(&ai.lock); - fail_lock: - for (i = 0; i < SYS_MAX_FLOWS; ++i) - pthread_cond_destroy(&ai.ports[i].state_cond); - fail_state_cond: - for (i = 0; i < SYS_MAX_FLOWS; ++i) - pthread_mutex_destroy(&ai.ports[i].state_lock); - fail_announce: - free(ai.prog); - fail_prog: - free(ai.ports); - fail_ports: + fail_flow_lock: + pthread_cond_destroy(&ai.cond); + fail_cond: + pthread_mutex_destroy(&ai.mtx); + fail_mtx: + free(ai.id_to_fd); + fail_id_to_fd: free(ai.flows); fail_flows: shm_rdrbuff_close(ai.rdrb); @@ -461,54 +692,60 @@ static void init(int argc, fail_fqueues: bmp_destroy(ai.fds); fail_fds: - fprintf(stderr, "FATAL: ouroboros-dev init failed. " - "Make sure an IRMd is running.\n\n"); memset(&ai, 0, sizeof(ai)); + fail_prog: exit(EXIT_FAILURE); } static void fini(void) { - int i = 0; + int i; if (ai.fds == NULL) return; - rxmwheel_fini(); - - if (ai.prog != NULL) - free(ai.prog); + pthread_cancel(ai.rx); + pthread_join(ai.rx, NULL); pthread_rwlock_wrlock(&ai.lock); for (i = 0; i < PROG_MAX_FLOWS; ++i) { - if (ai.flows[i].flow_id != -1) { + if (ai.flows[i].info.id != -1) { ssize_t idx; shm_rbuff_set_acl(ai.flows[i].rx_rb, ACL_FLOWDOWN); while ((idx = shm_rbuff_read(ai.flows[i].rx_rb)) >= 0) shm_rdrbuff_remove(ai.rdrb, idx); - flow_fini(i); + __flow_fini(i); } } - shm_flow_set_close(ai.fqset); + pthread_cond_destroy(&ai.cond); + pthread_mutex_destroy(&ai.mtx); - for (i = 0; i < SYS_MAX_FLOWS; ++i) { - pthread_mutex_destroy(&ai.ports[i].state_lock); - pthread_cond_destroy(&ai.ports[i].state_cond); - } + pthread_rwlock_unlock(&ai.lock); - shm_rdrbuff_close(ai.rdrb); +#ifdef PROC_FLOW_STATS + rib_fini(); +#endif + timerwheel_fini(); + + fset_destroy(ai.frct_set); + + shm_flow_set_close(ai.fqset); + + pthread_rwlock_destroy(&ai.lock); free(ai.flows); - free(ai.ports); + free(ai.id_to_fd); + + shm_rdrbuff_close(ai.rdrb); bmp_destroy(ai.fds); bmp_destroy(ai.fqueues); - pthread_rwlock_unlock(&ai.lock); + proc_exit(); - pthread_rwlock_destroy(&ai.lock); + memset(&ai, 0, sizeof(ai)); } #if defined(__MACH__) && defined(__APPLE__) @@ -525,275 +762,241 @@ __attribute__((section(FINI_SECTION))) __typeof__(fini) * __fini = fini; int flow_accept(qosspec_t * qs, const struct timespec * timeo) { - irm_msg_t msg = IRM_MSG__INIT; - irm_msg_t * recv_msg; - int fd; - void * pkp; /* public key pair */ - uint8_t s[SYMMKEYSZ]; /* secret key for flow */ - uint8_t buf[MSGBUFSZ]; - int err = -EIRMD; - ssize_t key_len; + struct flow_info flow; + uint8_t buf[SOCK_BUF_SIZE]; + buffer_t msg = {buf, SOCK_BUF_SIZE}; + buffer_t sk; + int fd; + int err; - memset(s, 0, SYMMKEYSZ); - - msg.code = IRM_MSG_CODE__IRM_FLOW_ACCEPT; - msg.has_pid = true; - msg.pid = ai.pid; - - if (timeo != NULL) { - msg.has_timeo_sec = true; - msg.has_timeo_nsec = true; - msg.timeo_sec = timeo->tv_sec; - msg.timeo_nsec = timeo->tv_nsec; - } +#ifdef QOS_DISABLE_CRC + if (qs != NULL) + qs->ber = 1; +#endif + memset(&flow, 0, sizeof(flow)); - key_len = crypt_dh_pkp_create(&pkp, buf); - if (key_len < 0) { - err = -ECRYPT; - goto fail_crypt_pkp; - } + flow.n_pid = getpid(); + flow.qs = qs == NULL ? qos_raw : *qs; - msg.has_pk = true; - msg.pk.data = buf; - msg.pk.len = (uint32_t) key_len; + if (flow_accept__irm_req_ser(&msg, &flow, timeo)) + return -ENOMEM; - pthread_cleanup_push(crypt_dh_pkp_destroy, pkp); + err = send_recv_msg(&msg); + if (err < 0) + return err; - recv_msg = send_recv_irm_msg(&msg); + err = flow__irm_result_des(&msg, &flow, &sk); + if (err < 0) + return err; - pthread_cleanup_pop(false); + fd = flow_init(&flow, &sk); - if (recv_msg == NULL) - goto fail_recv; + freebuf(sk); - if (!recv_msg->has_result) - goto fail_result; + if (qs != NULL) + *qs = flow.qs; - if (recv_msg->result != 0) { - err = recv_msg->result; - goto fail_result; - } + return fd; +} - if (!recv_msg->has_pid || !recv_msg->has_flow_id || - recv_msg->qosspec == NULL) - goto fail_result; +int flow_alloc(const char * dst, + qosspec_t * qs, + const struct timespec * timeo) +{ + struct flow_info flow; + uint8_t buf[SOCK_BUF_SIZE]; + buffer_t msg = {buf, SOCK_BUF_SIZE}; + buffer_t sk; /* symmetric key */ + int fd; + int err; - if (recv_msg->pk.len != 0 && - crypt_dh_derive(pkp, recv_msg->pk.data, - recv_msg->pk.len, s) < 0) { - err = -ECRYPT; - goto fail_result; - } +#ifdef QOS_DISABLE_CRC + if (qs != NULL) + qs->ber = 1; +#endif - crypt_dh_pkp_destroy(pkp); + memset(&flow, 0, sizeof(flow)); - fd = flow_init(recv_msg->flow_id, recv_msg->pid, - msg_to_spec(recv_msg->qosspec), s); + flow.n_pid = getpid(); + flow.qs = qs == NULL ? qos_raw : *qs; - irm_msg__free_unpacked(recv_msg, NULL); + if (flow_alloc__irm_req_ser(&msg, &flow, dst, timeo)) + return -ENOMEM; - if (fd < 0) - return fd; + err = send_recv_msg(&msg); + if (err < 0) + return err; - pthread_rwlock_wrlock(&ai.lock); + err = flow__irm_result_des(&msg, &flow, &sk); + if (err < 0) + return err; - assert(ai.flows[fd].frcti == NULL); + fd = flow_init(&flow, &sk); - if (ai.flows[fd].qs.in_order != 0) { - ai.flows[fd].frcti = frcti_create(fd); - if (ai.flows[fd].frcti == NULL) { - pthread_rwlock_unlock(&ai.lock); - flow_dealloc(fd); - return -ENOMEM; - } - } + freebuf(sk); if (qs != NULL) - *qs = ai.flows[fd].qs; - - pthread_rwlock_unlock(&ai.lock); + *qs = flow.qs; return fd; - - fail_result: - irm_msg__free_unpacked(recv_msg, NULL); - fail_recv: - crypt_dh_pkp_destroy(pkp); - fail_crypt_pkp: - return err; } -static int __flow_alloc(const char * dst, - qosspec_t * qs, - const struct timespec * timeo, - bool join) +int flow_join(const char * dst, + qosspec_t * qs, + const struct timespec * timeo) { - irm_msg_t msg = IRM_MSG__INIT; - qosspec_msg_t qs_msg = QOSSPEC_MSG__INIT; - irm_msg_t * recv_msg; - int fd; - void * pkp = NULL; /* public key pair */ - uint8_t s[SYMMKEYSZ]; /* secret key for flow */ - uint8_t buf[MSGBUFSZ]; - int err = -EIRMD; - - memset(s, 0, SYMMKEYSZ); + struct flow_info flow; + uint8_t buf[SOCK_BUF_SIZE]; + buffer_t msg = {buf, SOCK_BUF_SIZE}; + int fd; + int err; #ifdef QOS_DISABLE_CRC if (qs != NULL) qs->ber = 1; #endif - msg.code = join ? IRM_MSG_CODE__IRM_FLOW_JOIN - : IRM_MSG_CODE__IRM_FLOW_ALLOC; - msg.dst = (char *) dst; - msg.has_pid = true; - msg.pid = ai.pid; - qs_msg = spec_to_msg(qs); - msg.qosspec = &qs_msg; + if (qs != NULL && qs->cypher_s > 0) + return -ENOTSUP; /* TODO: Encrypted broadcast */ - if (timeo != NULL) { - msg.has_timeo_sec = true; - msg.has_timeo_nsec = true; - msg.timeo_sec = timeo->tv_sec; - msg.timeo_nsec = timeo->tv_nsec; - } + memset(&flow, 0, sizeof(flow)); - if (!join && qs != NULL && qs->cypher_s != 0) { - ssize_t key_len; + flow.n_pid = getpid(); + flow.qs = qs == NULL ? qos_raw : *qs; - key_len = crypt_dh_pkp_create(&pkp, buf); - if (key_len < 0) { - err = -ECRYPT; - goto fail_crypt_pkp; - } + if (flow_alloc__irm_req_ser(&msg, &flow, dst, timeo)) + return -ENOMEM; - msg.has_pk = true; - msg.pk.data = buf; - msg.pk.len = (uint32_t) key_len; - } + err = send_recv_msg(&msg); + if (err < 0) + return err; - recv_msg = send_recv_irm_msg(&msg); - if (recv_msg == NULL) - goto fail_send; + err = flow__irm_result_des(&msg, &flow, NULL); + if (err < 0) + return err; - if (!recv_msg->has_result) - goto fail_result; + fd = flow_init(&flow, NULL); - if (recv_msg->result != 0) { - err = recv_msg->result; - goto fail_result; - } + if (qs != NULL) + *qs = flow.qs; - if (!recv_msg->has_pid || !recv_msg->has_flow_id) - goto fail_result; + return fd; +} - if (!join && qs != NULL && qs->cypher_s != 0) { - if (!recv_msg->has_pk || recv_msg->pk.len == 0) { - err = -ECRYPT; - goto fail_result; - } +#define PKT_BUF_LEN 2048 +int flow_dealloc(int fd) +{ + struct flow_info info; + uint8_t pkt[PKT_BUF_LEN]; + uint8_t buf[SOCK_BUF_SIZE]; + buffer_t msg = {buf, SOCK_BUF_SIZE}; + struct timespec tic = TIMESPEC_INIT_NS(TICTIME); + struct timespec timeo = TIMESPEC_INIT_S(0); + struct flow * flow; + int err; - if (crypt_dh_derive(pkp, recv_msg->pk.data, - recv_msg->pk.len, s) < 0) { - err = -ECRYPT; - goto fail_result; - } + if (fd < 0 || fd >= SYS_MAX_FLOWS ) + return -EINVAL; + + memset(&info, 0, sizeof(flow)); - crypt_dh_pkp_destroy(pkp); + flow = &ai.flows[fd]; + + pthread_rwlock_rdlock(&ai.lock); + + if (flow->info.id < 0) { + pthread_rwlock_unlock(&ai.lock); + return -ENOTALLOC; } - fd = flow_init(recv_msg->flow_id, recv_msg->pid, - qs == NULL ? qos_raw : *qs, s); + flow->oflags = FLOWFDEFAULT | FLOWFRNOPART; - irm_msg__free_unpacked(recv_msg, NULL); + flow->rcv_timesout = true; + flow->rcv_timeo = tic; - if (fd < 0) - return fd; + pthread_rwlock_unlock(&ai.lock); - pthread_rwlock_wrlock(&ai.lock); + flow_read(fd, buf, SOCK_BUF_SIZE); - assert(ai.flows[fd].frcti == NULL); + pthread_rwlock_rdlock(&ai.lock); - if (ai.flows[fd].qs.in_order != 0) { - ai.flows[fd].frcti = frcti_create(fd); - if (ai.flows[fd].frcti == NULL) { - pthread_rwlock_unlock(&ai.lock); - flow_dealloc(fd); - return -ENOMEM; - } + timeo.tv_sec = frcti_dealloc(flow->frcti); + while (timeo.tv_sec < 0) { /* keep the flow active for rtx */ + ssize_t ret; + + pthread_rwlock_unlock(&ai.lock); + + ret = flow_read(fd, pkt, PKT_BUF_LEN); + + pthread_rwlock_rdlock(&ai.lock); + + timeo.tv_sec = frcti_dealloc(flow->frcti); + + if (ret == -EFLOWDOWN && timeo.tv_sec < 0) + timeo.tv_sec = -timeo.tv_sec; } - pthread_rwlock_unlock(&ai.lock); + pthread_cleanup_push(__cleanup_rwlock_unlock, &ai.lock); - return fd; + shm_rbuff_fini(flow->tx_rb); - fail_result: - irm_msg__free_unpacked(recv_msg, NULL); - fail_send: - crypt_dh_pkp_destroy(pkp); - fail_crypt_pkp: - return err; -} + pthread_cleanup_pop(true); -int flow_alloc(const char * dst, - qosspec_t * qs, - const struct timespec * timeo) -{ - return __flow_alloc(dst, qs, timeo, false); -} + info.id = flow->info.id; + info.n_pid = getpid(); -int flow_join(const char * dst, - qosspec_t * qs, - const struct timespec * timeo) -{ - if (qs != NULL && qs->cypher_s != 0) - return -ECRYPT; + if (flow_dealloc__irm_req_ser(&msg, &info, &timeo) < 0) + return -ENOMEM; + + err = send_recv_msg(&msg); + if (err < 0) + return err; + + err = irm__irm_result_des(&msg); + + flow_fini(fd); - return __flow_alloc(dst, qs, timeo, true); + return err; } -int flow_dealloc(int fd) +int ipcp_flow_dealloc(int fd) { - irm_msg_t msg = IRM_MSG__INIT; - irm_msg_t * recv_msg; + struct flow_info info; + uint8_t buf[SOCK_BUF_SIZE]; + buffer_t msg = {buf, SOCK_BUF_SIZE}; + struct flow * flow; + int err; if (fd < 0 || fd >= SYS_MAX_FLOWS ) return -EINVAL; - msg.code = IRM_MSG_CODE__IRM_FLOW_DEALLOC; - msg.has_flow_id = true; - msg.has_pid = true; - msg.pid = ai.pid; + flow = &ai.flows[fd]; + + memset(&info, 0, sizeof(flow)); pthread_rwlock_rdlock(&ai.lock); - if (ai.flows[fd].flow_id < 0) { + if (flow->info.id < 0) { pthread_rwlock_unlock(&ai.lock); return -ENOTALLOC; } - msg.flow_id = ai.flows[fd].flow_id; + info.id = flow->info.id; + info.n_1_pid = flow->info.n_1_pid; pthread_rwlock_unlock(&ai.lock); - recv_msg = send_recv_irm_msg(&msg); - if (recv_msg == NULL) - return -EIRMD; - - if (!recv_msg->has_result) { - irm_msg__free_unpacked(recv_msg, NULL); - return -EIRMD; - } + if (ipcp_flow_dealloc__irm_req_ser(&msg, &info) < 0) + return -ENOMEM; - irm_msg__free_unpacked(recv_msg, NULL); + err = send_recv_msg(&msg); + if (err < 0) + return err; - pthread_rwlock_wrlock(&ai.lock); + err = irm__irm_result_des(&msg); flow_fini(fd); - pthread_rwlock_unlock(&ai.lock); - - return 0; + return err; } int fccntl(int fd, @@ -802,6 +1005,7 @@ int fccntl(int fd, { uint32_t * fflags; uint16_t * cflags; + uint16_t csflags; va_list l; struct timespec * timeo; qosspec_t * qs; @@ -819,7 +1023,7 @@ int fccntl(int fd, pthread_rwlock_wrlock(&ai.lock); - if (flow->flow_id < 0) { + if (flow->info.id < 0) { pthread_rwlock_unlock(&ai.lock); va_end(l); return -ENOTALLOC; @@ -858,13 +1062,13 @@ int fccntl(int fd, goto einval; if (!flow->rcv_timesout) goto eperm; - *timeo = flow->snd_timeo; + *timeo = flow->rcv_timeo; break; case FLOWGQOSSPEC: qs = va_arg(l, qosspec_t *); if (qs == NULL) goto einval; - *qs = flow->qs; + *qs = flow->info.qs; break; case FLOWGRXQLEN: qlen = va_arg(l, size_t *); @@ -891,13 +1095,13 @@ int fccntl(int fd, rx_acl |= ACL_FLOWDOWN; tx_acl |= ACL_FLOWDOWN; shm_flow_set_notify(flow->set, - flow->flow_id, + flow->info.id, FLOW_DOWN); } else { rx_acl &= ~ACL_FLOWDOWN; tx_acl &= ~ACL_FLOWDOWN; shm_flow_set_notify(flow->set, - flow->flow_id, + flow->info.id, FLOW_UP); } @@ -911,13 +1115,19 @@ int fccntl(int fd, goto einval; *fflags = flow->oflags; break; + case FRCTSFLAGS: + csflags = (uint16_t) va_arg(l, uint32_t); + if (flow->frcti == NULL) + goto eperm; + frcti_setflags(flow->frcti, csflags); + break; case FRCTGFLAGS: - cflags = (uint16_t *) va_arg(l, int *); + cflags = (uint16_t *) va_arg(l, uint32_t *); if (cflags == NULL) goto einval; if (flow->frcti == NULL) goto eperm; - *cflags = frcti_getconf(flow->frcti); + *cflags = frcti_getflags(flow->frcti); break; default: pthread_rwlock_unlock(&ai.lock); @@ -965,6 +1175,60 @@ static int add_crc(struct shm_du_buff * sdb) return 0; } +static int flow_tx_sdb(struct flow * flow, + struct shm_du_buff * sdb, + bool block, + struct timespec * abstime) +{ + struct timespec now; + ssize_t idx; + int ret; + + clock_gettime(PTHREAD_COND_CLOCK, &now); + + pthread_rwlock_wrlock(&ai.lock); + + flow->snd_act = now; + + pthread_rwlock_unlock(&ai.lock); + + idx = shm_du_buff_get_idx(sdb); + + pthread_rwlock_rdlock(&ai.lock); + + if (shm_du_buff_len(sdb) > 0) { + if (frcti_snd(flow->frcti, sdb) < 0) + goto enomem; + + if (crypt_encrypt(&flow->crypt, sdb) < 0) + goto enomem; + + if (flow->info.qs.ber == 0 && add_crc(sdb) != 0) + goto enomem; + } + + pthread_cleanup_push(__cleanup_rwlock_unlock, &ai.lock); + + if (!block) + ret = shm_rbuff_write(flow->tx_rb, idx); + else + ret = shm_rbuff_write_b(flow->tx_rb, idx, abstime); + + if (ret < 0) + shm_rdrbuff_remove(ai.rdrb, idx); + else + shm_flow_set_notify(flow->set, flow->info.id, FLOW_PKT); + + pthread_cleanup_pop(true); + + return 0; + +enomem: + pthread_rwlock_unlock(&ai.lock); + shm_rdrbuff_remove(ai.rdrb, idx); + return -ENOMEM; +} + ssize_t flow_write(int fd, const void * buf, size_t count) @@ -978,24 +1242,24 @@ ssize_t flow_write(int fd, struct shm_du_buff * sdb; uint8_t * ptr; - if (buf == NULL) - return 0; + if (buf == NULL && count != 0) + return -EINVAL; - if (fd < 0 || fd > PROG_MAX_FLOWS) + if (fd < 0 || fd >= PROG_MAX_FLOWS) return -EBADF; flow = &ai.flows[fd]; clock_gettime(PTHREAD_COND_CLOCK, &abs); - pthread_rwlock_rdlock(&ai.lock); + pthread_rwlock_wrlock(&ai.lock); - if (flow->flow_id < 0) { + if (flow->info.id < 0) { pthread_rwlock_unlock(&ai.lock); return -ENOTALLOC; } - if (ai.flows[fd].snd_timesout) { + if (flow->snd_timesout) { ts_add(&abs, &flow->snd_timeo, &abs); abstime = &abs; } @@ -1007,57 +1271,71 @@ ssize_t flow_write(int fd, if ((flags & FLOWFACCMODE) == FLOWFRDONLY) return -EPERM; - /* TODO: partial writes. */ - if (flags & FLOWFWNOBLOCK) - idx = shm_rdrbuff_alloc(ai.rdrb, - count, - &ptr, - &sdb); - else /* Blocking. */ - idx = shm_rdrbuff_alloc_b(ai.rdrb, - count, - &ptr, - &sdb, - abstime); + if (flags & FLOWFWNOBLOCK) { + if (!frcti_is_window_open(flow->frcti)) + return -EAGAIN; + idx = shm_rdrbuff_alloc(ai.rdrb, count, &ptr, &sdb); + } else { + ret = frcti_window_wait(flow->frcti, abstime); + if (ret < 0) + return ret; + idx = shm_rdrbuff_alloc_b(ai.rdrb, count, &ptr, &sdb, abstime); + } + if (idx < 0) return idx; - memcpy(ptr, buf, count); + if (count > 0) + memcpy(ptr, buf, count); - if (frcti_snd(flow->frcti, sdb) < 0) { - shm_rdrbuff_remove(ai.rdrb, idx); - return -ENOMEM; - } + ret = flow_tx_sdb(flow, sdb, !(flags & FLOWFWNOBLOCK), abstime); - pthread_rwlock_wrlock(&ai.lock); - if (flow->qs.cypher_s > 0) - if (crypt_encrypt(flow, sdb) < 0) { - pthread_rwlock_unlock(&ai.lock); - shm_rdrbuff_remove(ai.rdrb, idx); - return -ENOMEM; - } - pthread_rwlock_unlock(&ai.lock); + return ret < 0 ? (ssize_t) ret : (ssize_t) count; +} - if (flow->qs.ber == 0 && add_crc(sdb) != 0) { - shm_rdrbuff_remove(ai.rdrb, idx); - return -ENOMEM; - } +static bool invalid_pkt(struct flow * flow, + struct shm_du_buff * sdb) +{ + if (shm_du_buff_len(sdb) == 0) + return true; - pthread_rwlock_rdlock(&ai.lock); + if (flow->info.qs.ber == 0 && chk_crc(sdb) != 0) + return true; - if (flags & FLOWFWNOBLOCK) - ret = shm_rbuff_write(flow->tx_rb, idx); - else - ret = shm_rbuff_write_b(flow->tx_rb, idx, abstime); + if (crypt_decrypt(&flow->crypt, sdb) < 0) + return true; - if (ret < 0) - shm_rdrbuff_remove(ai.rdrb, idx); - else - shm_flow_set_notify(flow->set, flow->flow_id, FLOW_PKT); + return false; +} + +static ssize_t flow_rx_sdb(struct flow * flow, + struct shm_du_buff ** sdb, + bool block, + struct timespec * abstime) +{ + ssize_t idx; + struct timespec now; + + idx = block ? shm_rbuff_read_b(flow->rx_rb, abstime) : + shm_rbuff_read(flow->rx_rb); + if (idx < 0) + return idx; + + clock_gettime(PTHREAD_COND_CLOCK, &now); + + pthread_rwlock_wrlock(&ai.lock); + + flow->rcv_act = now; pthread_rwlock_unlock(&ai.lock); - return ret < 0 ? (ssize_t) ret : (ssize_t) count; + *sdb = shm_rdrbuff_get(ai.rdrb, idx); + if (invalid_pkt(flow, *sdb)) { + shm_rdrbuff_remove(ai.rdrb, idx); + return -EAGAIN; + } + + return idx; } ssize_t flow_read(int fd, @@ -1067,92 +1345,100 @@ ssize_t flow_read(int fd, ssize_t idx; ssize_t n; uint8_t * packet; - struct shm_rbuff * rb; struct shm_du_buff * sdb; struct timespec abs; + struct timespec now; struct timespec * abstime = NULL; struct flow * flow; - bool noblock; + bool block; bool partrd; - if (fd < 0 || fd > PROG_MAX_FLOWS) + if (fd < 0 || fd >= PROG_MAX_FLOWS) return -EBADF; flow = &ai.flows[fd]; - if (flow->part_idx == DONE_PART) { - flow->part_idx = NO_PART; - return 0; - } - - clock_gettime(PTHREAD_COND_CLOCK, &abs); + clock_gettime(PTHREAD_COND_CLOCK, &now); pthread_rwlock_rdlock(&ai.lock); - if (flow->flow_id < 0) { + if (flow->info.id < 0) { pthread_rwlock_unlock(&ai.lock); return -ENOTALLOC; } - rb = flow->rx_rb; - noblock = flow->oflags & FLOWFRNOBLOCK; + if (flow->part_idx == DONE_PART) { + pthread_rwlock_unlock(&ai.lock); + flow->part_idx = NO_PART; + return 0; + } + + block = !(flow->oflags & FLOWFRNOBLOCK); partrd = !(flow->oflags & FLOWFRNOPART); - if (ai.flows[fd].rcv_timesout) { - ts_add(&abs, &flow->rcv_timeo, &abs); + if (flow->rcv_timesout) { + ts_add(&now, &flow->rcv_timeo, &abs); abstime = &abs; } - pthread_rwlock_unlock(&ai.lock); - idx = flow->part_idx; if (idx < 0) { - idx = frcti_queued_pdu(flow->frcti); - if (idx < 0) { - do { - idx = noblock ? shm_rbuff_read(rb) : - shm_rbuff_read_b(rb, abstime); - if (idx < 0) + while ((idx = frcti_queued_pdu(flow->frcti)) < 0) { + pthread_rwlock_unlock(&ai.lock); + + idx = flow_rx_sdb(flow, &sdb, block, abstime); + if (idx < 0) { + if (block && idx != -EAGAIN) return idx; + if (!block) + return idx; + + pthread_rwlock_rdlock(&ai.lock); + continue; + } - sdb = shm_rdrbuff_get(ai.rdrb, idx); - if (flow->qs.ber == 0 && chk_crc(sdb) != 0) { - shm_rdrbuff_remove(ai.rdrb, idx); - continue; - } - - pthread_rwlock_wrlock(&ai.lock); - if (flow->qs.cypher_s > 0) - if (crypt_decrypt(flow, sdb) < 0) { - pthread_rwlock_unlock(&ai.lock); - shm_rdrbuff_remove(ai.rdrb, - idx); - return -ENOMEM; - } - pthread_rwlock_unlock(&ai.lock); - } while (frcti_rcv(flow->frcti, sdb) != 0); + pthread_rwlock_rdlock(&ai.lock); + + frcti_rcv(flow->frcti, sdb); } } - n = shm_rdrbuff_read(&packet, ai.rdrb, idx); + sdb = shm_rdrbuff_get(ai.rdrb, idx); + + pthread_rwlock_unlock(&ai.lock); + + packet = shm_du_buff_head(sdb); + + n = shm_du_buff_len(sdb); assert(n >= 0); if (n <= (ssize_t) count) { memcpy(buf, packet, n); - shm_rdrbuff_remove(ai.rdrb, idx); + ipcp_sdb_release(sdb); + + pthread_rwlock_wrlock(&ai.lock); + flow->part_idx = (partrd && n == (ssize_t) count) ? DONE_PART : NO_PART; + + flow->rcv_act = now; + + pthread_rwlock_unlock(&ai.lock); return n; } else { if (partrd) { memcpy(buf, packet, count); - sdb = shm_rdrbuff_get(ai.rdrb, idx); shm_du_buff_head_release(sdb, n); + pthread_rwlock_wrlock(&ai.lock); flow->part_idx = idx; + + flow->rcv_act = now; + + pthread_rwlock_unlock(&ai.lock); return count; } else { - shm_rdrbuff_remove(ai.rdrb, idx); + ipcp_sdb_release(sdb); return -EMSGSIZE; } } @@ -1160,26 +1446,31 @@ ssize_t flow_read(int fd, /* fqueue functions. */ -struct flow_set * fset_create() +struct flow_set * fset_create(void) { - struct flow_set * set = malloc(sizeof(*set)); + struct flow_set * set; + + set = malloc(sizeof(*set)); if (set == NULL) - return NULL; + goto fail_malloc; assert(ai.fqueues); pthread_rwlock_wrlock(&ai.lock); set->idx = bmp_allocate(ai.fqueues); - if (!bmp_is_id_valid(ai.fqueues, set->idx)) { - pthread_rwlock_unlock(&ai.lock); - free(set); - return NULL; - } + if (!bmp_is_id_valid(ai.fqueues, set->idx)) + goto fail_bmp_alloc; pthread_rwlock_unlock(&ai.lock); return set; + + fail_bmp_alloc: + pthread_rwlock_unlock(&ai.lock); + free(set); + fail_malloc: + return NULL; } void fset_destroy(struct flow_set * set) @@ -1198,13 +1489,13 @@ void fset_destroy(struct flow_set * set) free(set); } -struct fqueue * fqueue_create() +struct fqueue * fqueue_create(void) { struct fqueue * fq = malloc(sizeof(*fq)); if (fq == NULL) return NULL; - memset(fq->fqueue, -1, (SHM_BUFFER_SIZE) * sizeof(*fq->fqueue)); + memset(fq->fqueue, -1, SHM_BUFFER_SIZE * sizeof(*fq->fqueue)); fq->fqsize = 0; fq->next = 0; @@ -1227,41 +1518,57 @@ void fset_zero(struct flow_set * set) int fset_add(struct flow_set * set, int fd) { - int ret; - size_t packets; - size_t i; + struct flow * flow; + int ret; - if (set == NULL || fd < 0 || fd > SYS_MAX_FLOWS) + if (set == NULL || fd < 0 || fd >= SYS_MAX_FLOWS) return -EINVAL; - pthread_rwlock_wrlock(&ai.lock); + flow = &ai.flows[fd]; - if (ai.flows[fd].flow_id < 0) { - pthread_rwlock_unlock(&ai.lock); - return -EINVAL; + pthread_rwlock_rdlock(&ai.lock); + + if (flow->info.id < 0) { + ret = -EINVAL; + goto fail; } - ret = shm_flow_set_add(ai.fqset, set->idx, ai.flows[fd].flow_id); + if (flow->frcti != NULL) + shm_flow_set_del(ai.fqset, 0, ai.flows[fd].info.id); - packets = shm_rbuff_queued(ai.flows[fd].rx_rb); - for (i = 0; i < packets; i++) - shm_flow_set_notify(ai.fqset, ai.flows[fd].flow_id, FLOW_PKT); + ret = shm_flow_set_add(ai.fqset, set->idx, ai.flows[fd].info.id); + if (ret < 0) + goto fail; + + if (shm_rbuff_queued(ai.flows[fd].rx_rb)) + shm_flow_set_notify(ai.fqset, ai.flows[fd].info.id, FLOW_PKT); pthread_rwlock_unlock(&ai.lock); return ret; + + fail: + pthread_rwlock_unlock(&ai.lock); + return ret; } void fset_del(struct flow_set * set, int fd) { - if (set == NULL || fd < 0 || fd > SYS_MAX_FLOWS) + struct flow * flow; + + if (set == NULL || fd < 0 || fd >= SYS_MAX_FLOWS) return; - pthread_rwlock_wrlock(&ai.lock); + flow = &ai.flows[fd]; - if (ai.flows[fd].flow_id >= 0) - shm_flow_set_del(ai.fqset, set->idx, ai.flows[fd].flow_id); + pthread_rwlock_rdlock(&ai.lock); + + if (flow->info.id >= 0) + shm_flow_set_del(ai.fqset, set->idx, flow->info.id); + + if (flow->frcti != NULL) + shm_flow_set_add(ai.fqset, 0, ai.flows[fd].info.id); pthread_rwlock_unlock(&ai.lock); } @@ -1269,28 +1576,86 @@ void fset_del(struct flow_set * set, bool fset_has(const struct flow_set * set, int fd) { - bool ret = false; + bool ret; - if (set == NULL || fd < 0 || fd > SYS_MAX_FLOWS) + if (set == NULL || fd < 0 || fd >= SYS_MAX_FLOWS) return false; pthread_rwlock_rdlock(&ai.lock); - if (ai.flows[fd].flow_id < 0) { + if (ai.flows[fd].info.id < 0) { pthread_rwlock_unlock(&ai.lock); return false; } - ret = (shm_flow_set_has(ai.fqset, set->idx, ai.flows[fd].flow_id) == 1); + ret = (shm_flow_set_has(ai.fqset, set->idx, ai.flows[fd].info.id) == 1); pthread_rwlock_unlock(&ai.lock); return ret; } +/* Filter fqueue events for non-data packets */ +static int fqueue_filter(struct fqueue * fq) +{ + struct shm_du_buff * sdb; + int fd; + ssize_t idx; + struct frcti * frcti; + + while (fq->next < fq->fqsize) { + if (fq->fqueue[fq->next].event != FLOW_PKT) + return 1; + + pthread_rwlock_rdlock(&ai.lock); + + fd = ai.id_to_fd[fq->fqueue[fq->next].flow_id].fd; + if (fd < 0) { + ++fq->next; + pthread_rwlock_unlock(&ai.lock); + continue; + } + + frcti = ai.flows[fd].frcti; + if (frcti == NULL) { + pthread_rwlock_unlock(&ai.lock); + return 1; + } + + if (__frcti_pdu_ready(frcti) >= 0) { + pthread_rwlock_unlock(&ai.lock); + return 1; + } + + pthread_rwlock_unlock(&ai.lock); + + idx = flow_rx_sdb(&ai.flows[fd], &sdb, false, NULL); + if (idx < 0) + return 0; + + pthread_rwlock_rdlock(&ai.lock); + + sdb = shm_rdrbuff_get(ai.rdrb, idx); + + __frcti_rcv(frcti, sdb); + + if (__frcti_pdu_ready(frcti) >= 0) { + pthread_rwlock_unlock(&ai.lock); + return 1; + } + + pthread_rwlock_unlock(&ai.lock); + + ++fq->next; + } + + return 0; +} + int fqueue_next(struct fqueue * fq) { - int fd; + int fd; + struct flowevent * e; if (fq == NULL) return -EINVAL; @@ -1298,11 +1663,16 @@ int fqueue_next(struct fqueue * fq) if (fq->fqsize == 0 || fq->next == fq->fqsize) return -EPERM; + if (fq->next != 0 && fqueue_filter(fq) == 0) + return -EPERM; + pthread_rwlock_rdlock(&ai.lock); - fd = ai.ports[fq->fqueue[fq->next]].fd; + e = fq->fqueue + fq->next; - fq->next += 2; + fd = ai.id_to_fd[e->flow_id].fd; + + ++fq->next; pthread_rwlock_unlock(&ai.lock); @@ -1317,60 +1687,80 @@ enum fqtype fqueue_type(struct fqueue * fq) if (fq->fqsize == 0 || fq->next == 0) return -EPERM; - return fq->fqueue[fq->next - 1]; + return fq->fqueue[(fq->next - 1)].event; } ssize_t fevent(struct flow_set * set, struct fqueue * fq, const struct timespec * timeo) { - ssize_t ret; - struct timespec abstime; + ssize_t ret = 0; + struct timespec abs; struct timespec * t = NULL; if (set == NULL || fq == NULL) return -EINVAL; if (fq->fqsize > 0 && fq->next != fq->fqsize) - return fq->fqsize; + return 1; + + clock_gettime(PTHREAD_COND_CLOCK, &abs); if (timeo != NULL) { - clock_gettime(PTHREAD_COND_CLOCK, &abstime); - ts_add(&abstime, timeo, &abstime); - t = &abstime; + ts_add(&abs, timeo, &abs); + t = &abs; } - ret = shm_flow_set_wait(ai.fqset, set->idx, fq->fqueue, t); - if (ret == -ETIMEDOUT) { - fq->fqsize = 0; - return -ETIMEDOUT; - } + while (ret == 0) { + ret = shm_flow_set_wait(ai.fqset, set->idx, fq->fqueue, t); + if (ret == -ETIMEDOUT) + return -ETIMEDOUT; - fq->fqsize = ret << 1; - fq->next = 0; + fq->fqsize = ret; + fq->next = 0; + + ret = fqueue_filter(fq); + } - assert(ret); + assert(ret != 0); - return ret; + return 1; } /* ipcp-dev functions. */ -int np1_flow_alloc(pid_t n_pid, - int flow_id, - qosspec_t qs) +int np1_flow_alloc(pid_t n_pid, + int flow_id) { - qs.cypher_s = 0; /* No encryption ctx for np1 */ - return flow_init(flow_id, n_pid, qs, NULL); + struct flow_info flow; + + memset(&flow, 0, sizeof(flow)); + + flow.id = flow_id; + flow.n_pid = getpid(); + flow.qs = qos_np1; + flow.mpl = 0; + flow.n_1_pid = n_pid; /* This "flow" is upside-down! */ + + return flow_init(&flow, NULL); } -int np1_flow_dealloc(int flow_id) +int np1_flow_dealloc(int flow_id, + time_t timeo) { int fd; + /* + * TODO: Don't pass timeo to the IPCP but wait in IRMd. + * This will need async ops, waiting until we bootstrap + * the IRMd over ouroboros. + */ + + sleep(timeo); + pthread_rwlock_rdlock(&ai.lock); - fd = ai.ports[flow_id].fd; + fd = ai.id_to_fd[flow_id].fd; pthread_rwlock_unlock(&ai.lock); @@ -1381,140 +1771,106 @@ int np1_flow_resp(int flow_id) { int fd; - if (port_wait_assign(flow_id) != PORT_ID_ASSIGNED) + if (flow_wait_assign(flow_id) != FLOW_ALLOCATED) return -1; pthread_rwlock_rdlock(&ai.lock); - fd = ai.ports[flow_id].fd; + fd = ai.id_to_fd[flow_id].fd; pthread_rwlock_unlock(&ai.lock); return fd; } -int ipcp_create_r(int result) +int ipcp_create_r(const struct ipcp_info * info) { - irm_msg_t msg = IRM_MSG__INIT; - irm_msg_t * recv_msg; - int ret; - - msg.code = IRM_MSG_CODE__IPCP_CREATE_R; - msg.has_pid = true; - msg.pid = getpid(); - msg.has_result = true; - msg.result = result; - - recv_msg = send_recv_irm_msg(&msg); - if (recv_msg == NULL) - return -EIRMD; - - if (!recv_msg->has_result) { - irm_msg__free_unpacked(recv_msg, NULL); - return -1; - } + uint8_t buf[SOCK_BUF_SIZE]; + buffer_t msg = {buf, SOCK_BUF_SIZE}; + int err; - ret = recv_msg->result; - irm_msg__free_unpacked(recv_msg, NULL); + if (ipcp_create_r__irm_req_ser(&msg,info) < 0) + return -ENOMEM; - return ret; + err = send_recv_msg(&msg); + if (err < 0) + return err; + + return irm__irm_result_des(&msg); } -int ipcp_flow_req_arr(const uint8_t * dst, - size_t len, - qosspec_t qs, - const void * data, - size_t dlen) +int ipcp_flow_req_arr(const buffer_t * dst, + qosspec_t qs, + time_t mpl, + const buffer_t * data) { - irm_msg_t msg = IRM_MSG__INIT; - irm_msg_t * recv_msg; - qosspec_msg_t qs_msg; - int fd; - - assert(dst != NULL); - - msg.code = IRM_MSG_CODE__IPCP_FLOW_REQ_ARR; - msg.has_pid = true; - msg.pid = getpid(); - msg.has_hash = true; - msg.hash.len = len; - msg.hash.data = (uint8_t *) dst; - qs_msg = spec_to_msg(&qs); - msg.qosspec = &qs_msg; - msg.has_pk = true; - msg.pk.data = (uint8_t *) data; - msg.pk.len = dlen; - - recv_msg = send_recv_irm_msg(&msg); - if (recv_msg == NULL) - return -EIRMD; - - if (!recv_msg->has_flow_id || !recv_msg->has_pid) { - irm_msg__free_unpacked(recv_msg, NULL); - return -1; - } + struct flow_info flow; + uint8_t buf[SOCK_BUF_SIZE]; + buffer_t msg = {buf, SOCK_BUF_SIZE}; + int err; - if (recv_msg->has_result && recv_msg->result) { - irm_msg__free_unpacked(recv_msg, NULL); - return -1; - } + memset(&flow, 0, sizeof(flow)); - qs.cypher_s = 0; /* No encryption ctx for np1 */ - fd = flow_init(recv_msg->flow_id, recv_msg->pid, qs, NULL); + assert(dst != NULL && dst->len != 0 && dst->data != NULL); - irm_msg__free_unpacked(recv_msg, NULL); + flow.n_1_pid = getpid(); + flow.qs = qs; + flow.mpl = mpl; - return fd; + if (ipcp_flow_req_arr__irm_req_ser(&msg, dst, &flow, data) < 0) + return -ENOMEM; + + err = send_recv_msg(&msg); + if (err < 0) + return err; + + err = flow__irm_result_des(&msg, &flow, NULL); + if (err < 0) + return err; + + /* inverted for np1_flow */ + flow.n_1_pid = flow.n_pid; + flow.n_pid = getpid(); + flow.mpl = 0; + + return flow_init(&flow, NULL); } -int ipcp_flow_alloc_reply(int fd, - int response, - const void * data, - size_t len) +int ipcp_flow_alloc_reply(int fd, + int response, + time_t mpl, + const buffer_t * data) { - irm_msg_t msg = IRM_MSG__INIT; - irm_msg_t * recv_msg; - int ret; + struct flow_info flow; + uint8_t buf[SOCK_BUF_SIZE]; + buffer_t msg = {buf, SOCK_BUF_SIZE}; + int err; assert(fd >= 0 && fd < SYS_MAX_FLOWS); - msg.code = IRM_MSG_CODE__IPCP_FLOW_ALLOC_REPLY; - msg.has_flow_id = true; - msg.has_pk = true; - msg.pk.data = (uint8_t *) data; - msg.pk.len = (uint32_t) len; - pthread_rwlock_rdlock(&ai.lock); - msg.flow_id = ai.flows[fd].flow_id; + flow.id = ai.flows[fd].info.id; pthread_rwlock_unlock(&ai.lock); - msg.has_response = true; - msg.response = response; - - recv_msg = send_recv_irm_msg(&msg); - if (recv_msg == NULL) - return -EIRMD; + flow.mpl = mpl; - if (!recv_msg->has_result) { - irm_msg__free_unpacked(recv_msg, NULL); - return -1; - } - - ret = recv_msg->result; + if (ipcp_flow_alloc_reply__irm_msg_ser(&msg, &flow, response, data) < 0) + return -ENOMEM; - irm_msg__free_unpacked(recv_msg, NULL); + err = send_recv_msg(&msg); + if (err < 0) + return err; - return ret; + return irm__irm_result_des(&msg); } int ipcp_flow_read(int fd, struct shm_du_buff ** sdb) { - struct flow * flow; - struct shm_rbuff * rb; - ssize_t idx; + struct flow * flow; + ssize_t idx = -1; assert(fd >= 0 && fd < SYS_MAX_FLOWS); assert(sdb); @@ -1523,28 +1879,21 @@ int ipcp_flow_read(int fd, pthread_rwlock_rdlock(&ai.lock); - assert(flow->flow_id >= 0); + assert(flow->info.id >= 0); - rb = flow->rx_rb; + while (frcti_queued_pdu(flow->frcti) < 0) { + pthread_rwlock_unlock(&ai.lock); - pthread_rwlock_unlock(&ai.lock); + idx = flow_rx_sdb(flow, sdb, false, NULL); + if (idx < 0) + return idx; - if (flow->frcti != NULL) { - idx = frcti_queued_pdu(flow->frcti); - if (idx >= 0) { - *sdb = shm_rdrbuff_get(ai.rdrb, idx); - return 0; - } + pthread_rwlock_rdlock(&ai.lock); + + frcti_rcv(flow->frcti, *sdb); } - do { - idx = shm_rbuff_read(rb); - if (idx < 0) - return idx; - *sdb = shm_rdrbuff_get(ai.rdrb, idx); - if (flow->qs.ber == 0 && chk_crc(*sdb) != 0) - continue; - } while (frcti_rcv(flow->frcti, *sdb) != 0); + pthread_rwlock_unlock(&ai.lock); return 0; } @@ -1552,18 +1901,17 @@ int ipcp_flow_read(int fd, int ipcp_flow_write(int fd, struct shm_du_buff * sdb) { - struct flow * flow; - int ret; - ssize_t idx; + struct flow * flow; + int ret; assert(fd >= 0 && fd < SYS_MAX_FLOWS); assert(sdb); flow = &ai.flows[fd]; - pthread_rwlock_rdlock(&ai.lock); + pthread_rwlock_wrlock(&ai.lock); - if (flow->flow_id < 0) { + if (flow->info.id < 0) { pthread_rwlock_unlock(&ai.lock); return -ENOTALLOC; } @@ -1573,30 +1921,74 @@ int ipcp_flow_write(int fd, return -EPERM; } - assert(flow->tx_rb); + pthread_rwlock_unlock(&ai.lock); - idx = shm_du_buff_get_idx(sdb); + ret = flow_tx_sdb(flow, sdb, true, NULL); + + return ret; +} + +int np1_flow_read(int fd, + struct shm_du_buff ** sdb) +{ + struct flow * flow; + ssize_t idx = -1; + + assert(fd >= 0 && fd < SYS_MAX_FLOWS); + assert(sdb); - if (frcti_snd(flow->frcti, sdb) < 0) { + flow = &ai.flows[fd]; + + assert(flow->info.id >= 0); + + pthread_rwlock_rdlock(&ai.lock); + + idx = shm_rbuff_read(flow->rx_rb);; + if (idx < 0) { pthread_rwlock_unlock(&ai.lock); - return -ENOMEM; + return idx; } - if (flow->qs.ber == 0 && add_crc(sdb) != 0) { + pthread_rwlock_unlock(&ai.lock); + + *sdb = shm_rdrbuff_get(ai.rdrb, idx); + + return 0; +} + +int np1_flow_write(int fd, + struct shm_du_buff * sdb) +{ + struct flow * flow; + int ret; + ssize_t idx; + + assert(fd >= 0 && fd < SYS_MAX_FLOWS); + assert(sdb); + + flow = &ai.flows[fd]; + + pthread_rwlock_rdlock(&ai.lock); + + if (flow->info.id < 0) { pthread_rwlock_unlock(&ai.lock); - shm_rdrbuff_remove(ai.rdrb, idx); - return -ENOMEM; + return -ENOTALLOC; } - ret = shm_rbuff_write_b(flow->tx_rb, idx, NULL); - if (ret == 0) - shm_flow_set_notify(flow->set, flow->flow_id, FLOW_PKT); - else - shm_rdrbuff_remove(ai.rdrb, idx); + if ((flow->oflags & FLOWFACCMODE) == FLOWFRDONLY) { + pthread_rwlock_unlock(&ai.lock); + return -EPERM; + } pthread_rwlock_unlock(&ai.lock); - assert(ret <= 0); + idx = shm_du_buff_get_idx(sdb); + + ret = shm_rbuff_write_b(flow->tx_rb, idx, NULL); + if (ret < 0) + shm_rdrbuff_remove(ai.rdrb, idx); + else + shm_flow_set_notify(flow->set, flow->info.id, FLOW_PKT); return ret; } @@ -1620,7 +2012,7 @@ int ipcp_flow_fini(int fd) pthread_rwlock_rdlock(&ai.lock); - if (ai.flows[fd].flow_id < 0) { + if (ai.flows[fd].info.id < 0) { pthread_rwlock_unlock(&ai.lock); return -1; } @@ -1629,7 +2021,7 @@ int ipcp_flow_fini(int fd) shm_rbuff_set_acl(ai.flows[fd].tx_rb, ACL_FLOWDOWN); shm_flow_set_notify(ai.flows[fd].set, - ai.flows[fd].flow_id, + ai.flows[fd].info.id, FLOW_DEALLOC); rx_rb = ai.flows[fd].rx_rb; @@ -1650,15 +2042,30 @@ int ipcp_flow_get_qoscube(int fd, pthread_rwlock_rdlock(&ai.lock); - assert(ai.flows[fd].flow_id >= 0); + assert(ai.flows[fd].info.id >= 0); - *cube = qos_spec_to_cube(ai.flows[fd].qs); + *cube = qos_spec_to_cube(ai.flows[fd].info.qs); pthread_rwlock_unlock(&ai.lock); return 0; } +size_t ipcp_flow_queued(int fd) +{ + size_t q; + + pthread_rwlock_rdlock(&ai.lock); + + assert(ai.flows[fd].info.id >= 0); + + q = shm_rbuff_queued(ai.flows[fd].tx_rb); + + pthread_rwlock_unlock(&ai.lock); + + return q; +} + ssize_t local_flow_read(int fd) { ssize_t ret; @@ -1686,13 +2093,14 @@ int local_flow_write(int fd, pthread_rwlock_rdlock(&ai.lock); - if (flow->flow_id < 0) { + if (flow->info.id < 0) { pthread_rwlock_unlock(&ai.lock); return -ENOTALLOC; } + ret = shm_rbuff_write_b(flow->tx_rb, idx, NULL); if (ret == 0) - shm_flow_set_notify(flow->set, flow->flow_id, FLOW_PKT); + shm_flow_set_notify(flow->set, flow->info.id, FLOW_PKT); else shm_rdrbuff_remove(ai.rdrb, idx); diff --git a/src/lib/frct.c b/src/lib/frct.c index e4b858d0..c6fef35c 100644 --- a/src/lib/frct.c +++ b/src/lib/frct.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Flow and Retransmission Control * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License @@ -20,47 +20,62 @@ * Foundation, Inc., http://www.fsf.org/about/contact/. */ -/* Default Delta-t parameters */ -#define DELT_MPL 60000 /* ms */ -#define DELT_A 0 /* ms */ -#define DELT_R 2000 /* ms */ +#include <ouroboros/endian.h> -#define RQ_SIZE 64 +#define DELT_RDV (100 * MILLION) /* ns */ +#define MAX_RDV (1 * BILLION) /* ns */ -#define TW_ELEMENTS 6000 -#define TW_RESOLUTION 1 /* ms */ - -#define FRCT_PCILEN (sizeof(struct frct_pci)) +#define FRCT "frct" +#define FRCT_PCILEN (sizeof(struct frct_pci)) +#define FRCT_NAME_STRLEN 32 struct frct_cr { - uint32_t lwe; - uint32_t rwe; + uint32_t lwe; /* Left window edge */ + uint32_t rwe; /* Right window edge */ - uint8_t cflags; - uint32_t seqno; + uint8_t cflags; + uint32_t seqno; /* SEQ to send, or last SEQ Ack'd */ - time_t act; /* s */ - time_t inact; /* s */ + struct timespec act; /* Last seen activity */ + time_t inact; /* Inactivity (s) */ }; struct frcti { - int fd; - - time_t mpl; - time_t a; - time_t r; + int fd; + + time_t mpl; + time_t a; + time_t r; + time_t rdv; + + time_t srtt; /* Smoothed rtt */ + time_t mdev; /* Deviation */ + time_t rto; /* Retransmission timeout */ + uint32_t rttseq; + struct timespec t_probe; /* Probe time */ + bool probe; /* Probe active */ +#ifdef PROC_FLOW_STATS + size_t n_rtx; /* Number of rxm packets */ + size_t n_prb; /* Number of rtt probes */ + size_t n_rtt; /* Number of estimates */ + size_t n_dup; /* Duplicates received */ + size_t n_dak; /* Delayed ACKs received */ + size_t n_rdv; /* Number of rdv packets */ + size_t n_out; /* Packets out of window */ + size_t n_rqo; /* Packets out of rqueue */ +#endif + struct frct_cr snd_cr; + struct frct_cr rcv_cr; - time_t srtt_us; /* smoothed rtt */ - time_t mdev_us; /* mdev */ - uint32_t rttseq; - struct timespec t_probe; /* probe time */ - bool probe; /* probe active */ - struct frct_cr snd_cr; - struct frct_cr rcv_cr; + ssize_t rq[RQ_SIZE]; + pthread_rwlock_t lock; - ssize_t rq[RQ_SIZE]; - pthread_rwlock_t lock; + bool open; /* Window open/closed */ + struct timespec t_wnd; /* Window closed time */ + struct timespec t_rdvs; /* Last rendez-vous sent */ + pthread_cond_t cond; + pthread_mutex_t mtx; }; enum frct_flags { @@ -68,28 +83,265 @@ enum frct_flags { FRCT_DRF = 0x02, /* Data run flag */ FRCT_ACK = 0x04, /* ACK field valid */ FRCT_FC = 0x08, /* FC window valid */ - FRCT_RDVZ = 0x10, /* Rendez-vous */ + FRCT_RDVS = 0x10, /* Rendez-vous */ FRCT_FFGM = 0x20, /* First Fragment */ FRCT_MFGM = 0x40, /* More fragments */ }; struct frct_pci { - uint16_t flags; + uint8_t flags; + uint8_t pad; /* 24 bit window! */ uint16_t window; uint32_t seqno; uint32_t ackno; } __attribute__((packed)); -#include <rxmwheel.c> +#ifdef PROC_FLOW_STATS -static struct frcti * frcti_create(int fd) +static int frct_rib_read(const char * path, + char * buf, + size_t len) { - struct frcti * frcti; - time_t delta_t; - ssize_t idx; struct timespec now; + char * entry; + struct flow * flow; + struct frcti * frcti; + int fd; + + (void) len; + + entry = strstr(path, RIB_SEPARATOR); + assert(entry); + *entry = '\0'; + + fd = atoi(path); + + flow = &ai.flows[fd]; + + clock_gettime(PTHREAD_COND_CLOCK, &now); + + pthread_rwlock_rdlock(&ai.lock); + + frcti = flow->frcti; + + pthread_rwlock_rdlock(&frcti->lock); + + sprintf(buf, + "Maximum packet lifetime (ns): %20ld\n" + "Max time to Ack (ns): %20ld\n" + "Max time to Retransmit (ns): %20ld\n" + "Smoothed rtt (ns): %20ld\n" + "RTT standard deviation (ns): %20ld\n" + "Retransmit timeout RTO (ns): %20ld\n" + "Sender left window edge: %20u\n" + "Sender right window edge: %20u\n" + "Sender inactive (ns): %20ld\n" + "Sender current sequence number: %20u\n" + "Receiver left window edge: %20u\n" + "Receiver right window edge: %20u\n" + "Receiver inactive (ns): %20ld\n" + "Receiver last ack: %20u\n" + "Number of pkt retransmissions: %20zu\n" + "Number of rtt probes: %20zu\n" + "Number of rtt estimates: %20zu\n" + "Number of duplicates received: %20zu\n" + "Number of delayed acks received: %20zu\n" + "Number of rendez-vous sent: %20zu\n" + "Number of packets out of window: %20zu\n" + "Number of packets out of rqueue: %20zu\n", + frcti->mpl, + frcti->a, + frcti->r, + frcti->srtt, + frcti->mdev, + frcti->rto, + frcti->snd_cr.lwe, + frcti->snd_cr.rwe, + ts_diff_ns(&frcti->snd_cr.act, &now), + frcti->snd_cr.seqno, + frcti->rcv_cr.lwe, + frcti->rcv_cr.rwe, + ts_diff_ns(&frcti->rcv_cr.act, &now), + frcti->rcv_cr.seqno, + frcti->n_rtx, + frcti->n_prb, + frcti->n_rtt, + frcti->n_dup, + frcti->n_dak, + frcti->n_rdv, + frcti->n_out, + frcti->n_rqo); + + pthread_rwlock_unlock(&flow->frcti->lock); + + pthread_rwlock_unlock(&ai.lock); + + return strlen(buf); +} + +static int frct_rib_readdir(char *** buf) +{ + *buf = malloc(sizeof(**buf)); + if (*buf == NULL) + goto fail_malloc; + + (*buf)[0] = strdup("frct"); + if ((*buf)[0] == NULL) + goto fail_strdup; + + return 1; + + fail_strdup: + free(*buf); + fail_malloc: + return -ENOMEM; +} + +static int frct_rib_getattr(const char * path, + struct rib_attr * attr) +{ + (void) path; + (void) attr; + + attr->size = 1189; + attr->mtime = 0; + + return 0; +} + + +static struct rib_ops r_ops = { + .read = frct_rib_read, + .readdir = frct_rib_readdir, + .getattr = frct_rib_getattr +}; + +#endif /* PROC_FLOW_STATS */ + +static bool before(uint32_t seq1, + uint32_t seq2) +{ + return (int32_t)(seq1 - seq2) < 0; +} + +static bool after(uint32_t seq1, + uint32_t seq2) +{ + return (int32_t)(seq2 - seq1) < 0; +} + +static void __send_frct_pkt(int fd, + uint8_t flags, + uint32_t ackno, + uint32_t rwe) +{ + struct shm_du_buff * sdb; + struct frct_pci * pci; + ssize_t idx; + struct flow * f; + + /* Raw calls needed to bypass frcti. */ +#ifdef RXM_BLOCKING + idx = shm_rdrbuff_alloc_b(ai.rdrb, sizeof(*pci), NULL, &sdb, NULL); +#else + idx = shm_rdrbuff_alloc(ai.rdrb, sizeof(*pci), NULL, &sdb); +#endif + if (idx < 0) + return; + + pci = (struct frct_pci *) shm_du_buff_head(sdb); + memset(pci, 0, sizeof(*pci)); + + *((uint32_t *) pci) = hton32(rwe); + + pci->flags = flags; + pci->ackno = hton32(ackno); + + f = &ai.flows[fd]; + + if (crypt_encrypt(&f->crypt, sdb) < 0) + goto fail; + +#ifdef RXM_BLOCKING + if (shm_rbuff_write_b(f->tx_rb, idx, NULL)) +#else + if (shm_rbuff_write(f->tx_rb, idx)) +#endif + goto fail; + + shm_flow_set_notify(f->set, f->info.id, FLOW_PKT); + + return; + + fail: + ipcp_sdb_release(sdb); + return; +} + +static void send_frct_pkt(struct frcti * frcti) +{ + struct timespec now; + time_t diff; + uint32_t ackno; + uint32_t rwe; + int fd; + + assert(frcti); + + clock_gettime(PTHREAD_COND_CLOCK, &now); + + pthread_rwlock_wrlock(&frcti->lock); + + if (!after(frcti->rcv_cr.lwe, frcti->rcv_cr.seqno)) { + pthread_rwlock_unlock(&frcti->lock); + return; + } + + fd = frcti->fd; + ackno = frcti->rcv_cr.lwe; + rwe = frcti->rcv_cr.rwe; + + diff = ts_diff_ns(&frcti->rcv_cr.act, &now); + if (diff > frcti->a) { + pthread_rwlock_unlock(&frcti->lock); + return; + } + + diff = ts_diff_ns(&frcti->snd_cr.act, &now); + if (diff < TICTIME) { + pthread_rwlock_unlock(&frcti->lock); + return; + } + + frcti->rcv_cr.seqno = frcti->rcv_cr.lwe; + + pthread_rwlock_unlock(&frcti->lock); + + __send_frct_pkt(fd, FRCT_ACK | FRCT_FC, ackno, rwe); +} + +static void __send_rdv(int fd) +{ + __send_frct_pkt(fd, FRCT_RDVS, 0, 0); +} + +static struct frcti * frcti_create(int fd, + time_t a, + time_t r, + time_t mpl) +{ + struct frcti * frcti; + ssize_t idx; + struct timespec now; + pthread_condattr_t cattr; +#ifdef PROC_FLOW_STATS + char frctstr[FRCT_NAME_STRLEN + 1]; +#endif + mpl *= BILLION; + a *= BILLION; + r *= BILLION; frcti = malloc(sizeof(*frcti)); if (frcti == NULL) @@ -100,36 +352,79 @@ static struct frcti * frcti_create(int fd) if (pthread_rwlock_init(&frcti->lock, NULL)) goto fail_lock; + if (pthread_mutex_init(&frcti->mtx, NULL)) + goto fail_mutex; + + if (pthread_condattr_init(&cattr)) + goto fail_cattr; +#ifndef __APPLE__ + pthread_condattr_setclock(&cattr, PTHREAD_COND_CLOCK); +#endif + if (pthread_cond_init(&frcti->cond, &cattr)) + goto fail_cond; + +#ifdef PROC_FLOW_STATS + sprintf(frctstr, "%d", fd); + if (rib_reg(frctstr, &r_ops)) + goto fail_rib_reg; +#endif + pthread_condattr_destroy(&cattr); + for (idx = 0; idx < RQ_SIZE; ++idx) frcti->rq[idx] = -1; - clock_gettime(CLOCK_REALTIME_COARSE, &now); + clock_gettime(PTHREAD_COND_CLOCK, &now); - frcti->mpl = DELT_MPL; - frcti->a = DELT_A; - frcti->r = DELT_R; + frcti->mpl = mpl; + frcti->a = a; + frcti->r = r; + frcti->rdv = DELT_RDV; frcti->fd = fd; - delta_t = (frcti->mpl + frcti->a + frcti->r) / 1000; - - frcti->snd_cr.inact = 3 * delta_t; - frcti->snd_cr.act = now.tv_sec - (frcti->snd_cr.inact + 1); - /* rtt estimator. rto is currently srtt + 2 * mdev */ - frcti->srtt_us = 0; /* updated on first ACK */ - frcti->mdev_us = 100000; /* initial rxm will be after 200 ms */ - frcti->rttseq = 0; - frcti->probe = false; - if (ai.flows[fd].qs.loss == 0) { - frcti->snd_cr.cflags |= FRCTFRTX; + frcti->rttseq = 0; + frcti->probe = false; + + frcti->srtt = 0; /* Updated on first ACK */ + frcti->mdev = 10 * MILLION; /* Updated on first ACK */ + frcti->rto = BILLION; /* Initial rxm will be after 1 s */ +#ifdef PROC_FLOW_STATS + frcti->n_rtx = 0; + frcti->n_prb = 0; + frcti->n_rtt = 0; + frcti->n_dup = 0; + frcti->n_dak = 0; + frcti->n_rdv = 0; + frcti->n_out = 0; + frcti->n_rqo = 0; +#endif + if (ai.flows[fd].info.qs.loss == 0) { + frcti->snd_cr.cflags |= FRCTFRTX | FRCTFLINGER; frcti->rcv_cr.cflags |= FRCTFRTX; } - frcti->rcv_cr.inact = 2 * delta_t; - frcti->rcv_cr.act = now.tv_sec - (frcti->rcv_cr.inact + 1); + frcti->snd_cr.cflags |= FRCTFRESCNTL; + + frcti->snd_cr.rwe = START_WINDOW; + + frcti->snd_cr.inact = (3 * mpl + a + r) / BILLION + 1; /* s */ + frcti->snd_cr.act.tv_sec = now.tv_sec - (frcti->snd_cr.inact + 1); + + frcti->rcv_cr.inact = (2 * mpl + a + r) / BILLION + 1; /* s */ + frcti->rcv_cr.act.tv_sec = now.tv_sec - (frcti->rcv_cr.inact + 1); return frcti; +#ifdef PROC_FLOW_STATS + fail_rib_reg: + pthread_cond_destroy(&frcti->cond); +#endif + fail_cond: + pthread_condattr_destroy(&cattr); + fail_cattr: + pthread_mutex_destroy(&frcti->mtx); + fail_mutex: + pthread_rwlock_destroy(&frcti->lock); fail_lock: free(frcti); fail_malloc: @@ -138,23 +433,23 @@ static struct frcti * frcti_create(int fd) static void frcti_destroy(struct frcti * frcti) { - /* - * FIXME: In case of reliable transmission we should - * make sure everything we sent is acked. - */ - - rxmwheel_clear(frcti->fd); - +#ifdef PROC_FLOW_STATS + char frctstr[FRCT_NAME_STRLEN + 1]; + sprintf(frctstr, "%d", frcti->fd); + rib_unreg(frctstr); +#endif + pthread_cond_destroy(&frcti->cond); + pthread_mutex_destroy(&frcti->mtx); pthread_rwlock_destroy(&frcti->lock); free(frcti); } -static uint16_t frcti_getconf(struct frcti * frcti) +static uint16_t frcti_getflags(struct frcti * frcti) { uint16_t ret; - assert (frcti); + assert(frcti); pthread_rwlock_rdlock(&frcti->lock); @@ -165,15 +460,148 @@ static uint16_t frcti_getconf(struct frcti * frcti) return ret; } -#define frcti_queued_pdu(frcti) \ - (frcti == NULL ? -1 : __frcti_queued_pdu(frcti)) +static void frcti_setflags(struct frcti * frcti, + uint16_t flags) +{ + flags |= FRCTFRTX; /* Should not be set by command */ + + assert(frcti); + + pthread_rwlock_wrlock(&frcti->lock); -#define frcti_snd(frcti, sdb) \ + frcti->snd_cr.cflags &= FRCTFRTX; /* Zero other flags */ + + frcti->snd_cr.cflags &= flags; + + pthread_rwlock_unlock(&frcti->lock); +} + +#define frcti_queued_pdu(frcti) \ + (frcti == NULL ? idx : __frcti_queued_pdu(frcti)) + +#define frcti_snd(frcti, sdb) \ (frcti == NULL ? 0 : __frcti_snd(frcti, sdb)) -#define frcti_rcv(frcti, sdb) \ +#define frcti_rcv(frcti, sdb) \ (frcti == NULL ? 0 : __frcti_rcv(frcti, sdb)) +#define frcti_dealloc(frcti) \ + (frcti == NULL ? 0 : __frcti_dealloc(frcti)) + +#define frcti_is_window_open(frcti) \ + (frcti == NULL ? true : __frcti_is_window_open(frcti)) + +#define frcti_window_wait(frcti, abstime) \ + (frcti == NULL ? 0 : __frcti_window_wait(frcti, abstime)) + + +static bool __frcti_is_window_open(struct frcti * frcti) +{ + struct frct_cr * snd_cr = &frcti->snd_cr; + bool ret = true; + + pthread_rwlock_rdlock(&frcti->lock); + + if (snd_cr->cflags & FRCTFRESCNTL) + ret = before(snd_cr->seqno, snd_cr->rwe); + + if (!ret) { + struct timespec now; + + clock_gettime(PTHREAD_COND_CLOCK, &now); + + pthread_mutex_lock(&frcti->mtx); + if (frcti->open) { + frcti->open = false; + frcti->t_wnd = now; + frcti->t_rdvs = now; + } else { + time_t diff; + diff = ts_diff_ns(&frcti->t_wnd, &now); + if (diff > MAX_RDV) { + pthread_mutex_unlock(&frcti->mtx); + pthread_rwlock_unlock(&frcti->lock); + return false; + } + + diff = ts_diff_ns(&frcti->t_rdvs, &now); + if (diff > frcti->rdv) { + frcti->t_rdvs = now; + __send_rdv(frcti->fd); +#ifdef PROC_FLOW_STATS + frcti->n_rdv++; +#endif + + } + } + + pthread_mutex_unlock(&frcti->mtx); + } + + pthread_rwlock_unlock(&frcti->lock); + + return ret; +} + +static int __frcti_window_wait(struct frcti * frcti, + struct timespec * abstime) +{ + struct frct_cr * snd_cr = &frcti->snd_cr; + int ret = 0; + + pthread_rwlock_rdlock(&frcti->lock); + + if (!(snd_cr->cflags & FRCTFRESCNTL)) { + pthread_rwlock_unlock(&frcti->lock); + return 0; + } + + while (snd_cr->seqno == snd_cr->rwe && ret != -ETIMEDOUT) { + struct timespec now; + pthread_rwlock_unlock(&frcti->lock); + pthread_mutex_lock(&frcti->mtx); + + if (frcti->open) { + clock_gettime(PTHREAD_COND_CLOCK, &now); + + frcti->t_wnd = now; + frcti->t_rdvs = now; + frcti->open = false; + } + + pthread_cleanup_push(__cleanup_mutex_unlock, &frcti->mtx); + + ret = -__timedwait(&frcti->cond, &frcti->mtx, abstime); + + pthread_cleanup_pop(false); + + if (ret == -ETIMEDOUT) { + time_t diff; + + clock_gettime(PTHREAD_COND_CLOCK, &now); + + diff = ts_diff_ns(&frcti->t_wnd, &now); + if (diff > MAX_RDV) { + pthread_mutex_unlock(&frcti->mtx); + return -ECONNRESET; /* write fails! */ + } + + diff = ts_diff_ns(&frcti->t_rdvs, &now); + if (diff > frcti->rdv) { + frcti->t_rdvs = now; + __send_rdv(frcti->fd); + } + } + + pthread_mutex_unlock(&frcti->mtx); + pthread_rwlock_rdlock(&frcti->lock); + } + + pthread_rwlock_unlock(&frcti->lock); + + return ret; +} + static ssize_t __frcti_queued_pdu(struct frcti * frcti) { ssize_t idx; @@ -184,10 +612,12 @@ static ssize_t __frcti_queued_pdu(struct frcti * frcti) /* See if we already have the next PDU. */ pthread_rwlock_wrlock(&frcti->lock); - pos = frcti->rcv_cr.seqno & (RQ_SIZE - 1); + pos = frcti->rcv_cr.lwe & (RQ_SIZE - 1); + idx = frcti->rq[pos]; if (idx != -1) { - ++frcti->rcv_cr.seqno; + ++frcti->rcv_cr.lwe; + ++frcti->rcv_cr.rwe; frcti->rq[pos] = -1; } @@ -196,27 +626,60 @@ static ssize_t __frcti_queued_pdu(struct frcti * frcti) return idx; } -static struct frct_pci * frcti_alloc_head(struct shm_du_buff * sdb) +static ssize_t __frcti_pdu_ready(struct frcti * frcti) { - struct frct_pci * pci; + ssize_t idx; + size_t pos; - pci = (struct frct_pci *) shm_du_buff_head_alloc(sdb, FRCT_PCILEN); - if (pci != NULL) - memset(pci, 0, sizeof(*pci)); + assert(frcti); - return pci; -} + /* See if we already have the next PDU. */ + pthread_rwlock_rdlock(&frcti->lock); -static bool before(uint32_t seq1, - uint32_t seq2) -{ - return (int32_t)(seq1 - seq2) < 0; + pos = frcti->rcv_cr.lwe & (RQ_SIZE - 1); + idx = frcti->rq[pos]; + + pthread_rwlock_unlock(&frcti->lock); + + return idx; } -static bool after(uint32_t seq1, - uint32_t seq2) +#include <timerwheel.c> + +/* + * Send a final ACK for everything that has not been ACK'd. + * If the flow should be kept active for retransmission, + * the returned time will be negative. + */ +static time_t __frcti_dealloc(struct frcti * frcti) { - return (int32_t)(seq2 - seq1) < 0; + struct timespec now; + time_t wait; + int ackno; + int fd = -1; + + clock_gettime(PTHREAD_COND_CLOCK, &now); + + pthread_rwlock_rdlock(&frcti->lock); + + ackno = frcti->rcv_cr.lwe; + if (frcti->rcv_cr.lwe != frcti->rcv_cr.seqno) + fd = frcti->fd; + + wait = MAX(frcti->rcv_cr.inact - now.tv_sec + frcti->rcv_cr.act.tv_sec, + frcti->snd_cr.inact - now.tv_sec + frcti->snd_cr.act.tv_sec); + wait = MAX(wait, 0); + + if (frcti->snd_cr.cflags & FRCTFLINGER + && before(frcti->snd_cr.lwe, frcti->snd_cr.seqno)) + wait = -wait; + + pthread_rwlock_unlock(&frcti->lock); + + if (fd != -1) + __send_frct_pkt(fd, FRCT_ACK, ackno, 0); + + return wait; } static int __frcti_snd(struct frcti * frcti, @@ -226,22 +689,29 @@ static int __frcti_snd(struct frcti * frcti, struct timespec now; struct frct_cr * snd_cr; struct frct_cr * rcv_cr; + uint32_t seqno; + bool rtx; assert(frcti); + assert(shm_du_buff_len(sdb) != 0); snd_cr = &frcti->snd_cr; rcv_cr = &frcti->rcv_cr; - rxmwheel_move(); + timerwheel_move(); - pci = frcti_alloc_head(sdb); + pci = (struct frct_pci *) shm_du_buff_head_alloc(sdb, FRCT_PCILEN); if (pci == NULL) - return -1; + return -ENOMEM; + + memset(pci, 0, sizeof(*pci)); - clock_gettime(CLOCK_REALTIME, &now); + clock_gettime(PTHREAD_COND_CLOCK, &now); pthread_rwlock_wrlock(&frcti->lock); + rtx = snd_cr->cflags & FRCTFRTX; + pci->flags |= FRCT_DATA; /* Set DRF if there are no unacknowledged packets. */ @@ -249,145 +719,216 @@ static int __frcti_snd(struct frcti * frcti, pci->flags |= FRCT_DRF; /* Choose a new sequence number if sender inactivity expired. */ - if (now.tv_sec - snd_cr->act > snd_cr->inact) { + if (now.tv_sec - snd_cr->act.tv_sec > snd_cr->inact) { /* There are no unacknowledged packets. */ assert(snd_cr->seqno == snd_cr->lwe); -#ifdef CONFIG_OUROBOROS_DEBUG - snd_cr->seqno = 0; -#else random_buffer(&snd_cr->seqno, sizeof(snd_cr->seqno)); -#endif - frcti->snd_cr.lwe = snd_cr->seqno - 1; + snd_cr->lwe = snd_cr->seqno; + snd_cr->rwe = snd_cr->lwe + START_WINDOW; + } + + seqno = snd_cr->seqno; + pci->seqno = hton32(seqno); + + if (now.tv_sec - rcv_cr->act.tv_sec < rcv_cr->inact) { + pci->flags |= FRCT_FC; + *((uint32_t *) pci) |= hton32(rcv_cr->rwe & 0x00FFFFFF); } - pci->seqno = hton32(snd_cr->seqno); - if (!(snd_cr->cflags & FRCTFRTX)) { + if (!rtx) { snd_cr->lwe++; } else { if (!frcti->probe) { frcti->rttseq = snd_cr->seqno; frcti->t_probe = now; frcti->probe = true; +#ifdef PROC_FLOW_STATS + frcti->n_prb++; +#endif } - - if (now.tv_sec - rcv_cr->act <= rcv_cr->inact) { - rxmwheel_add(frcti, snd_cr->seqno, sdb); - if (rcv_cr->lwe <= rcv_cr->seqno) { - pci->flags |= FRCT_ACK; - pci->ackno = hton32(rcv_cr->seqno); - rcv_cr->lwe = rcv_cr->seqno; - } + if ((now.tv_sec - rcv_cr->act.tv_sec) * BILLION <= frcti->a) { + pci->flags |= FRCT_ACK; + pci->ackno = hton32(rcv_cr->lwe); + rcv_cr->seqno = rcv_cr->lwe; } } snd_cr->seqno++; - snd_cr->act = now.tv_sec; + snd_cr->act = now; pthread_rwlock_unlock(&frcti->lock); + if (rtx) + timerwheel_rxm(frcti, seqno, sdb); + return 0; } static void rtt_estimator(struct frcti * frcti, - time_t mrtt_us) + time_t mrtt) { - time_t srtt = frcti->srtt_us; - time_t mdev = frcti->mdev_us; - - if (srtt != 0) { - srtt -= (srtt >> 3); - srtt += mrtt_us >> 3; /* rtt = 7/8 rtt + 1/8 new */ - mdev -= (mdev >> 2); - mdev += ABS(srtt - mrtt_us) >> 2; + time_t srtt = frcti->srtt; + time_t rttvar = frcti->mdev; + + if (srtt == 0) { /* first measurement */ + srtt = mrtt; + rttvar = mrtt >> 1; } else { - srtt = mrtt_us << 3; /* take the measured time to be rtt */ - mdev = mrtt_us >> 1; /* take half mrtt_us as deviation */ + time_t delta = mrtt - srtt; + srtt += (delta >> 3); + delta = (ABS(delta) - rttvar) >> 2; +#ifdef FRCT_LINUX_RTT_ESTIMATOR + if (delta < 0) + delta >>= 3; +#endif + rttvar += delta; } - - frcti->srtt_us = MAX(1U, srtt); - frcti->mdev_us = MAX(1U, mdev); +#ifdef PROC_FLOW_STATS + frcti->n_rtt++; +#endif + frcti->srtt = MAX(1000L, srtt); + frcti->mdev = MAX(100L, rttvar); + frcti->rto = MAX(RTO_MIN, frcti->srtt + (frcti->mdev << MDEV_MUL)); } -/* Returns 0 when idx contains a packet for the application. */ -static int __frcti_rcv(struct frcti * frcti, - struct shm_du_buff * sdb) +/* Always queues the next application packet on the RQ. */ +static void __frcti_rcv(struct frcti * frcti, + struct shm_du_buff * sdb) { ssize_t idx; + size_t pos; struct frct_pci * pci; struct timespec now; - struct frct_cr * snd_cr; struct frct_cr * rcv_cr; + struct frct_cr * snd_cr; uint32_t seqno; - int ret = 0; + uint32_t ackno; + uint32_t rwe; + int fd = -1; assert(frcti); rcv_cr = &frcti->rcv_cr; snd_cr = &frcti->snd_cr; - pci = (struct frct_pci *) shm_du_buff_head_release(sdb, FRCT_PCILEN); - - clock_gettime(CLOCK_REALTIME, &now); + clock_gettime(PTHREAD_COND_CLOCK, &now); - pthread_rwlock_wrlock(&frcti->lock); + pci = (struct frct_pci *) shm_du_buff_head_release(sdb, FRCT_PCILEN); idx = shm_du_buff_get_idx(sdb); - seqno = ntoh32(pci->seqno); + pos = seqno & (RQ_SIZE - 1); - /* Check if receiver inactivity is true. */ - if (now.tv_sec - rcv_cr->act > rcv_cr->inact) { - /* Inactive receiver, check for DRF. */ - if (pci->flags & FRCT_DRF) /* New run. */ + pthread_rwlock_wrlock(&frcti->lock); + + if (now.tv_sec - rcv_cr->act.tv_sec > rcv_cr->inact) { + if (pci->flags & FRCT_DRF) { /* New run. */ + rcv_cr->lwe = seqno; + rcv_cr->rwe = seqno + RQ_SIZE; rcv_cr->seqno = seqno; - else + } else if (pci->flags & FRCT_DATA) { goto drop_packet; + } } - if (seqno == rcv_cr->seqno) { - ++rcv_cr->seqno; - } else { /* Out of order. */ - if (before(seqno, rcv_cr->seqno)) - goto drop_packet; + rcv_cr->act = now; - if (rcv_cr->cflags & FRCTFRTX) { - size_t pos = seqno & (RQ_SIZE - 1); - if ((seqno - rcv_cr->lwe) > RQ_SIZE /* Out of rq. */ - || frcti->rq[pos] != -1) /* Duplicate in rq. */ - goto drop_packet; - /* Queue. */ - frcti->rq[pos] = idx; - ret = -EAGAIN; - } else { - rcv_cr->seqno = seqno + 1; - } + /* For now, just send an immediate window update. */ + if (pci->flags & FRCT_RDVS) { + fd = frcti->fd; + rwe = rcv_cr->rwe; + pthread_rwlock_unlock(&frcti->lock); + + __send_frct_pkt(fd, FRCT_FC, 0, rwe); + + shm_rdrbuff_remove(ai.rdrb, idx); + return; } - if (rcv_cr->cflags & FRCTFRTX && pci->flags & FRCT_ACK) { - uint32_t ackno = ntoh32(pci->ackno); - /* Check for duplicate (old) acks. */ - if ((int32_t)(ackno - snd_cr->lwe) >= 0) - snd_cr->lwe = ackno; + if (pci->flags & FRCT_ACK) { + ackno = ntoh32(pci->ackno); + if (after(ackno, frcti->snd_cr.lwe)) + frcti->snd_cr.lwe = ackno; + if (frcti->probe && after(ackno, frcti->rttseq)) { - rtt_estimator(frcti, ts_diff_us(&frcti->t_probe, &now)); +#ifdef PROC_FLOW_STATS + if (!(pci->flags & FRCT_DATA)) + frcti->n_dak++; +#endif + rtt_estimator(frcti, ts_diff_ns(&frcti->t_probe, &now)); frcti->probe = false; } } - rcv_cr->act = now.tv_sec; + if (pci->flags & FRCT_FC) { + uint32_t rwe; - pthread_rwlock_unlock(&frcti->lock); + rwe = ntoh32(*((uint32_t *)pci) & hton32(0x00FFFFFF)); + rwe |= snd_cr->rwe & 0xFF000000; + + /* Rollover for 24 bit */ + if (before(rwe, snd_cr->rwe) && snd_cr->rwe - rwe > 0x007FFFFF) + rwe += 0x01000000; + + snd_cr->rwe = rwe; + + pthread_mutex_lock(&frcti->mtx); + if (!frcti->open) { + frcti->open = true; + pthread_cond_broadcast(&frcti->cond); + } + pthread_mutex_unlock(&frcti->mtx); + } if (!(pci->flags & FRCT_DATA)) - shm_rdrbuff_remove(ai.rdrb, idx); + goto drop_packet; - rxmwheel_move(); + if (before(seqno, rcv_cr->lwe)) { + rcv_cr->seqno = seqno; /* Ensures we send a new ACK. */ +#ifdef PROC_FLOW_STATS + frcti->n_dup++; +#endif + goto drop_packet; + } - return ret; + if (rcv_cr->cflags & FRCTFRTX) { + + if (!before(seqno, rcv_cr->rwe)) { /* Out of window. */ +#ifdef PROC_FLOW_STATS + frcti->n_out++; +#endif + goto drop_packet; + } + + if (!before(seqno, rcv_cr->lwe + RQ_SIZE)) { +#ifdef PROC_FLOW_STATS + frcti->n_rqo++; +#endif + goto drop_packet; /* Out of rq. */ + } + if (frcti->rq[pos] != -1) { +#ifdef PROC_FLOW_STATS + frcti->n_dup++; +#endif + goto drop_packet; /* Duplicate in rq. */ + } + fd = frcti->fd; + } else { + rcv_cr->lwe = seqno; + } + + frcti->rq[pos] = idx; + + pthread_rwlock_unlock(&frcti->lock); + + if (fd != -1) + timerwheel_delayed_ack(fd, frcti); + + return; drop_packet: pthread_rwlock_unlock(&frcti->lock); shm_rdrbuff_remove(ai.rdrb, idx); - rxmwheel_move(); - return -EAGAIN; + send_frct_pkt(frcti); + return; } diff --git a/src/lib/hash.c b/src/lib/hash.c index e5f90ee0..b465f894 100644 --- a/src/lib/hash.c +++ b/src/lib/hash.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Hashing * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This implementation is adapted and redistributed from the RHASH * project @@ -29,43 +29,50 @@ #include "config.h" +#include <ouroboros/endian.h> #include <ouroboros/hash.h> -#ifndef HAVE_LIBGCRYPT +#ifdef HAVE_LIBGCRYPT +#include <gcrypt.h> +#else #include <ouroboros/crc32.h> #include <ouroboros/md5.h> #include <ouroboros/sha3.h> -#else -#include <gcrypt.h> #endif #include <string.h> #include <assert.h> #include <stdbool.h> +#ifdef HAVE_LIBGCRYPT +int gcry_algo_tbl [] = { + /* DIR_HASH policies first */ + GCRY_MD_SHA3_224, + GCRY_MD_SHA3_256, + GCRY_MD_SHA3_384, + GCRY_MD_SHA3_512, + /* Below for internal use only */ + GCRY_MD_CRC32, + GCRY_MD_MD5, +}; +#else +int hash_len_tbl [] = { + /* DIR_HASH policies first */ + SHA3_224_HASH_LEN, + SHA3_256_HASH_LEN, + SHA3_384_HASH_LEN, + SHA3_512_HASH_LEN, + /* Below for internal use only */ + CRC32_HASH_LEN, + MD5_HASH_LEN +}; +#endif + uint16_t hash_len(enum hash_algo algo) { #ifdef HAVE_LIBGCRYPT - return (uint16_t) gcry_md_get_algo_dlen(algo); + return (uint16_t) gcry_md_get_algo_dlen(gcry_algo_tbl[algo]); #else - switch (algo) { - case HASH_CRC32: - return CRC32_HASH_LEN; - case HASH_MD5: - return MD5_HASH_LEN; - case HASH_SHA3_224: - return SHA3_224_HASH_LEN; - case HASH_SHA3_256: - return SHA3_256_HASH_LEN; - case HASH_SHA3_384: - return SHA3_384_HASH_LEN; - case HASH_SHA3_512: - return SHA3_512_HASH_LEN; - default: - assert(false); - break; - } - - return 0; + return hash_len_tbl[algo]; #endif } @@ -75,7 +82,7 @@ void mem_hash(enum hash_algo algo, size_t len) { #ifdef HAVE_LIBGCRYPT - gcry_md_hash_buffer(algo, dst, buf, len); + gcry_md_hash_buffer(gcry_algo_tbl[algo], dst, buf, len); #else struct sha3_ctx sha3_ctx; struct md5_ctx md5_ctx; @@ -84,6 +91,7 @@ void mem_hash(enum hash_algo algo, case HASH_CRC32: memset(dst, 0, CRC32_HASH_LEN); crc32((uint32_t *) dst, buf, len); + *(uint32_t *) dst = htobe32(*(uint32_t *) dst); break; case HASH_MD5: rhash_md5_init(&md5_ctx); diff --git a/src/lib/ipcp_config.proto b/src/lib/ipcp_config.proto deleted file mode 100644 index 23c65e94..00000000 --- a/src/lib/ipcp_config.proto +++ /dev/null @@ -1,62 +0,0 @@ -/* - * Ouroboros - Copyright (C) 2016 - 2020 - * - * Layer configuration message - * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> - * - * 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/. - */ - -syntax = "proto2"; - -message layer_info_msg { - required string layer_name = 1; - required uint32 dir_hash_algo = 2; -} - -message ipcp_config_msg { - required layer_info_msg layer_info = 1; - required int32 ipcp_type = 2; - // Config for unicast IPCP - optional uint32 addr_size = 3; - optional uint32 eid_size = 4; - optional uint32 max_ttl = 5; - optional uint32 addr_auth_type = 6; - optional uint32 routing_type = 7; - // Config for UDP - optional uint32 ip_addr = 8; - optional uint32 dns_addr = 9; - optional uint32 clt_port = 10; - optional uint32 srv_port = 11; - // Config for the Ethernet - optional string dev = 12; - // Config for DIX Ethernet - optional uint32 ethertype = 13; -} - -enum enroll_code { - ENROLL_REQ = 1; - ENROLL_BOOT = 2; - ENROLL_DONE = 4; -}; - -message enroll_msg { - required enroll_code code = 1; - optional ipcp_config_msg conf = 2; - optional int64 t_sec = 3; - optional uint32 t_nsec = 4; - optional int32 result = 5; -};
\ No newline at end of file diff --git a/src/lib/irm.c b/src/lib/irm.c index 08dffb6c..d25101f3 100644 --- a/src/lib/irm.c +++ b/src/lib/irm.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * The API to instruct the IRM * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License @@ -31,28 +31,37 @@ #include <ouroboros/irm.h> #include <ouroboros/utils.h> #include <ouroboros/sockets.h> +#include <ouroboros/protobuf.h> #include <stdbool.h> #include <string.h> #include <stdlib.h> #include <sys/stat.h> -pid_t irm_create_ipcp(const char * name, - enum ipcp_type type) +int irm_create_ipcp(const char * name, + enum ipcp_type type) { - irm_msg_t msg = IRM_MSG__INIT; - irm_msg_t * recv_msg = NULL; - int ret = -1; + irm_msg_t msg = IRM_MSG__INIT; + irm_msg_t * recv_msg; + int ret; + struct ipcp_info info; - if (name == NULL) + if (name == NULL || strlen(name) > IPCP_NAME_SIZE) return -EINVAL; - msg.code = IRM_MSG_CODE__IRM_CREATE_IPCP; - msg.name = (char *) name; - msg.has_ipcp_type = true; - msg.ipcp_type = type; + memset(&info, 0, sizeof(info)); + + info.type = type; + strcpy(info.name, name); + + msg.code = IRM_MSG_CODE__IRM_CREATE_IPCP; + msg.ipcp_info = ipcp_info_s_to_msg(&info); + if (msg.ipcp_info == NULL) + return -ENOMEM; recv_msg = send_recv_irm_msg(&msg); + ipcp_info_msg__free_unpacked(msg.ipcp_info, NULL); + if (recv_msg == NULL) return -EIRMD; @@ -98,11 +107,9 @@ int irm_destroy_ipcp(pid_t pid) int irm_bootstrap_ipcp(pid_t pid, const struct ipcp_config * conf) { - irm_msg_t msg = IRM_MSG__INIT; - ipcp_config_msg_t config = IPCP_CONFIG_MSG__INIT; - layer_info_msg_t layer_info = LAYER_INFO_MSG__INIT; - irm_msg_t * recv_msg = NULL; - int ret = -1; + irm_msg_t msg = IRM_MSG__INIT; + irm_msg_t * recv_msg; + int ret; if (pid == -1 || conf == NULL) return -EINVAL; @@ -110,56 +117,10 @@ int irm_bootstrap_ipcp(pid_t pid, msg.code = IRM_MSG_CODE__IRM_BOOTSTRAP_IPCP; msg.has_pid = true; msg.pid = pid; - - config.layer_info = &layer_info; - msg.conf = &config; - layer_info.layer_name = (char *) conf->layer_info.layer_name; - - config.ipcp_type = conf->type; - - if (conf->type != IPCP_UDP) - layer_info.dir_hash_algo = conf->layer_info.dir_hash_algo; - - switch (conf->type) { - case IPCP_UNICAST: - config.has_addr_size = true; - config.addr_size = conf->addr_size; - config.has_eid_size = true; - config.eid_size = conf->eid_size; - config.has_max_ttl = true; - config.max_ttl = conf->max_ttl; - config.has_addr_auth_type = true; - config.addr_auth_type = conf->addr_auth_type; - config.has_routing_type = true; - config.routing_type = conf->routing_type; - break; - case IPCP_UDP: - config.has_ip_addr = true; - config.ip_addr = conf->ip_addr; - config.has_dns_addr = true; - config.dns_addr = conf->dns_addr; - config.has_srv_port = true; - config.srv_port = conf->srv_port; - config.has_clt_port = true; - config.clt_port = conf->clt_port; - break; - case IPCP_LOCAL: - case IPCP_BROADCAST: - case IPCP_RAPTOR: - break; - case IPCP_ETH_LLC: - config.dev = conf->dev; - break; - case IPCP_ETH_DIX: - config.dev = conf->dev; - config.has_ethertype = true; - config.ethertype = conf->ethertype; - break; - default: - return -EIPCPTYPE; - } + msg.conf = ipcp_config_s_to_msg(conf); recv_msg = send_recv_irm_msg(&msg); + ipcp_config_msg__free_unpacked(msg.conf, NULL); if (recv_msg == NULL) return -EIRMD; @@ -179,8 +140,7 @@ int irm_connect_ipcp(pid_t pid, const char * component, qosspec_t qs) { - irm_msg_t msg = IRM_MSG__INIT; - qosspec_msg_t qs_msg = QOSSPEC_MSG__INIT; + irm_msg_t msg = IRM_MSG__INIT; irm_msg_t * recv_msg; int ret; @@ -190,10 +150,11 @@ int irm_connect_ipcp(pid_t pid, msg.comp = (char *) component; msg.has_pid = true; msg.pid = pid; - qs_msg = spec_to_msg(&qs); - msg.qosspec = &qs_msg; + msg.qosspec = qos_spec_s_to_msg(&qs); recv_msg = send_recv_irm_msg(&msg); + qosspec_msg__free_unpacked(msg.qosspec, NULL); + if (recv_msg == NULL) return -EIRMD; @@ -237,7 +198,7 @@ int irm_disconnect_ipcp(pid_t pid, return ret; } -ssize_t irm_list_ipcps(struct ipcp_info ** ipcps) +ssize_t irm_list_ipcps(struct ipcp_list_info ** ipcps) { irm_msg_t msg = IRM_MSG__INIT; irm_msg_t * recv_msg; @@ -407,50 +368,69 @@ int irm_bind_program(const char * prog, int argc, char ** argv) { - irm_msg_t msg = IRM_MSG__INIT; - irm_msg_t * recv_msg = NULL; - int ret = -1; - char * full_name; + irm_msg_t msg = IRM_MSG__INIT; + irm_msg_t * recv_msg; + char ** exec; + int ret; + int i; if (prog == NULL || name == NULL) return -EINVAL; - full_name = strdup(prog); - if (full_name == NULL) - return -ENOMEM; + exec = malloc((argc + 2) * sizeof(*exec)); + if (exec== NULL) { + ret = -ENOMEM; + goto fail_exec; + } - if ((ret = check_prog_path(&full_name)) < 0) { - free(full_name); - return ret; + exec[0] = strdup(prog); + if (exec[0] == NULL) { + ret = -ENOMEM; + goto fail_exec0; } + ret = check_prog_path(&exec[0]); + if (ret < 0) + goto fail; + + for (i = 0; i < argc; i++) + exec[i + 1] = argv[i]; + + exec[argc + 1] = ""; + msg.code = IRM_MSG_CODE__IRM_BIND_PROGRAM; msg.name = (char *) name; - msg.prog = full_name; - if (argv != NULL) { - msg.n_args = argc; - msg.args = (char **) argv; - } + msg.n_exec = argc + 2; + msg.exec = exec; msg.has_opts = true; msg.opts = opts; recv_msg = send_recv_irm_msg(&msg); - - free(full_name); - - if (recv_msg == NULL) - return -EIRMD; + if (recv_msg == NULL) { + ret = -EIRMD; + goto fail; + } if (!recv_msg->has_result) { irm_msg__free_unpacked(recv_msg, NULL); - return -1; + ret = -EPERM; + goto fail; } ret = recv_msg->result; irm_msg__free_unpacked(recv_msg, NULL); + free(exec[0]); + free(exec); + + return ret; + fail: + free(exec[0]); + fail_exec0: + free(exec); + fail_exec: return ret; } diff --git a/src/lib/irmd_messages.proto b/src/lib/irmd_messages.proto deleted file mode 100644 index 5b23ee9d..00000000 --- a/src/lib/irmd_messages.proto +++ /dev/null @@ -1,90 +0,0 @@ -/* - * Ouroboros - Copyright (C) 2016 - 2020 - * - * IRMd message - * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> - * - * 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/. - */ - -syntax = "proto2"; - -import "ipcp_config.proto"; -import "qosspec.proto"; - -enum irm_msg_code { - IRM_CREATE_IPCP = 1; - IPCP_CREATE_R = 2; - IRM_DESTROY_IPCP = 3; - IRM_LIST_IPCPS = 4; - IRM_BOOTSTRAP_IPCP = 5; - IRM_ENROLL_IPCP = 6; - IRM_CONNECT_IPCP = 7; - IRM_DISCONNECT_IPCP = 8; - IRM_BIND_PROGRAM = 9; - IRM_UNBIND_PROGRAM = 10; - IRM_PROC_ANNOUNCE = 11; - IRM_BIND_PROCESS = 12; - IRM_UNBIND_PROCESS = 13; - IRM_CREATE_NAME = 14; - IRM_DESTROY_NAME = 15; - IRM_LIST_NAMES = 16; - IRM_REG_NAME = 17; - IRM_UNREG_NAME = 18; - IRM_FLOW_ALLOC = 19; - IRM_FLOW_ACCEPT = 20; - IRM_FLOW_JOIN = 21; - IRM_FLOW_DEALLOC = 22; - IPCP_FLOW_REQ_ARR = 23; - IPCP_FLOW_ALLOC_REPLY = 24; - IRM_REPLY = 25; -}; - -message ipcp_info_msg { - required uint32 pid = 1; - required uint32 type = 2; - required string name = 3; - required string layer = 4; -}; - -message name_info_msg { - required string name = 1; - required uint32 pol_lb = 2; -}; - -message irm_msg { - required irm_msg_code code = 1; - optional string prog = 2; - optional sint32 pid = 3; - optional string name = 4; - optional uint32 ipcp_type = 5; - optional string layer = 6; - repeated string args = 7; - optional sint32 response = 8; - optional string dst = 9; - optional bytes hash = 10; - optional sint32 flow_id = 11; - optional qosspec_msg qosspec = 12; - optional ipcp_config_msg conf = 13; - optional uint32 opts = 14; - repeated ipcp_info_msg ipcps = 15; - repeated name_info_msg names = 16; - optional uint32 timeo_sec = 17; - optional uint32 timeo_nsec = 18; - optional string comp = 19; - optional bytes pk = 20; /* piggyback */ - optional sint32 result = 21; -}; diff --git a/src/lib/list.c b/src/lib/list.c index 83950a5c..62b2eb27 100644 --- a/src/lib/list.c +++ b/src/lib/list.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Simple doubly linked list implementation. * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License @@ -62,10 +62,10 @@ void list_add_tail(struct list_head * n, void list_del(struct list_head * e) { del_list(e->prv, e->nxt); - e->nxt = e->prv = NULL; + e->nxt = e->prv = e; } -bool list_is_empty(struct list_head * h) +bool list_is_empty(const struct list_head * h) { return h->nxt == h; } diff --git a/src/lib/lockfile.c b/src/lib/lockfile.c index ac3c3062..cf6d3c94 100644 --- a/src/lib/lockfile.c +++ b/src/lib/lockfile.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Lockfile for Ouroboros * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License @@ -36,81 +36,69 @@ #include <sys/stat.h> #define LF_SIZE (sizeof(pid_t)) +#define LF_PROT (PROT_READ | PROT_WRITE) struct lockfile { pid_t * pid; }; -struct lockfile * lockfile_create() { - int fd; - mode_t mask; - struct lockfile * lf = malloc(sizeof(*lf)); - if (lf == NULL) - return NULL; - - mask = umask(0); +static struct lockfile * __lockfile_open(int oflag) +{ + int fd; + struct lockfile * lf; - fd = shm_open(SHM_LOCKFILE_NAME, O_CREAT | O_EXCL | O_RDWR, 0666); - if (fd == -1) { - free(lf); - return NULL; - } + lf = malloc(sizeof(*lf)); + if (lf == NULL) + goto fail_lockfile; - umask(mask); + fd = shm_open(SHM_LOCKFILE_NAME, oflag, 0666); + if (fd == -1) + goto fail_shm_open; - if (ftruncate(fd, LF_SIZE - 1) < 0) { - free(lf); - return NULL; - } + if ((oflag & O_CREAT) && ftruncate(fd, LF_SIZE) < 0) + goto fail_truncate; - lf->pid = mmap(NULL, - LF_SIZE, PROT_READ | PROT_WRITE, - MAP_SHARED, - fd, - 0); + lf->pid = mmap(NULL, LF_SIZE, LF_PROT, MAP_SHARED, fd, 0); + if (lf->pid == MAP_FAILED) + goto fail_mmap; close (fd); - if (lf->pid == MAP_FAILED) { - shm_unlink(SHM_LOCKFILE_NAME); - free(lf); - return NULL; - } - - *lf->pid = getpid(); - return lf; + + fail_mmap: + shm_unlink(SHM_LOCKFILE_NAME); + fail_truncate: + close(fd); + fail_shm_open: + free(lf); + fail_lockfile: + return NULL; } -struct lockfile * lockfile_open() { - int fd; - struct lockfile * lf = malloc(sizeof(*lf)); - if (lf == NULL) - return NULL; +struct lockfile * lockfile_create(void) +{ + struct lockfile * lf; + mode_t mask; - fd = shm_open(SHM_LOCKFILE_NAME, O_RDWR, 0666); - if (fd < 0) { - free(lf); - return NULL; - } + mask = umask(0); - lf->pid = mmap(NULL, - LF_SIZE, PROT_READ | PROT_WRITE, - MAP_SHARED, - fd, - 0); + lf = __lockfile_open(O_CREAT | O_EXCL | O_RDWR); + if (lf == NULL) + return NULL; - close(fd); + umask(mask); - if (lf->pid == MAP_FAILED) { - shm_unlink(SHM_LOCKFILE_NAME); - free(lf); - return NULL; - } + *lf->pid = getpid(); return lf; } +struct lockfile * lockfile_open(void) +{ + return __lockfile_open(O_RDWR); +} + void lockfile_close(struct lockfile * lf) { assert(lf); @@ -127,11 +115,9 @@ void lockfile_destroy(struct lockfile * lf) if (getpid() != *lf->pid && kill(*lf->pid, 0) == 0) return; - munmap(lf->pid, LF_SIZE); + lockfile_close(lf); shm_unlink(SHM_LOCKFILE_NAME); - - free(lf); } pid_t lockfile_owner(struct lockfile * lf) diff --git a/src/lib/logs.c b/src/lib/logs.c index 986b01ec..d90bcd63 100644 --- a/src/lib/logs.c +++ b/src/lib/logs.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Logging facilities * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License diff --git a/src/lib/md5.c b/src/lib/md5.c index 4dfa8f66..ad0dd4d7 100644 --- a/src/lib/md5.c +++ b/src/lib/md5.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * MD5 algorithm * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This implementation is adapted and redistributed from the RHASH * project diff --git a/src/lib/notifier.c b/src/lib/notifier.c index dd211dd3..45745b9a 100644 --- a/src/lib/notifier.c +++ b/src/lib/notifier.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Notifier event system using callbacks * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License @@ -25,8 +25,9 @@ #include <ouroboros/errno.h> #include <ouroboros/notifier.h> #include <ouroboros/list.h> +#include <ouroboros/utils.h> +#include <ouroboros/pthread.h> -#include <pthread.h> #include <stdlib.h> struct listener { @@ -75,8 +76,7 @@ void notifier_event(int event, pthread_rwlock_rdlock(¬ifier.lock); - pthread_cleanup_push((void (*) (void *)) pthread_rwlock_unlock, - (void *) ¬ifier.lock) + pthread_cleanup_push(__cleanup_rwlock_unlock, ¬ifier.lock) list_for_each(p, ¬ifier.listeners) { struct listener * l = list_entry(p, struct listener, next); diff --git a/src/lib/cacep.proto b/src/lib/pb/cep.proto index 4f4ae9df..d31cf4f7 100644 --- a/src/lib/cacep.proto +++ b/src/lib/pb/cep.proto @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * - * Message for Connection Information in CACEP + * Message for Connection Information in OCEP * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License @@ -27,7 +27,7 @@ message fixed_conc_syntax_msg { repeated uint32 lens = 2; } -message cacep_msg { +message cep_msg { required string comp_name = 1; required string protocol = 2; required int32 pref_version = 3; @@ -36,4 +36,4 @@ message cacep_msg { repeated int32 supp_syntax = 6; optional fixed_conc_syntax_msg syntax_spec = 7; required uint64 address = 8; -}
\ No newline at end of file +} diff --git a/src/lib/pb/enroll.proto b/src/lib/pb/enroll.proto new file mode 100644 index 00000000..7fe612a8 --- /dev/null +++ b/src/lib/pb/enroll.proto @@ -0,0 +1,42 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2024 + * + * Enrollment protocol + * + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> + * + * 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/. + */ + +syntax = "proto2"; +import "ipcp_config.proto"; + +message enroll_req_msg { + /* TODO authentication */ + required bytes id = 1; +} + +message enroll_resp_msg { + required bytes id = 1; + required int64 t_sec = 2; + required int32 t_nsec = 3; + required int32 response = 4; + optional ipcp_config_msg conf = 5; +} + +message enroll_ack_msg { + required bytes id = 1; + required int32 result = 2; +} diff --git a/src/lib/ipcpd_messages.proto b/src/lib/pb/ipcp.proto index b0efe9ab..c2c7f48b 100644 --- a/src/lib/ipcpd_messages.proto +++ b/src/lib/pb/ipcp.proto @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * - * IPCPd message + * Messages sent to IPCPds * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License @@ -23,7 +23,8 @@ syntax = "proto2"; import "ipcp_config.proto"; -import "qosspec.proto"; +import "model.proto"; + enum ipcp_msg_code { IPCP_BOOTSTRAP = 1; @@ -38,7 +39,7 @@ enum ipcp_msg_code { IPCP_CONNECT = 10; IPCP_DISCONNECT = 11; IPCP_REPLY = 12; -}; +} message ipcp_msg { required ipcp_msg_code code = 1; @@ -52,5 +53,7 @@ message ipcp_msg { optional layer_info_msg layer_info = 9; optional int32 response = 10; optional string comp = 11; - optional int32 result = 12; -}; + optional uint32 timeo_sec = 12; + optional sint32 mpl = 13; + optional int32 result = 14; +} diff --git a/src/lib/pb/ipcp_config.proto b/src/lib/pb/ipcp_config.proto new file mode 100644 index 00000000..28528b0c --- /dev/null +++ b/src/lib/pb/ipcp_config.proto @@ -0,0 +1,57 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2024 + * + * Layer configuration message + * + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> + * + * 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/. + */ + +syntax = "proto2"; + +import "model.proto"; + +message dt_config_msg { + required uint32 addr_size = 1; + required uint32 eid_size = 2; + required uint32 max_ttl = 3; + required uint32 routing_type = 4; +} + +message uni_config_msg { + required dt_config_msg dt = 1; + required uint32 addr_auth_type = 2; + required uint32 cong_avoid = 3; +} + +message eth_config_msg { + required string dev = 1; + required uint32 ethertype = 2; +} + +message udp_config_msg { + required uint32 ip_addr = 1; + required uint32 port = 2; + required uint32 dns_addr = 3; /* set to 0 if unused */ +} + +message ipcp_config_msg { + required layer_info_msg layer_info = 1; + required uint32 ipcp_type = 2; + optional uni_config_msg unicast = 3; + optional udp_config_msg udp = 4; + optional eth_config_msg eth = 5; +} diff --git a/src/lib/pb/irm.proto b/src/lib/pb/irm.proto new file mode 100644 index 00000000..da3bd982 --- /dev/null +++ b/src/lib/pb/irm.proto @@ -0,0 +1,97 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2024 + * + * Messages sent to IRMd + * + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> + * + * 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/. + */ + +syntax = "proto2"; + +import "ipcp_config.proto"; +import "model.proto"; + +enum irm_msg_code { + IRM_CREATE_IPCP = 1; + IPCP_CREATE_R = 2; + IRM_DESTROY_IPCP = 3; + IRM_LIST_IPCPS = 4; + IRM_BOOTSTRAP_IPCP = 5; + IRM_ENROLL_IPCP = 6; + IRM_CONNECT_IPCP = 7; + IRM_DISCONNECT_IPCP = 8; + IRM_BIND_PROGRAM = 9; + IRM_UNBIND_PROGRAM = 10; + IRM_PROC_ANNOUNCE = 11; + IRM_PROC_EXIT = 12; + IRM_BIND_PROCESS = 13; + IRM_UNBIND_PROCESS = 14; + IRM_CREATE_NAME = 15; + IRM_DESTROY_NAME = 16; + IRM_LIST_NAMES = 17; + IRM_REG_NAME = 18; + IRM_UNREG_NAME = 19; + IRM_FLOW_ALLOC = 20; + IRM_FLOW_ACCEPT = 21; + IRM_FLOW_JOIN = 22; + IRM_FLOW_DEALLOC = 23; + IPCP_FLOW_DEALLOC = 24; + IPCP_FLOW_REQ_ARR = 25; + IPCP_FLOW_ALLOC_REPLY = 26; + IRM_REPLY = 27; +} + +message timespec_msg { + required uint64 tv_sec = 1; + required uint32 tv_nsec = 2; +} + +message ipcp_list_msg { + required uint32 pid = 1; + required uint32 type = 2; + required string name = 3; + required string layer = 4; + required uint32 hash_algo = 5; +} + +message irm_msg { + required irm_msg_code code = 1; + optional string prog = 2; + optional sint32 pid = 3; + optional string name = 4; + optional flow_info_msg flow_info = 5; + optional ipcp_info_msg ipcp_info = 6; + optional string layer = 7; + repeated string exec = 8; + optional sint32 response = 9; + optional string dst = 10; + optional bytes hash = 11; + optional sint32 flow_id = 12; + optional qosspec_msg qosspec = 13; + optional ipcp_config_msg conf = 14; + optional uint32 opts = 15; + repeated ipcp_list_msg ipcps = 16; + repeated name_info_msg names = 17; + optional timespec_msg timeo = 18; + optional sint32 mpl = 20; + optional string comp = 21; + optional bytes pk = 22; /* piggyback */ + optional bytes symmkey = 23; + optional uint32 timeo_sec = 24; + optional uint32 timeo_nsec = 25; + optional sint32 result = 26; +} diff --git a/src/lib/pb/model.proto b/src/lib/pb/model.proto new file mode 100644 index 00000000..f1e401f9 --- /dev/null +++ b/src/lib/pb/model.proto @@ -0,0 +1,61 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2024 + * + * Model description messages + * + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> + * + * 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/. + */ + +syntax = "proto2"; + +message qosspec_msg { + required uint32 delay = 1; /* In ms. */ + required uint64 bandwidth = 2; /* In bits/s. */ + required uint32 availability = 3; /* Class of 9s. */ + required uint32 loss = 4; /* Packet loss. */ + required uint32 ber = 5; /* Bit error rate, ppb. */ + required uint32 in_order = 6; /* In-order delivery. */ + required uint32 max_gap = 7; /* In ms. */ + required uint32 cypher_s = 8; /* Crypto strength in bits. */ + required uint32 timeout = 9; /* Timeout in ms. */ +} + +message flow_info_msg { + required uint32 id = 1; + required uint32 n_pid = 2; + required uint32 n_1_pid = 3; + required uint32 mpl = 4; + required uint32 state = 5; + required qosspec_msg qos = 6; +} + +message name_info_msg { + required string name = 1; + required uint32 pol_lb = 2; +} + +message layer_info_msg { + required string name = 1; + required uint32 dir_hash_algo = 2; +} + +message ipcp_info_msg { + required uint32 type = 1; + required string name = 2; + required uint32 pid = 3; + required uint32 state = 4; +} diff --git a/src/lib/protobuf.c b/src/lib/protobuf.c new file mode 100644 index 00000000..b586168c --- /dev/null +++ b/src/lib/protobuf.c @@ -0,0 +1,605 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2024 + * + * Protobuf syntax conversion + * + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> + * + * 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 _DEFAULT_SOURCE + +#include <ouroboros/protobuf.h> + +#include <stdlib.h> +#include <string.h> +#include <time.h> + +timespec_msg_t * timespec_s_to_msg(const struct timespec * s) +{ + timespec_msg_t * msg; + + assert(s != NULL); + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_malloc; + + timespec_msg__init(msg); + + msg->tv_sec = s->tv_sec; + msg->tv_nsec = s->tv_nsec; + + return msg; + + fail_malloc: + return NULL; +} + +struct timespec timespec_msg_to_s(timespec_msg_t * msg) +{ + struct timespec s; + + assert(msg != NULL); + + s.tv_sec = msg->tv_sec; + s.tv_nsec = msg->tv_nsec; + + return s; +} + +flow_info_msg_t * flow_info_s_to_msg(const struct flow_info * s) +{ + flow_info_msg_t * msg; + + assert(s != NULL); + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_malloc; + + flow_info_msg__init(msg); + + msg->id = s->id; + msg->n_pid = s->n_pid; + msg->n_1_pid = s->n_1_pid; + msg->mpl = s->mpl; + msg->state = s->state; + msg->qos = qos_spec_s_to_msg(&s->qs); + if (msg->qos == NULL) + goto fail_msg; + + return msg; + + fail_msg: + flow_info_msg__free_unpacked(msg, NULL); + fail_malloc: + return NULL; +} + +struct flow_info flow_info_msg_to_s(const flow_info_msg_t * msg) +{ + struct flow_info s; + + assert(msg != NULL); + + s.id = msg->id; + s.n_pid = msg->n_pid; + s.n_1_pid = msg->n_1_pid; + s.mpl = msg->mpl; + s.state = msg->state; + s.qs = qos_spec_msg_to_s(msg->qos); + + return s; +} + +layer_info_msg_t * layer_info_s_to_msg(const struct layer_info * s) +{ + layer_info_msg_t * msg; + + assert(s != NULL); + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_malloc; + + layer_info_msg__init(msg); + + msg->name = strdup(s->name); + if (msg->name == NULL) + goto fail_msg; + + msg->dir_hash_algo = s->dir_hash_algo; + + return msg; + + fail_msg: + layer_info_msg__free_unpacked(msg, NULL); + fail_malloc: + return NULL; +} + +struct layer_info layer_info_msg_to_s(const layer_info_msg_t * msg) +{ + struct layer_info s; + + assert(msg != NULL); + assert(strlen(msg->name) <= LAYER_NAME_SIZE); + + s.dir_hash_algo = msg->dir_hash_algo; + strcpy(s.name, msg->name); + + return s; +} + +ipcp_info_msg_t * ipcp_info_s_to_msg(const struct ipcp_info * s) +{ + ipcp_info_msg_t * msg; + + assert(s != NULL); + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_malloc; + + ipcp_info_msg__init(msg); + + msg->name = strdup(s->name); + if (msg->name == NULL) + goto fail_msg; + + msg->type = s->type; + msg->pid = s->pid; + msg->state = s->state; + + return msg; + fail_msg: + ipcp_info_msg__free_unpacked(msg, NULL); + fail_malloc: + return NULL; +} + +struct ipcp_info ipcp_info_msg_to_s(const ipcp_info_msg_t * msg) +{ + struct ipcp_info s; + + assert(msg != NULL); + assert(msg->name != NULL); + assert(strlen(msg->name) <= NAME_SIZE); + + strcpy(s.name, msg->name); + s.type = msg->type; + s.pid = msg->pid; + s.state = msg->state; + + return s; +} + +dt_config_msg_t * dt_config_s_to_msg(const struct dt_config * s) +{ + dt_config_msg_t * msg; + + assert(s != NULL); + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + return NULL; + + dt_config_msg__init(msg); + + msg->addr_size = s->addr_size; + msg->eid_size = s->eid_size; + msg->max_ttl = s->max_ttl; + msg->routing_type = s->routing_type; + + return msg; +} + +struct dt_config dt_config_msg_to_s(const dt_config_msg_t * msg) +{ + struct dt_config s; + + assert(msg != NULL); + + s.addr_size = msg->addr_size; + s.eid_size = msg->eid_size; + s.max_ttl = msg->max_ttl; + s.routing_type = msg->routing_type; + + return s; +} + +uni_config_msg_t * uni_config_s_to_msg(const struct uni_config * s) +{ + uni_config_msg_t * msg; + + assert(s != NULL); + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_malloc; + + uni_config_msg__init(msg); + + msg->dt = dt_config_s_to_msg(&s->dt); + if (msg->dt == NULL) + goto fail_msg; + + msg->addr_auth_type = s->addr_auth_type; + msg->cong_avoid = s->cong_avoid; + + return msg; + + fail_msg: + uni_config_msg__free_unpacked(msg, NULL); + fail_malloc: + return NULL; +} + +struct uni_config uni_config_msg_to_s(const uni_config_msg_t * msg) +{ + struct uni_config s; + + s.dt = dt_config_msg_to_s(msg->dt); + + s.addr_auth_type = msg->addr_auth_type; + s.cong_avoid = msg->cong_avoid; + + return s; +} + +udp_config_msg_t * udp_config_s_to_msg(const struct udp_config * s) +{ + udp_config_msg_t * msg; + + assert(s != NULL); + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + return NULL; + + udp_config_msg__init(msg); + + msg->ip_addr = s->ip_addr; + msg->dns_addr = s->dns_addr; + msg->port = s->port; + + return msg; +} + +struct udp_config udp_config_msg_to_s(const udp_config_msg_t * msg) +{ + struct udp_config s; + + assert(msg != NULL); + + s.ip_addr = msg->ip_addr; + s.dns_addr = msg->dns_addr; + s.port = msg->port; + + return s; +} + +eth_config_msg_t * eth_config_s_to_msg(const struct eth_config * s) +{ + eth_config_msg_t * msg; + + assert(s != NULL); + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_malloc; + + eth_config_msg__init(msg); + + msg->dev = strdup(s->dev); + if (msg->dev == NULL) + goto fail_msg; + + msg->ethertype = s->ethertype; + + return msg; + + fail_msg: + eth_config_msg__free_unpacked(msg, NULL); + fail_malloc: + return NULL; +} + +struct eth_config eth_config_msg_to_s(const eth_config_msg_t * msg) +{ + struct eth_config s; + + assert(msg != NULL); + assert(strlen(msg->dev) <= DEV_NAME_SIZE); + + strcpy(s.dev, msg->dev); + s.ethertype = msg->ethertype; + + return s; +} + + +ipcp_config_msg_t * ipcp_config_s_to_msg(const struct ipcp_config * s) +{ + ipcp_config_msg_t * msg; + + assert(s != NULL); + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_malloc; + + ipcp_config_msg__init(msg); + + switch (s->type) { + case IPCP_LOCAL: + break; + case IPCP_UNICAST: + msg->unicast = uni_config_s_to_msg(&s->unicast); + if (msg->unicast == NULL) + goto fail_msg; + break; + case IPCP_BROADCAST: + break; + case IPCP_ETH_LLC: + /* FALLTHRU */ + case IPCP_ETH_DIX: + msg->eth = eth_config_s_to_msg(&s->eth); + if (msg->eth == NULL) + goto fail_msg; + break; + case IPCP_UDP: + msg->udp = udp_config_s_to_msg(&s->udp); + if (msg->udp == NULL) + goto fail_msg; + break; + default: + /* No checks here */ + break; + } + + msg->ipcp_type = s->type; + + msg->layer_info = layer_info_s_to_msg(&s->layer_info); + if (msg->layer_info == NULL) + goto fail_msg; + + return msg; + + fail_msg: + ipcp_config_msg__free_unpacked(msg, NULL); + fail_malloc: + return NULL; +} + +struct ipcp_config ipcp_config_msg_to_s(const ipcp_config_msg_t * msg) +{ + struct ipcp_config s; + + assert(msg != NULL); + + s.type = msg->ipcp_type; + + s.layer_info = layer_info_msg_to_s(msg->layer_info); + + switch(msg->ipcp_type) { + case IPCP_LOCAL: + break; + case IPCP_UNICAST: + s.unicast = uni_config_msg_to_s(msg->unicast); + break; + case IPCP_ETH_LLC: + /* FALLTHRU */ + case IPCP_ETH_DIX: + s.eth = eth_config_msg_to_s(msg->eth); + break; + case IPCP_UDP: + s.udp = udp_config_msg_to_s(msg->udp); + break; + case IPCP_BROADCAST: + break; + default: + /* No checks here */ + break; + } + + return s; +} + +qosspec_msg_t * qos_spec_s_to_msg(const struct qos_spec * s) +{ + qosspec_msg_t * msg; + + assert(s != NULL); + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + return NULL; + + qosspec_msg__init(msg); + + msg->delay = s->delay; + msg->bandwidth = s->bandwidth; + msg->availability = s->availability; + msg->loss = s->loss; + msg->ber = s->ber; + msg->in_order = s->in_order; + msg->max_gap = s->max_gap; + msg->cypher_s = s->cypher_s; + msg->timeout = s->timeout; + + return msg; +} + +struct qos_spec qos_spec_msg_to_s(const qosspec_msg_t * msg) +{ + struct qos_spec s; + + assert(msg != NULL); + + s.delay = msg->delay; + s.bandwidth = msg->bandwidth; + s.availability = msg->availability; + s.loss = msg->loss; + s.ber = msg->ber; + s.in_order = msg->in_order; + s.max_gap = msg->max_gap; + s.cypher_s = msg->cypher_s; + s.timeout = msg->timeout; + + return s; +} + +enroll_req_msg_t * enroll_req_s_to_msg(const struct enroll_req * s) +{ + enroll_req_msg_t * msg; + uint8_t * id; + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_msg; + + id = malloc(ENROLL_ID_LEN); + if (id == NULL) + goto fail_id; + + memcpy(id, s->id, ENROLL_ID_LEN); + + enroll_req_msg__init(msg); + + msg->id.len = ENROLL_ID_LEN; + msg->id.data = id; + + return msg; + + fail_id: + free(msg); + fail_msg: + return NULL; +} + +struct enroll_req enroll_req_msg_to_s(const enroll_req_msg_t * msg) +{ + struct enroll_req s; + + assert(msg != NULL); + + memcpy(s.id, msg->id.data, ENROLL_ID_LEN); + + return s; +} + +enroll_resp_msg_t * enroll_resp_s_to_msg(const struct enroll_resp * s) +{ + enroll_resp_msg_t * msg; + uint8_t * id; + + assert(s != NULL); + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_msg; + + id = malloc(ENROLL_ID_LEN); + if (id == NULL) + goto fail_id; + + memcpy(id, s->id, ENROLL_ID_LEN); + + enroll_resp_msg__init(msg); + + msg->id.len = ENROLL_ID_LEN; + msg->id.data = id; + + msg->t_sec = s->t.tv_sec; + msg->t_nsec = s->t.tv_nsec; + msg->response = s->response; + if (msg->response < 0) + return msg; + + msg->conf = ipcp_config_s_to_msg(&s->conf); + if (msg->conf == NULL) + goto fail_id; + + return msg; + + fail_id: + enroll_resp_msg__free_unpacked(msg, NULL); + fail_msg: + return NULL; +} + +struct enroll_resp enroll_resp_msg_to_s(const enroll_resp_msg_t * msg) +{ + struct enroll_resp s; + + assert (msg != NULL); + + s.response = msg->response; + if (s.response >= 0) + s.conf = ipcp_config_msg_to_s(msg->conf); + + s.t.tv_sec = msg->t_sec; + s.t.tv_nsec = msg->t_nsec; + + memcpy(s.id, msg->id.data, ENROLL_ID_LEN); + + return s; +} + +enroll_ack_msg_t * enroll_ack_s_to_msg(const struct enroll_ack * s) +{ + enroll_ack_msg_t * msg; + uint8_t * id; + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_msg; + + id = malloc(ENROLL_ID_LEN); + if (id == NULL) + goto fail_id; + + memcpy(id, s->id, ENROLL_ID_LEN); + + enroll_ack_msg__init(msg); + + msg->id.len = ENROLL_ID_LEN; + msg->id.data = id; + + msg->result = s->result; + + return msg; + + fail_id: + enroll_ack_msg__free_unpacked(msg, NULL); + fail_msg: + return NULL; +} + +struct enroll_ack enroll_ack_msg_to_s(const enroll_ack_msg_t * msg) +{ + struct enroll_ack s; + + assert(msg != NULL); + + memcpy(s.id, msg->id.data, ENROLL_ID_LEN); + + s.result = msg->result; + + return s; +} diff --git a/src/lib/qoscube.c b/src/lib/qoscube.c index 3e6f884d..267b3a87 100644 --- a/src/lib/qoscube.c +++ b/src/lib/qoscube.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Quality of Service cube * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License diff --git a/src/lib/qosspec.proto b/src/lib/qosspec.proto deleted file mode 100644 index 5d8abb79..00000000 --- a/src/lib/qosspec.proto +++ /dev/null @@ -1,34 +0,0 @@ -/* - * Ouroboros - Copyright (C) 2016 - 2020 - * - * QoS specification message - * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> - * - * 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/. - */ - -syntax = "proto2"; - -message qosspec_msg { - required uint32 delay = 1; /* In ms */ - required uint64 bandwidth = 2; /* In bits/s */ - required uint32 availability = 3; /* Class of 9s */ - required uint32 loss = 4; /* Packet loss */ - required uint32 ber = 5; /* Bit error rate, ppb */ - required uint32 in_order = 6; /* In-order delivery */ - required uint32 max_gap = 7; /* In ms */ - required uint32 cypher_s = 8; /* Crypto strength in bits */ -}; diff --git a/src/lib/random.c b/src/lib/random.c index b2cdabbc..09e0b844 100644 --- a/src/lib/random.c +++ b/src/lib/random.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Pseudo random generator * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License diff --git a/src/lib/rib.c b/src/lib/rib.c index 684c5dcd..97a20f47 100644 --- a/src/lib/rib.c +++ b/src/lib/rib.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * RIB export using FUSE * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License @@ -43,15 +43,14 @@ #define __USE_XOPEN #elif defined (__FreeBSD__) #define __XSI_VISIBLE 500 -#endif +#endif /* __linux__ */ +#include <sys/stat.h> #include <fuse.h> #ifndef CLOCK_REALTIME_COARSE #define CLOCK_REALTIME_COARSE CLOCK_REALTIME #endif -#define RT "/" - struct reg_comp { struct list_head next; @@ -106,8 +105,7 @@ static int rib_read(const char * path, strcpy(comp, path + 1); - c = strstr(comp, "/"); - + c = strstr(comp, RIB_SEPARATOR); if (c != NULL) *c = '\0'; @@ -119,7 +117,7 @@ static int rib_read(const char * path, list_for_each(p, &rib.reg_comps) { struct reg_comp * r = list_entry(p, struct reg_comp, next); if (strcmp(comp, r->path) == 0) { - int ret = r->ops->read(c + 1, buf, size); + int ret = r->ops->read(path + 1, buf, size); pthread_rwlock_unlock(&rib.lock); return ret; } @@ -141,12 +139,16 @@ static int rib_readdir(const char * path, (void) offset; (void) info; + /* Fix ls calling readdir in an infinite loop on raspbian. */ + if (info != NULL && info->nonseekable != 0) + return -ENOENT; + filler(buf, ".", NULL, 0); filler(buf, "..", NULL, 0); pthread_rwlock_rdlock(&rib.lock); - if (strcmp(path, RT) == 0) { + if (strcmp(path, RIB_SEPARATOR) == 0) { list_for_each(p, &rib.reg_comps) { struct reg_comp * c; c = list_entry(p, struct reg_comp, next); @@ -185,24 +187,35 @@ static size_t __getattr(const char * path, struct list_head * p; char comp[RIB_PATH_LEN + 1]; char * c; + struct rib_attr attr; if (strlen(path) > RIB_PATH_LEN) return -1; strcpy(comp, path + 1); - c = strstr(comp, "/"); - + c = strstr(comp, RIB_SEPARATOR); if (c != NULL) *c = '\0'; + memset(&attr, 0, sizeof(attr)); + pthread_rwlock_rdlock(&rib.lock); list_for_each(p, &rib.reg_comps) { struct reg_comp * r = list_entry(p, struct reg_comp, next); if (strcmp(comp, r->path) == 0) { - size_t ret = r->ops->getattr(c + 1, st); + size_t ret = r->ops->getattr(path + 1, &attr); pthread_rwlock_unlock(&rib.lock); + st->st_mode = S_IFREG | 0644; + st->st_blocks = 1; + st->st_nlink = 1; + st->st_uid = getuid(); + st->st_gid = getgid(); + st->st_size = attr.size; + st->st_atime = attr.mtime; + st->st_mtime = attr.mtime; + st->st_ctime = attr.mtime; return ret; } } @@ -220,7 +233,7 @@ static int rib_getattr(const char * path, memset(st, 0, sizeof(*st)); - if (strcmp(path, RT) == 0) + if (strcmp(path, RIB_SEPARATOR) == 0) goto finish_dir; pthread_rwlock_rdlock(&rib.lock); @@ -246,6 +259,8 @@ static int rib_getattr(const char * path, st->st_uid = getuid(); st->st_gid = getgid(); st->st_mtime = now.tv_sec; + st->st_atime = now.tv_sec; + st->st_ctime = now.tv_sec; return 0; } @@ -266,7 +281,6 @@ static void * fuse_thr(void * o) } #endif /* HAVE_FUSE */ - int rib_init(const char * mountpt) { #ifdef HAVE_FUSE @@ -280,8 +294,11 @@ int rib_init(const char * mountpt) NULL}; struct fuse_args args = FUSE_ARGS_INIT(3, argv); + if (access("/dev/fuse", R_OK)) + goto fail; + if (stat(FUSE_PREFIX, &st) == -1) - return -1; + goto fail; sprintf(rib.mnt, FUSE_PREFIX "/%s", mountpt); @@ -289,13 +306,13 @@ int rib_init(const char * mountpt) switch(errno) { case ENOENT: if (mkdir(rib.mnt, 0777)) - return -1; + goto fail_mnt; break; case ENOTCONN: fuse_unmount(rib.mnt, rib.ch); break; default: - return -1; + goto fail_mnt; } fuse_opt_parse(&args, NULL, NULL, NULL); @@ -329,6 +346,9 @@ int rib_init(const char * mountpt) fail_mount: fuse_opt_free_args(&args); rmdir(rib.mnt); + fail_mnt: + memset(rib.mnt, 0, sizeof(rib.mnt)); + fail: return -1; #else (void) mountpt; @@ -342,6 +362,9 @@ void rib_fini(void) struct list_head * p; struct list_head * h; + if (strlen(rib.mnt) == 0) + return; + fuse_exit(rib.fuse); fuse_unmount(rib.mnt, rib.ch); @@ -363,6 +386,18 @@ void rib_fini(void) pthread_rwlock_unlock(&rib.lock); pthread_rwlock_destroy(&rib.lock); + + memset(rib.mnt, 0, sizeof(rib.mnt)); +#endif +} + +void rib_cleanup(const char * mnt) +{ +#ifdef HAVE_FUSE + fuse_unmount(mnt, NULL); + rmdir(mnt); +#else + (void) mnt; #endif } @@ -373,6 +408,9 @@ int rib_reg(const char * path, struct reg_comp * rc; struct list_head * p; + if (strlen(rib.mnt) == 0) + return 0; + pthread_rwlock_wrlock(&rib.lock); list_for_each(p, &rib.reg_comps) { @@ -417,6 +455,9 @@ void rib_unreg(const char * path) struct list_head * p; struct list_head * h; + if (strlen(rib.mnt) == 0) + return; + pthread_rwlock_wrlock(&rib.lock); list_for_each_safe(p, h, &rib.reg_comps) { diff --git a/src/lib/rxmwheel.c b/src/lib/rxmwheel.c deleted file mode 100644 index ce7ef8e4..00000000 --- a/src/lib/rxmwheel.c +++ /dev/null @@ -1,271 +0,0 @@ -/* - * Ouroboros - Copyright (C) 2016 - 2020 - * - * Timerwheel - * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> - * - * 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/. - */ - -#include <ouroboros/list.h> - -#define RXMQ_S 16 /* defines #slots */ -#define RXMQ_M 24 /* defines max delay */ -#define RXMQ_R (RXMQ_M - RXMQ_S) /* defines resolution */ -#define RXMQ_SLOTS (1 << RXMQ_S) -#define RXMQ_MAX (1 << RXMQ_M) /* ms */ - -/* Small inacurracy to avoid slow division by MILLION. */ -#define ts_to_ms(ts) (ts.tv_sec * 1000 + (ts.tv_nsec >> 20)) -#define ts_to_us(ts) (ts.tv_sec * MILLION + (ts.tv_nsec >> 10)) -#define ts_to_slot(ts) ((ts_to_us(ts) >> RXMQ_R) & (RXMQ_SLOTS - 1)) - -struct rxm { - struct list_head next; - uint32_t seqno; - struct shm_du_buff * sdb; - uint8_t * head; - uint8_t * tail; - time_t t0; /* Time when original was sent (s). */ - size_t mul; /* RTO multiplier. */ - struct frcti * frcti; -}; - -struct { - struct list_head wheel[RXMQ_SLOTS]; - - size_t prv; /* Last processed slot. */ - pthread_mutex_t lock; -} rw; - -static void rxmwheel_fini(void) -{ - size_t i; - struct list_head * p; - struct list_head * h; - - for (i = 0; i < RXMQ_SLOTS; ++i) { - list_for_each_safe(p, h, &rw.wheel[i]) { - struct rxm * rxm = list_entry(p, struct rxm, next); - list_del(&rxm->next); - free(rxm); - } - } -} - -static int rxmwheel_init(void) -{ - struct timespec now; - size_t i; - - if (pthread_mutex_init(&rw.lock, NULL)) - return -1; - - clock_gettime(PTHREAD_COND_CLOCK, &now); - - /* Mark the previous timeslot as the last one processed. */ - rw.prv = (ts_to_slot(now) - 1) & (RXMQ_SLOTS - 1); - - for (i = 0; i < RXMQ_SLOTS; ++i) - list_head_init(&rw.wheel[i]); - - return 0; -} - -static void rxmwheel_clear(int fd) -{ - size_t i; - - /* FIXME: Add list element to avoid looping over full rxmwheel. */ - pthread_mutex_lock(&rw.lock); - - for (i = 0; i < RXMQ_SLOTS; ++i) { - struct list_head * p; - struct list_head * h; - - list_for_each_safe(p, h, &rw.wheel[i]) { - struct rxm * r = list_entry(p, struct rxm, next); - if (r->frcti->fd == fd) { - list_del(&r->next); - shm_du_buff_ack(r->sdb); - ipcp_sdb_release(r->sdb); - free(r); - } - } - } - - pthread_mutex_unlock(&rw.lock); -} - -static void check_probe(struct frcti * frcti, - uint32_t seqno) -{ - /* disable rtt probe if this packet */ - - /* TODO: This should be locked, but lock reversal! */ - - if (frcti->probe && ((frcti->rttseq + 1) == seqno)) { - /* Backoff to avoid never updating rtt */ - frcti->srtt_us <<= 1; - frcti->probe = false; - } -} - -#define rto(frcti) (frcti->srtt_us + (frcti->mdev_us << 1)) -/* Return fd on r-timer expiry. */ -static int rxmwheel_move(void) -{ - struct timespec now; - struct list_head * p; - struct list_head * h; - size_t slot; - size_t i; - - clock_gettime(PTHREAD_COND_CLOCK, &now); - - slot = ts_to_slot(now); - - pthread_mutex_lock(&rw.lock); - - for (i = rw.prv; (ssize_t) (i - slot) <= 0; ++i) { - list_for_each_safe(p, h, &rw.wheel[i]) { - struct rxm * r; - struct frct_cr * snd_cr; - struct frct_cr * rcv_cr; - size_t rslot; - time_t newtime; - ssize_t idx; - struct shm_du_buff * sdb; - uint8_t * head; - struct flow * f; - - r = list_entry(p, struct rxm, next); - list_del(&r->next); - - snd_cr = &r->frcti->snd_cr; - rcv_cr = &r->frcti->rcv_cr; - /* Has been ack'd, remove. */ - if ((int) (r->seqno - snd_cr->lwe) <= 0) { - shm_du_buff_ack(r->sdb); - ipcp_sdb_release(r->sdb); - free(r); - continue; - } - - /* Disable using this seqno as rto probe. */ - check_probe(r->frcti, r->seqno); - - /* Check for r-timer expiry. */ - if (ts_to_ms(now) - r->t0 > r->frcti->r) { - int fd = r->frcti->fd; - pthread_mutex_unlock(&rw.lock); - shm_du_buff_ack(r->sdb); - ipcp_sdb_release(r->sdb); - free(r); - return fd; - } - - /* Copy the payload, safe rtx in other layers. */ - if (ipcp_sdb_reserve(&sdb, r->tail - r->head)) { - /* FIXME: reschedule send? */ - int fd = r->frcti->fd; - pthread_mutex_unlock(&rw.lock); - shm_du_buff_ack(r->sdb); - ipcp_sdb_release(r->sdb); - free(r); - return fd; - } - - idx = shm_du_buff_get_idx(sdb); - - head = shm_du_buff_head(sdb); - memcpy(head, r->head, r->tail - r->head); - - /* Release the old copy. */ - shm_du_buff_ack(r->sdb); - ipcp_sdb_release(r->sdb); - - /* Update ackno and make sure DRF is not set. */ - ((struct frct_pci *) head)->ackno = ntoh32(rcv_cr->lwe); - ((struct frct_pci *) head)->flags &= ~FRCT_DRF; - - f = &ai.flows[r->frcti->fd]; - - /* Retransmit the copy. */ - if (shm_rbuff_write(f->tx_rb, idx)) { - ipcp_sdb_release(sdb); - free(r); - /* FIXME: reschedule send? */ - continue; - } - - shm_flow_set_notify(f->set, f->flow_id, FLOW_PKT); - - /* Reschedule. */ - shm_du_buff_wait_ack(sdb); - - r->head = head; - r->tail = shm_du_buff_tail(sdb); - r->sdb = sdb; - - newtime = ts_to_us(now) + rto(f->frcti); - rslot = (newtime >> RXMQ_R) & (RXMQ_SLOTS - 1); - - list_add_tail(&r->next, &rw.wheel[rslot]); - } - } - - rw.prv = slot; - - pthread_mutex_unlock(&rw.lock); - - return 0; -} - -static int rxmwheel_add(struct frcti * frcti, - uint32_t seqno, - struct shm_du_buff * sdb) -{ - struct timespec now; - struct rxm * r; - size_t slot; - - r = malloc(sizeof(*r)); - if (r == NULL) - return -ENOMEM; - - clock_gettime(PTHREAD_COND_CLOCK, &now); - - pthread_mutex_lock(&rw.lock); - - r->t0 = ts_to_us(now); - r->mul = 0; - r->seqno = seqno; - r->sdb = sdb; - r->head = shm_du_buff_head(sdb); - r->tail = shm_du_buff_tail(sdb); - r->frcti = frcti; - - slot = ((r->t0 + rto(frcti)) >> RXMQ_R) & (RXMQ_SLOTS - 1); - - list_add_tail(&r->next, &rw.wheel[slot]); - - pthread_mutex_unlock(&rw.lock); - - shm_du_buff_wait_ack(sdb); - - return 0; -} diff --git a/src/lib/serdes-irm.c b/src/lib/serdes-irm.c new file mode 100644 index 00000000..c4ba3053 --- /dev/null +++ b/src/lib/serdes-irm.c @@ -0,0 +1,478 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2024 + * + * Ouroboros IRM Protocol - serialization/deserialization + * + * 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/. + */ + +#define _POSIX_C_SOURCE 200809L + +#include "config.h" + +#include <ouroboros/errno.h> +#include <ouroboros/serdes-irm.h> +#include <ouroboros/protobuf.h> + +#include <stdlib.h> +#include <string.h> + +int flow_accept__irm_req_ser(buffer_t * buf, + const struct flow_info * flow, + const struct timespec * timeo) +{ + irm_msg_t * msg; + size_t len; + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_malloc; + + irm_msg__init(msg); + + msg->code = IRM_MSG_CODE__IRM_FLOW_ACCEPT; + msg->flow_info = flow_info_s_to_msg(flow); + if (msg->flow_info == NULL) + goto fail_msg; + + msg->timeo = timeo == NULL ? NULL : timespec_s_to_msg(timeo); + if (timeo != NULL && msg->timeo == NULL) + goto fail_msg; + + len = irm_msg__get_packed_size(msg); + if (len == 0 || len > buf->len) + goto fail_msg; + + buf->len = len; + + irm_msg__pack(msg, buf->data); + irm_msg__free_unpacked(msg, NULL); + + return 0; + + fail_msg: + irm_msg__free_unpacked(msg, NULL); + fail_malloc: + return -ENOMEM; +} + +static int __flow_alloc_ser(buffer_t * buf, + const struct flow_info * flow, + const char * dst, + const struct timespec * timeo, + int msg_code) +{ + irm_msg_t * msg; + size_t len; + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_malloc; + + irm_msg__init(msg); + + msg->code = msg_code; + msg->flow_info = flow_info_s_to_msg(flow); + if (msg->flow_info == NULL) + goto fail_msg; + + msg->dst = strdup(dst); + if (msg->dst == NULL) + goto fail_msg; + + msg->timeo = timeo == NULL ? NULL : timespec_s_to_msg(timeo); + if (timeo != NULL && msg->timeo == NULL) + goto fail_msg; + + len = irm_msg__get_packed_size(msg); + if (len == 0 || len > buf->len) + goto fail_msg; + + buf->len = len; + + irm_msg__pack(msg, buf->data); + irm_msg__free_unpacked(msg, NULL); + + return 0; + + fail_msg: + irm_msg__free_unpacked(msg, NULL); + fail_malloc: + return -ENOMEM; +} + +int flow_alloc__irm_req_ser(buffer_t * buf, + const struct flow_info * flow, + const char * dst, + const struct timespec * timeo) +{ + return __flow_alloc_ser(buf, flow, dst, timeo, + IRM_MSG_CODE__IRM_FLOW_ALLOC); +} + +int flow_join__irm_req_ser(buffer_t * buf, + const struct flow_info * flow, + const char * dst, + const struct timespec * timeo) +{ + return __flow_alloc_ser(buf, flow, dst, timeo, + IRM_MSG_CODE__IRM_FLOW_JOIN); +} + +int flow__irm_result_des(buffer_t * buf, + struct flow_info * flow, + buffer_t * sk) +{ + irm_msg_t * msg; + int err; + + if (sk != NULL) + sk->data = NULL; + + msg = irm_msg__unpack(NULL, buf->len, buf->data); + if (msg == NULL) { + err = -EIRMD; + goto fail_msg; + } + + if (!msg->has_result) { + err = -EIRMD; + goto fail; + } + + if (msg->result < 0) { + err = msg->result; + goto fail; + } + + if (msg->flow_info == NULL) { + err = -EBADF; + goto fail; + } + + *flow = flow_info_msg_to_s(msg->flow_info); + + if (flow->qs.cypher_s > 0 && sk != NULL) { + if (msg->symmkey.data == NULL || msg->symmkey.len == 0) { + err = -ECRYPT; + goto fail; + } + + sk->len = msg->symmkey.len; + sk->data = msg->symmkey.data; + + msg->symmkey.data = NULL; + msg->symmkey.len = 0; + } + + irm_msg__free_unpacked(msg, NULL); + + return 0; + fail: + irm_msg__free_unpacked(msg, NULL); + fail_msg: + return err; +} + +int flow_dealloc__irm_req_ser(buffer_t * buf, + const struct flow_info * flow, + const struct timespec * timeo) +{ + irm_msg_t * msg; + size_t len; + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_malloc; + + irm_msg__init(msg); + + msg->code = IRM_MSG_CODE__IRM_FLOW_DEALLOC; + msg->flow_info = flow_info_s_to_msg(flow); + if (msg->flow_info == NULL) + goto fail_msg; + + msg->timeo = timespec_s_to_msg(timeo); + if (msg->timeo == NULL) + goto fail_msg; + + len = irm_msg__get_packed_size(msg); + if (len == 0 || len > buf->len) + goto fail_msg; + + buf->len = len; + + irm_msg__pack(msg, buf->data); + irm_msg__free_unpacked(msg, NULL); + + return 0; + + fail_msg: + irm_msg__free_unpacked(msg, NULL); + fail_malloc: + return -ENOMEM; +} + +int ipcp_flow_dealloc__irm_req_ser(buffer_t * buf, + const struct flow_info * flow) +{ + irm_msg_t * msg; + size_t len; + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_malloc; + + irm_msg__init(msg); + + msg->code = IRM_MSG_CODE__IPCP_FLOW_DEALLOC; + msg->flow_info = flow_info_s_to_msg(flow); + if (msg->flow_info == NULL) + goto fail_msg; + + len = irm_msg__get_packed_size(msg); + if (len == 0 || len > buf->len) + goto fail_msg; + + buf->len = len; + + irm_msg__pack(msg, buf->data); + irm_msg__free_unpacked(msg, NULL); + + return 0; + fail_msg: + irm_msg__free_unpacked(msg, NULL); + fail_malloc: + return -ENOMEM; +} + + +int ipcp_create_r__irm_req_ser(buffer_t * buf, + const struct ipcp_info * ipcp) +{ + irm_msg_t * msg; + size_t len; + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_malloc; + + irm_msg__init(msg); + + msg->code = IRM_MSG_CODE__IPCP_CREATE_R; + msg->ipcp_info = ipcp_info_s_to_msg(ipcp); + if (msg->ipcp_info == NULL) + goto fail_msg; + + len = irm_msg__get_packed_size(msg); + if (len == 0 || len > buf->len) + goto fail_msg; + + buf->len = len; + + irm_msg__pack(msg, buf->data); + irm_msg__free_unpacked(msg, NULL); + + return 0; + fail_msg: + irm_msg__free_unpacked(msg, NULL); + fail_malloc: + return -ENOMEM; +} + +int proc_announce__irm_req_ser(buffer_t * buf, + const char * prog) +{ + irm_msg_t * msg; + size_t len; + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_malloc; + + irm_msg__init(msg); + + msg->code = IRM_MSG_CODE__IRM_PROC_ANNOUNCE; + msg->has_pid = true; + msg->pid = getpid(); + msg->prog = strdup(prog); + if (msg->prog == NULL) + goto fail_msg; + + len = irm_msg__get_packed_size(msg); + if (len == 0 || len > buf->len) + goto fail_msg; + + buf->len = len; + + irm_msg__pack(msg, buf->data); + irm_msg__free_unpacked(msg, NULL); + + return 0; + fail_msg: + irm_msg__free_unpacked(msg, NULL); + fail_malloc: + return -ENOMEM; +} + +int proc_exit__irm_req_ser(buffer_t * buf) +{ + irm_msg_t * msg; + size_t len; + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_malloc; + + irm_msg__init(msg); + + msg->code = IRM_MSG_CODE__IRM_PROC_EXIT; + msg->has_pid = true; + msg->pid = getpid(); + + len = irm_msg__get_packed_size(msg); + if (len == 0 || len > buf->len) + goto fail_msg; + + buf->len = len; + + irm_msg__pack(msg, buf->data); + irm_msg__free_unpacked(msg, NULL); + + return 0; + fail_msg: + irm_msg__free_unpacked(msg, NULL); + fail_malloc: + return -ENOMEM; +} + +int ipcp_flow_req_arr__irm_req_ser(buffer_t * buf, + const buffer_t * dst, + const struct flow_info * flow, + const buffer_t * data) +{ + irm_msg_t * msg; + size_t len; + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_malloc; + + irm_msg__init(msg); + + msg->code = IRM_MSG_CODE__IPCP_FLOW_REQ_ARR; + msg->flow_info = flow_info_s_to_msg(flow); + if (msg->flow_info == NULL) + goto fail_msg; + + msg->has_hash = true; + msg->hash.len = dst->len; + msg->hash.data = dst->data; + msg->has_pk = true; + msg->pk.len = data->len; + msg->pk.data = data->data; + + len = irm_msg__get_packed_size(msg); + if (len == 0 || len > buf->len) + goto fail_msg; + + buf->len = len; + + irm_msg__pack(msg, buf->data); + + /* Don't free * dst or data! */ + msg->hash.len = 0; + msg->hash.data = NULL; + msg->pk.len = 0; + msg->pk.data = NULL; + irm_msg__free_unpacked(msg, NULL); + + return 0; + fail_msg: + irm_msg__free_unpacked(msg, NULL); + fail_malloc: + return -ENOMEM; +} + +int ipcp_flow_alloc_reply__irm_msg_ser(buffer_t * buf, + const struct flow_info * flow, + int response, + const buffer_t * data) +{ + irm_msg_t * msg; + size_t len; + + msg = malloc(sizeof(*msg)); + if (msg == NULL) + goto fail_malloc; + + irm_msg__init(msg); + + msg->code = IRM_MSG_CODE__IPCP_FLOW_ALLOC_REPLY; + msg->flow_info = flow_info_s_to_msg(flow); + if (msg->flow_info == NULL) + goto fail_msg; + + msg->has_pk = true; + msg->pk.len = data->len; + msg->pk.data = data->data; + msg->has_response = true; + msg->response = response; + + len = irm_msg__get_packed_size(msg); + if (len == 0 || len > buf->len) + goto fail_msg; + + buf->len = len; + + irm_msg__pack(msg, buf->data); + + /* Don't free * data! */ + msg->pk.len = 0; + msg->pk.data = NULL; + + irm_msg__free_unpacked(msg, NULL); + + return 0; + fail_msg: + irm_msg__free_unpacked(msg, NULL); + fail_malloc: + return -ENOMEM; +} + +int irm__irm_result_des(buffer_t * buf) +{ + irm_msg_t * msg; + int err; + + msg = irm_msg__unpack(NULL, buf->len, buf->data); + if (msg == NULL) { + err = -EIRMD; + goto fail_msg; + } + + if (!msg->has_result) { + err = -EIRMD; + goto fail; + } + + err = msg->result; + fail: + irm_msg__free_unpacked(msg, NULL); + fail_msg: + return err; +} diff --git a/src/lib/serdes-oep.c b/src/lib/serdes-oep.c new file mode 100644 index 00000000..8a836b3b --- /dev/null +++ b/src/lib/serdes-oep.c @@ -0,0 +1,161 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2024 + * + * Ouroboros Enrollment Protocol - serialization/deserialization + * + * 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/. + */ + +#define _POSIX_C_SOURCE 200112L + +#include <ouroboros/protobuf.h> +#include <ouroboros/serdes-oep.h> + +ssize_t enroll_req_ser(const struct enroll_req * req, + buffer_t buf) +{ + enroll_req_msg_t * msg; + ssize_t sz; + + msg = enroll_req_s_to_msg(req); + if (msg == NULL) + goto fail_msg; + + sz = enroll_req_msg__get_packed_size(msg); + if (sz < 0 || (size_t) sz > buf.len) + goto fail_pack; + + enroll_req_msg__pack(msg, buf.data); + + enroll_req_msg__free_unpacked(msg, NULL); + + return sz; + + fail_pack: + enroll_req_msg__free_unpacked(msg, NULL); + fail_msg: + return -1; +} + +int enroll_req_des(struct enroll_req * req, + const buffer_t buf) +{ + enroll_req_msg_t * msg; + + msg = enroll_req_msg__unpack(NULL, buf.len, buf.data); + if (msg == NULL) + goto fail_unpack; + + if (msg->id.len != ENROLL_ID_LEN) + goto fail_id; + + *req = enroll_req_msg_to_s(msg); + + enroll_req_msg__free_unpacked(msg, NULL); + + return 0; + + fail_id: + enroll_req_msg__free_unpacked(msg, NULL); + fail_unpack: + return -1; +} + +ssize_t enroll_resp_ser(const struct enroll_resp * resp, + buffer_t buf) +{ + enroll_resp_msg_t * msg; + ssize_t sz; + + msg = enroll_resp_s_to_msg(resp); + if (msg == NULL) + goto fail_msg; + + sz = enroll_resp_msg__get_packed_size(msg); + if (sz < 0 || (size_t) sz > buf.len) + goto fail_pack; + + enroll_resp_msg__pack(msg, buf.data); + + enroll_resp_msg__free_unpacked(msg, NULL); + + return sz; + + fail_pack: + enroll_resp_msg__free_unpacked(msg, NULL); + fail_msg: + return -1; +} + +int enroll_resp_des(struct enroll_resp * resp, + const buffer_t buf) +{ + enroll_resp_msg_t * msg; + + msg = enroll_resp_msg__unpack(NULL, buf.len, buf.data); + if (msg == NULL) + return -1; + + *resp = enroll_resp_msg_to_s(msg); + + enroll_resp_msg__free_unpacked(msg, NULL); + + return 0; +} + +ssize_t enroll_ack_ser(const struct enroll_ack * ack, + buffer_t buf) +{ + enroll_ack_msg_t * msg; + ssize_t sz; + + msg = enroll_ack_s_to_msg(ack); + if (msg == NULL) + goto fail_msg; + + sz = enroll_ack_msg__get_packed_size(msg); + if (sz < 0 || (size_t) sz > buf.len) + goto fail_pack; + + enroll_ack_msg__pack(msg, buf.data); + + enroll_ack_msg__free_unpacked(msg, NULL); + + return sz; + + fail_pack: + enroll_ack_msg__free_unpacked(msg, NULL); + fail_msg: + return -1; + +} + +int enroll_ack_des(struct enroll_ack * ack, + const buffer_t buf) +{ + enroll_ack_msg_t * msg; + + msg = enroll_ack_msg__unpack(NULL, buf.len, buf.data); + if (msg == NULL) + return -1; + + *ack = enroll_ack_msg_to_s(msg); + + enroll_ack_msg__free_unpacked(msg, NULL); + + return 0; +} diff --git a/src/lib/sha3.c b/src/lib/sha3.c index 7bdafcb5..b9d6b07f 100644 --- a/src/lib/sha3.c +++ b/src/lib/sha3.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * SHA3 algorithm * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This implementation is adapted and redistributed from the RHASH * project @@ -52,8 +52,7 @@ #include <assert.h> #include <string.h> -#define IS_ALIGNED_64(p) (0 == (7 & ((const uint8_t *) (p) \ - - (const uint8_t *) 0))) +#define IS_ALIGNED_64(p) (0 == (7 & ((uintptr_t) (p)))) #define I64(x) x##LL #define ROTL64(qword, n) ((qword) << (n) ^ ((qword) >> (64 - (n)))) diff --git a/src/lib/shm_flow_set.c b/src/lib/shm_flow_set.c index 0f701b63..39913fd1 100644 --- a/src/lib/shm_flow_set.c +++ b/src/lib/shm_flow_set.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Management of flow_sets for fqueue * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License @@ -24,21 +24,21 @@ #include "config.h" +#include <ouroboros/errno.h> #include <ouroboros/lockfile.h> -#include <ouroboros/time_utils.h> +#include <ouroboros/pthread.h> #include <ouroboros/shm_flow_set.h> -#include <ouroboros/errno.h> +#include <ouroboros/time.h> -#include <pthread.h> -#include <sys/mman.h> +#include <assert.h> #include <fcntl.h> +#include <signal.h> #include <stdlib.h> #include <stdio.h> +#include <string.h> #include <unistd.h> -#include <signal.h> +#include <sys/mman.h> #include <sys/stat.h> -#include <string.h> -#include <assert.h> /* * pthread_cond_timedwait has a WONTFIX bug as of glibc 2.25 where it @@ -52,39 +52,35 @@ #endif #define FN_MAX_CHARS 255 +#define FS_PROT (PROT_READ | PROT_WRITE) -#define QUEUESIZE ((SHM_BUFFER_SIZE) * sizeof(struct portevent)) +#define QUEUESIZE ((SHM_BUFFER_SIZE) * sizeof(struct flowevent)) -#define SHM_FLOW_SET_FILE_SIZE (SYS_MAX_FLOWS * sizeof(ssize_t) \ - + PROG_MAX_FQUEUES * sizeof(size_t) \ - + PROG_MAX_FQUEUES * sizeof(pthread_cond_t) \ - + PROG_MAX_FQUEUES * QUEUESIZE \ - + sizeof(pthread_mutex_t)) +#define SHM_FSET_FILE_SIZE (SYS_MAX_FLOWS * sizeof(ssize_t) \ + + PROG_MAX_FQUEUES * sizeof(size_t) \ + + PROG_MAX_FQUEUES * sizeof(pthread_cond_t) \ + + PROG_MAX_FQUEUES * QUEUESIZE \ + + sizeof(pthread_mutex_t)) #define fqueue_ptr(fs, idx) (fs->fqueues + (SHM_BUFFER_SIZE) * idx) -struct portevent { - int flow_id; - int event; -}; - struct shm_flow_set { ssize_t * mtable; size_t * heads; pthread_cond_t * conds; - struct portevent * fqueues; + struct flowevent * fqueues; pthread_mutex_t * lock; pid_t pid; }; static struct shm_flow_set * flow_set_create(pid_t pid, - int flags) + int oflags) { struct shm_flow_set * set; ssize_t * shm_base; char fn[FN_MAX_CHARS]; - int shm_fd; + int fd; sprintf(fn, SHM_FLOW_SET_PREFIX "%d", pid); @@ -92,39 +88,33 @@ static struct shm_flow_set * flow_set_create(pid_t pid, if (set == NULL) goto fail_malloc; - shm_fd = shm_open(fn, flags, 0666); - if (shm_fd == -1) + fd = shm_open(fn, oflags, 0666); + if (fd == -1) goto fail_shm_open; - if (ftruncate(shm_fd, SHM_FLOW_SET_FILE_SIZE - 1) < 0) { - close(shm_fd); - goto fail_shm_open; - } - - shm_base = mmap(NULL, - SHM_FLOW_SET_FILE_SIZE, - PROT_READ | PROT_WRITE, - MAP_SHARED, - shm_fd, - 0); - - close(shm_fd); + if ((oflags & O_CREAT) && ftruncate(fd, SHM_FSET_FILE_SIZE) < 0) + goto fail_truncate; + shm_base = mmap(NULL, SHM_FSET_FILE_SIZE, FS_PROT, MAP_SHARED, fd, 0); if (shm_base == MAP_FAILED) goto fail_mmap; + close(fd); + set->mtable = shm_base; set->heads = (size_t *) (set->mtable + SYS_MAX_FLOWS); set->conds = (pthread_cond_t *)(set->heads + PROG_MAX_FQUEUES); - set->fqueues = (struct portevent *) (set->conds + PROG_MAX_FQUEUES); + set->fqueues = (struct flowevent *) (set->conds + PROG_MAX_FQUEUES); set->lock = (pthread_mutex_t *) (set->fqueues + PROG_MAX_FQUEUES * (SHM_BUFFER_SIZE)); return set; fail_mmap: - if (flags & O_CREAT) + if (oflags & O_CREAT) shm_unlink(fn); + fail_truncate: + close(fd); fail_shm_open: free(set); fail_malloc: @@ -206,7 +196,7 @@ struct shm_flow_set * shm_flow_set_open(pid_t pid) void shm_flow_set_destroy(struct shm_flow_set * set) { - char fn[25]; + char fn[FN_MAX_CHARS]; assert(set); @@ -221,7 +211,7 @@ void shm_flow_set_close(struct shm_flow_set * set) { assert(set); - munmap(set->mtable, SHM_FLOW_SET_FILE_SIZE); + munmap(set->mtable, SHM_FSET_FILE_SIZE); free(set); } @@ -307,6 +297,8 @@ void shm_flow_set_notify(struct shm_flow_set * set, int flow_id, int event) { + struct flowevent * e; + assert(set); assert(!(flow_id < 0) && flow_id < SYS_MAX_FLOWS); @@ -317,10 +309,13 @@ void shm_flow_set_notify(struct shm_flow_set * set, return; } - (fqueue_ptr(set, set->mtable[flow_id]) + - (set->heads[set->mtable[flow_id]]))->flow_id = flow_id; - (fqueue_ptr(set, set->mtable[flow_id]) + - (set->heads[set->mtable[flow_id]])++)->event = event; + e = fqueue_ptr(set, set->mtable[flow_id]) + + set->heads[set->mtable[flow_id]]; + + e->flow_id = flow_id; + e->event = event; + + ++set->heads[set->mtable[flow_id]]; pthread_cond_signal(&set->conds[set->mtable[flow_id]]); @@ -330,7 +325,7 @@ void shm_flow_set_notify(struct shm_flow_set * set, ssize_t shm_flow_set_wait(const struct shm_flow_set * set, size_t idx, - int * fqueue, + struct flowevent * fqueue, const struct timespec * abstime) { ssize_t ret = 0; @@ -346,22 +341,14 @@ ssize_t shm_flow_set_wait(const struct shm_flow_set * set, pthread_mutex_consistent(set->lock); #endif - pthread_cleanup_push((void(*)(void *))pthread_mutex_unlock, - (void *) set->lock); + pthread_cleanup_push(__cleanup_mutex_unlock, set->lock); while (set->heads[idx] == 0 && ret != -ETIMEDOUT) { - if (abstime != NULL) { - ret = -pthread_cond_timedwait(set->conds + idx, - set->lock, - abstime); + ret = -__timedwait(set->conds + idx, set->lock, abstime); #ifdef HAVE_CANCEL_BUG - if (ret == -ETIMEDOUT) - pthread_testcancel(); + if (ret == -ETIMEDOUT) + pthread_testcancel(); #endif - } else { - ret = -pthread_cond_wait(set->conds + idx, - set->lock); - } #ifdef HAVE_ROBUST_MUTEX if (ret == -EOWNERDEAD) pthread_mutex_consistent(set->lock); @@ -371,7 +358,7 @@ ssize_t shm_flow_set_wait(const struct shm_flow_set * set, if (ret != -ETIMEDOUT) { memcpy(fqueue, fqueue_ptr(set, idx), - set->heads[idx] * sizeof(struct portevent)); + set->heads[idx] * sizeof(*fqueue)); ret = set->heads[idx]; set->heads[idx] = 0; } diff --git a/src/lib/shm_rbuff.c b/src/lib/shm_rbuff.c index 0a07e799..22cff41c 100644 --- a/src/lib/shm_rbuff.c +++ b/src/lib/shm_rbuff.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Ring buffer implementations for incoming packets * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License @@ -26,22 +26,22 @@ #include <ouroboros/shm_rbuff.h> #include <ouroboros/lockfile.h> -#include <ouroboros/time_utils.h> #include <ouroboros/errno.h> #include <ouroboros/fccntl.h> +#include <ouroboros/pthread.h> +#include <ouroboros/time.h> -#include <pthread.h> -#include <sys/mman.h> +#include <assert.h> #include <fcntl.h> +#include <signal.h> +#include <stdbool.h> +#include <stdint.h> +#include <stdio.h> #include <stdlib.h> #include <string.h> -#include <stdio.h> -#include <stdint.h> #include <unistd.h> -#include <signal.h> +#include <sys/mman.h> #include <sys/stat.h> -#include <assert.h> -#include <stdbool.h> #define FN_MAX_CHARS 255 @@ -69,20 +69,11 @@ struct shm_rbuff { int flow_id; /* flow_id of the flow */ }; -void shm_rbuff_close(struct shm_rbuff * rb) -{ - assert(rb); - - munmap(rb->shm_base, SHM_RB_FILE_SIZE); - - free(rb); -} - #define MM_FLAGS (PROT_READ | PROT_WRITE) -struct shm_rbuff * rbuff_create(pid_t pid, - int flow_id, - int flags) +static struct shm_rbuff * rbuff_create(pid_t pid, + int flow_id, + int flags) { struct shm_rbuff * rb; int fd; @@ -99,7 +90,7 @@ struct shm_rbuff * rbuff_create(pid_t pid, if (fd == -1) goto fail_open; - if ((flags & O_CREAT) && ftruncate(fd, SHM_RB_FILE_SIZE - 1) < 0) + if ((flags & O_CREAT) && ftruncate(fd, SHM_RB_FILE_SIZE) < 0) goto fail_truncate; shm_base = mmap(NULL, SHM_RB_FILE_SIZE, MM_FLAGS, MAP_SHARED, fd, 0); @@ -130,6 +121,13 @@ struct shm_rbuff * rbuff_create(pid_t pid, return NULL; } +static void rbuff_destroy(struct shm_rbuff * rb) +{ + munmap(rb->shm_base, SHM_RB_FILE_SIZE); + + free(rb); +} + struct shm_rbuff * shm_rbuff_create(pid_t pid, int flow_id) { @@ -174,7 +172,7 @@ struct shm_rbuff * shm_rbuff_create(pid_t pid, *rb->head = 0; *rb->tail = 0; - rb->pid = pid; + rb->pid = pid; rb->flow_id = flow_id; pthread_mutexattr_destroy(&mattr); @@ -202,6 +200,13 @@ struct shm_rbuff * shm_rbuff_open(pid_t pid, return rbuff_create(pid, flow_id, O_RDWR); } +void shm_rbuff_close(struct shm_rbuff * rb) +{ + assert(rb); + + rbuff_destroy(rb); +} + #if (defined(SHM_RBUFF_LOCKLESS) && \ (defined(__GNUC__) || defined (__clang__))) #include "shm_rbuff_ll.c" diff --git a/src/lib/shm_rbuff_ll.c b/src/lib/shm_rbuff_ll.c index 7c2a7d60..46a5314e 100644 --- a/src/lib/shm_rbuff_ll.c +++ b/src/lib/shm_rbuff_ll.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Lockless ring buffer for incoming packets * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License @@ -31,6 +31,11 @@ void shm_rbuff_destroy(struct shm_rbuff * rb) sprintf(fn, SHM_RBUFF_PREFIX "%d.%d", rb->pid, rb->flow_id); + __sync_bool_compare_and_swap(rb->acl, *rb->acl, ACL_FLOWDOWN); + + pthread_cond_broadcast(rb->del); + pthread_cond_broadcast(rb->add); + shm_rbuff_close(rb); shm_unlink(fn); @@ -100,16 +105,10 @@ int shm_rbuff_write_b(struct shm_rbuff * rb, goto err; } - pthread_cleanup_push((void(*)(void *))pthread_mutex_unlock, - (void *) rb->lock); + pthread_cleanup_push(__cleanup_mutex_unlock, rb->lock); while (!shm_rbuff_free(rb) && ret != -ETIMEDOUT) { - if (abstime != NULL) - ret = -pthread_cond_timedwait(rb->add, - rb->lock, - abstime); - else - ret = -pthread_cond_wait(rb->add, rb->lock); + ret = -__timedwait(rb->add, rb->lock, abstime); #ifdef HAVE_ROBUST_MUTEX if (ret == -EOWNERDEAD) pthread_mutex_consistent(rb->lock); @@ -139,9 +138,15 @@ ssize_t shm_rbuff_read(struct shm_rbuff * rb) assert(rb); - if (shm_rbuff_empty(rb)) - return __sync_fetch_and_add(rb->acl, 0) & ACL_FLOWDOWN ? - -EFLOWDOWN : -EAGAIN; + if (shm_rbuff_empty(rb)) { + if (_sync_fetch_and_add(rb->acl, 0) & ACL_FLOWDOWN) + return -EFLOWDOWN; + + if (_sync_fetch_and_add(rb->acl, 0) & ACL_FLOWPEER) + return -EFLOWPEER; + + return -EAGAIN; + } ntail = RB_TAIL; @@ -174,16 +179,10 @@ ssize_t shm_rbuff_read_b(struct shm_rbuff * rb, if (pthread_mutex_lock(rb->lock) == EOWNERDEAD) pthread_mutex_consistent(rb->lock); #endif - pthread_cleanup_push((void(*)(void *))pthread_mutex_unlock, - (void *) rb->lock); + pthread_cleanup_push(__cleanup_mutex_unlock, rb->lock); while (shm_rbuff_empty(rb) && (idx != -ETIMEDOUT)) { - if (abstime != NULL) - idx = -pthread_cond_timedwait(rb->add, - rb->lock, - abstime); - else - idx = -pthread_cond_wait(rb->add, rb->lock); + idx = -__timedwait(rb->add, rb->lock, abstime); #ifdef HAVE_ROBUST_MUTEX if (idx == -EOWNERDEAD) pthread_mutex_consistent(rb->lock); @@ -230,8 +229,7 @@ void shm_rbuff_fini(struct shm_rbuff * rb) pthread_mutex_consistent(rb->lock); #endif - pthread_cleanup_push((void(*)(void *))pthread_mutex_unlock, - (void *) rb->lock); + pthread_cleanup_push(__cleanup_mutex_unlock, rb->lock); while (!shm_rbuff_empty(rb)) #ifndef HAVE_ROBUST_MUTEX diff --git a/src/lib/shm_rbuff_pthr.c b/src/lib/shm_rbuff_pthr.c index 00ffd583..b543fb07 100644 --- a/src/lib/shm_rbuff_pthr.c +++ b/src/lib/shm_rbuff_pthr.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Ring buffer for incoming packets * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License @@ -24,12 +24,15 @@ void shm_rbuff_destroy(struct shm_rbuff * rb) { char fn[FN_MAX_CHARS]; - assert(rb); + assert(rb != NULL); #ifdef CONFIG_OUROBOROS_DEBUG pthread_mutex_lock(rb->lock); - assert(shm_rbuff_empty(rb)); + *rb->acl = *rb->acl & ACL_FLOWDOWN; + + pthread_cond_broadcast(rb->del); + pthread_cond_broadcast(rb->add); pthread_mutex_unlock(rb->lock); #endif @@ -45,7 +48,7 @@ int shm_rbuff_write(struct shm_rbuff * rb, { int ret = 0; - assert(rb); + assert(rb != NULL); assert(idx < SHM_BUFFER_SIZE); #ifndef HAVE_ROBUST_MUTEX @@ -88,7 +91,7 @@ int shm_rbuff_write_b(struct shm_rbuff * rb, { int ret = 0; - assert(rb); + assert(rb != NULL); assert(idx < SHM_BUFFER_SIZE); #ifndef HAVE_ROBUST_MUTEX @@ -106,16 +109,12 @@ int shm_rbuff_write_b(struct shm_rbuff * rb, goto err; } - pthread_cleanup_push((void(*)(void *))pthread_mutex_unlock, - (void *) rb->lock); + pthread_cleanup_push(__cleanup_mutex_unlock, rb->lock); - while (!shm_rbuff_free(rb) && ret != -ETIMEDOUT) { - if (abstime != NULL) - ret = -pthread_cond_timedwait(rb->del, - rb->lock, - abstime); - else - ret = -pthread_cond_wait(rb->del, rb->lock); + while (!shm_rbuff_free(rb) + && ret != -ETIMEDOUT + && !(*rb->acl & ACL_FLOWDOWN)) { + ret = -__timedwait(rb->del, rb->lock, abstime); #ifdef HAVE_ROBUST_MUTEX if (ret == -EOWNERDEAD) pthread_mutex_consistent(rb->lock); @@ -137,11 +136,24 @@ int shm_rbuff_write_b(struct shm_rbuff * rb, return ret; } +static int check_rb_acl(struct shm_rbuff * rb) +{ + assert(rb != NULL); + + if (*rb->acl & ACL_FLOWDOWN) + return -EFLOWDOWN; + + if (*rb->acl & ACL_FLOWPEER) + return -EFLOWPEER; + + return -EAGAIN; +} + ssize_t shm_rbuff_read(struct shm_rbuff * rb) { ssize_t ret = 0; - assert(rb); + assert(rb != NULL); #ifndef HAVE_ROBUST_MUTEX pthread_mutex_lock(rb->lock); @@ -151,7 +163,7 @@ ssize_t shm_rbuff_read(struct shm_rbuff * rb) #endif if (shm_rbuff_empty(rb)) { - ret = *rb->acl & ACL_FLOWDOWN ? -EFLOWDOWN : -EAGAIN; + ret = check_rb_acl(rb); pthread_mutex_unlock(rb->lock); return ret; } @@ -170,7 +182,7 @@ ssize_t shm_rbuff_read_b(struct shm_rbuff * rb, { ssize_t idx = -1; - assert(rb); + assert(rb != NULL); #ifndef HAVE_ROBUST_MUTEX pthread_mutex_lock(rb->lock); @@ -184,37 +196,37 @@ ssize_t shm_rbuff_read_b(struct shm_rbuff * rb, return -EFLOWDOWN; } - pthread_cleanup_push((void(*)(void *))pthread_mutex_unlock, - (void *) rb->lock); + pthread_cleanup_push(__cleanup_mutex_unlock, rb->lock); - while (shm_rbuff_empty(rb) && (idx != -ETIMEDOUT)) { - if (abstime != NULL) - idx = -pthread_cond_timedwait(rb->add, - rb->lock, - abstime); - else - idx = -pthread_cond_wait(rb->add, rb->lock); + while (shm_rbuff_empty(rb) && + idx != -ETIMEDOUT && + check_rb_acl(rb) == -EAGAIN) { + idx = -__timedwait(rb->add, rb->lock, abstime); #ifdef HAVE_ROBUST_MUTEX if (idx == -EOWNERDEAD) pthread_mutex_consistent(rb->lock); #endif } - if (idx != -ETIMEDOUT) { + if (!shm_rbuff_empty(rb)) { idx = *tail_el_ptr(rb); *rb->tail = (*rb->tail + 1) & ((SHM_RBUFF_SIZE) - 1); pthread_cond_broadcast(rb->del); + } else if (idx != -ETIMEDOUT) { + idx = check_rb_acl(rb); } pthread_cleanup_pop(true); + assert(idx != -EAGAIN); + return idx; } void shm_rbuff_set_acl(struct shm_rbuff * rb, uint32_t flags) { - assert(rb); + assert(rb != NULL); #ifndef HAVE_ROBUST_MUTEX pthread_mutex_lock(rb->lock); @@ -224,6 +236,9 @@ void shm_rbuff_set_acl(struct shm_rbuff * rb, #endif *rb->acl = (size_t) flags; + pthread_cond_broadcast(rb->del); + pthread_cond_broadcast(rb->add); + pthread_mutex_unlock(rb->lock); } @@ -231,7 +246,7 @@ uint32_t shm_rbuff_get_acl(struct shm_rbuff * rb) { uint32_t flags; - assert(rb); + assert(rb != NULL); #ifndef HAVE_ROBUST_MUTEX pthread_mutex_lock(rb->lock); @@ -248,7 +263,7 @@ uint32_t shm_rbuff_get_acl(struct shm_rbuff * rb) void shm_rbuff_fini(struct shm_rbuff * rb) { - assert(rb); + assert(rb != NULL); #ifndef HAVE_ROBUST_MUTEX pthread_mutex_lock(rb->lock); @@ -256,8 +271,7 @@ void shm_rbuff_fini(struct shm_rbuff * rb) if (pthread_mutex_lock(rb->lock) == EOWNERDEAD) pthread_mutex_consistent(rb->lock); #endif - pthread_cleanup_push((void(*)(void *))pthread_mutex_unlock, - (void *) rb->lock); + pthread_cleanup_push(__cleanup_mutex_unlock, rb->lock); while (!shm_rbuff_empty(rb)) #ifndef HAVE_ROBUST_MUTEX @@ -273,7 +287,7 @@ size_t shm_rbuff_queued(struct shm_rbuff * rb) { size_t ret; - assert(rb); + assert(rb != NULL); #ifndef HAVE_ROBUST_MUTEX pthread_mutex_lock(rb->lock); diff --git a/src/lib/shm_rdrbuff.c b/src/lib/shm_rdrbuff.c index 9a003e21..7ad1bd2e 100644 --- a/src/lib/shm_rdrbuff.c +++ b/src/lib/shm_rdrbuff.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Random Deletion Ring Buffer for Data Units * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License @@ -25,21 +25,19 @@ #include "config.h" #include <ouroboros/errno.h> +#include <ouroboros/pthread.h> #include <ouroboros/shm_rdrbuff.h> -#include <ouroboros/shm_du_buff.h> -#include <ouroboros/time_utils.h> -#include <pthread.h> -#include <sys/mman.h> +#include <assert.h> #include <fcntl.h> -#include <unistd.h> +#include <signal.h> +#include <stdbool.h> +#include <stdio.h> #include <stdlib.h> #include <string.h> -#include <stdio.h> -#include <signal.h> +#include <unistd.h> +#include <sys/mman.h> #include <sys/stat.h> -#include <stdbool.h> -#include <assert.h> #define SHM_BLOCKS_SIZE ((SHM_BUFFER_SIZE) * SHM_RDRB_BLOCK_SIZE) #define SHM_FILE_SIZE (SHM_BLOCKS_SIZE + 2 * sizeof(size_t) \ @@ -101,12 +99,14 @@ static void garbage_collect(struct shm_rdrbuff * rdrb) pthread_cond_broadcast(rdrb->healthy); } +#ifdef HAVE_ROBUST_MUTEX static void sanitize(struct shm_rdrbuff * rdrb) { --get_head_ptr(rdrb)->refs; garbage_collect(rdrb); pthread_mutex_consistent(rdrb->lock); } +#endif static char * rdrb_filename(void) { @@ -171,7 +171,7 @@ static struct shm_rdrbuff * rdrb_create(int flags) if (fd == -1) goto fail_open; - if ((flags & O_CREAT) && ftruncate(fd, SHM_FILE_SIZE - 1) < 0) + if ((flags & O_CREAT) && ftruncate(fd, SHM_FILE_SIZE) < 0) goto fail_truncate; shm_base = mmap(NULL, SHM_FILE_SIZE, MM_FLAGS, MAP_SHARED, fd, 0); @@ -203,7 +203,7 @@ static struct shm_rdrbuff * rdrb_create(int flags) return NULL; } -struct shm_rdrbuff * shm_rdrbuff_create() +struct shm_rdrbuff * shm_rdrbuff_create(void) { struct shm_rdrbuff * rdrb; mode_t mask; @@ -261,7 +261,7 @@ struct shm_rdrbuff * shm_rdrbuff_create() return NULL; } -struct shm_rdrbuff * shm_rdrbuff_open() +struct shm_rdrbuff * shm_rdrbuff_open(void) { return rdrb_create(O_RDWR); } @@ -390,8 +390,7 @@ ssize_t shm_rdrbuff_alloc_b(struct shm_rdrbuff * rdrb, if (pthread_mutex_lock(rdrb->lock) == EOWNERDEAD) sanitize(rdrb); #endif - pthread_cleanup_push((void (*) (void *)) pthread_mutex_unlock, - (void *) rdrb->lock); + pthread_cleanup_push(__cleanup_mutex_unlock, rdrb->lock); #ifdef SHM_RDRB_MULTI_BLOCK if (blocks + *rdrb->head > (SHM_BUFFER_SIZE)) @@ -401,13 +400,7 @@ ssize_t shm_rdrbuff_alloc_b(struct shm_rdrbuff * rdrb, #else while (!shm_rdrb_free(rdrb, 1) && ret != ETIMEDOUT) { #endif - if (abstime != NULL) - ret = pthread_cond_timedwait(rdrb->healthy, - rdrb->lock, - abstime); - else - ret = pthread_cond_wait(rdrb->healthy, rdrb->lock); - + ret = __timedwait(rdrb->healthy, rdrb->lock, abstime); #ifdef SHM_RDRB_MULTI_BLOCK if (blocks + *rdrb->head > (SHM_BUFFER_SIZE)) padblocks = (SHM_BUFFER_SIZE) - *rdrb->head; @@ -495,7 +488,7 @@ int shm_rdrbuff_remove(struct shm_rdrbuff * rdrb, if (pthread_mutex_lock(rdrb->lock) == EOWNERDEAD) sanitize(rdrb); #endif - assert(!shm_rdrb_empty(rdrb)); + /* assert(!shm_rdrb_empty(rdrb)); */ sdb = idx_to_du_buff_ptr(rdrb, idx); @@ -531,6 +524,13 @@ uint8_t * shm_du_buff_tail(struct shm_du_buff * sdb) return (uint8_t *) (sdb + 1) + sdb->du_tail; } +size_t shm_du_buff_len(struct shm_du_buff * sdb) +{ + assert(sdb); + + return sdb->du_tail - sdb->du_head; +} + uint8_t * shm_du_buff_head_alloc(struct shm_du_buff * sdb, size_t size) { @@ -577,7 +577,7 @@ uint8_t * shm_du_buff_head_release(struct shm_du_buff * sdb, } uint8_t * shm_du_buff_tail_release(struct shm_du_buff * sdb, - size_t size) + size_t size) { assert(sdb); assert(!(size > sdb->du_tail - sdb->du_head)); diff --git a/src/lib/sockets.c b/src/lib/sockets.c index f25ee2ce..13219db0 100644 --- a/src/lib/sockets.c +++ b/src/lib/sockets.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * The sockets layer to communicate between daemons * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License @@ -21,6 +21,7 @@ */ #include <ouroboros/errno.h> +#include <ouroboros/pthread.h> #include <ouroboros/sockets.h> #include <ouroboros/utils.h> @@ -29,9 +30,7 @@ #include <string.h> #include <stdio.h> #include <stdlib.h> -#include <pthread.h> #include <stdbool.h> -#include <sys/time.h> /* Apple doesn't support SEQPACKET. */ #ifdef __APPLE__ @@ -40,6 +39,11 @@ #define SOCK_TYPE SOCK_SEQPACKET #endif +void __cleanup_close_ptr(void * fd) +{ + close(*(int *) fd); +} + int client_socket_open(char * file_name) { int sockfd; @@ -52,8 +56,7 @@ int client_socket_open(char * file_name) serv_addr.sun_family = AF_UNIX; sprintf(serv_addr.sun_path, "%s", file_name); - if (connect(sockfd, - (struct sockaddr *) &serv_addr, + if (connect(sockfd, (struct sockaddr *) &serv_addr, sizeof(serv_addr))) { close(sockfd); return -1; @@ -95,17 +98,11 @@ int server_socket_open(char * file_name) return sockfd; } -static void close_ptr(void * o) -{ - close(*(int *) o); -} - irm_msg_t * send_recv_irm_msg(irm_msg_t * msg) { - int sockfd; - uint8_t buf[SOCK_BUF_SIZE]; - ssize_t len; - irm_msg_t * recv_msg = NULL; + int sockfd; + uint8_t buf[SOCK_BUF_SIZE]; + ssize_t len; sockfd = client_socket_open(IRM_SOCK_PATH); if (sockfd < 0) @@ -117,19 +114,44 @@ irm_msg_t * send_recv_irm_msg(irm_msg_t * msg) return NULL; } - pthread_cleanup_push(close_ptr, &sockfd); - irm_msg__pack(msg, buf); - if (write(sockfd, buf, len) != -1) + pthread_cleanup_push(__cleanup_close_ptr, &sockfd); + + len = write(sockfd, buf, len); + if (len >= 0) len = read(sockfd, buf, SOCK_BUF_SIZE); - if (len > 0) - recv_msg = irm_msg__unpack(NULL, len, buf); + pthread_cleanup_pop(true); + + if (len < 0) + goto fail; + + return irm_msg__unpack(NULL, len, buf); + fail: + return NULL; +} + +int send_recv_msg(buffer_t * msg) +{ + int sockfd; + ssize_t len = 0; + + sockfd = client_socket_open(IRM_SOCK_PATH); + if (sockfd < 0) + return -1; + + pthread_cleanup_push(__cleanup_close_ptr, &sockfd); + + len = write(sockfd, msg->data, msg->len); + if (len >= 0) + len = read(sockfd, msg->data, SOCK_BUF_SIZE); pthread_cleanup_pop(true); - return recv_msg; + msg->len = (size_t) len; + + return len < 0 ? -1 : 0; } char * ipcp_sock_path(pid_t pid) @@ -165,40 +187,3 @@ char * ipcp_sock_path(pid_t pid) return full_name; } - -qosspec_msg_t spec_to_msg(const qosspec_t * qs) -{ - qosspec_t spec; - qosspec_msg_t msg = QOSSPEC_MSG__INIT; - - spec = (qs == NULL ? qos_raw : *qs); - - msg.delay = spec.delay; - msg.bandwidth = spec.bandwidth; - msg.availability = spec.availability; - msg.loss = spec.loss; - msg.ber = spec.ber; - msg.in_order = spec.in_order; - msg.max_gap = spec.max_gap; - msg.cypher_s = spec.cypher_s; - - return msg; -} - -qosspec_t msg_to_spec(const qosspec_msg_t * msg) -{ - qosspec_t spec; - - assert(msg); - - spec.delay = msg->delay; - spec.bandwidth = msg->bandwidth; - spec.availability = msg->availability; - spec.loss = msg->loss; - spec.ber = msg->ber; - spec.in_order = msg->in_order; - spec.max_gap = msg->max_gap; - spec.cypher_s = msg->cypher_s; - - return spec; -} diff --git a/src/lib/tests/CMakeLists.txt b/src/lib/tests/CMakeLists.txt index 9e23b0ee..dc90671b 100644 --- a/src/lib/tests/CMakeLists.txt +++ b/src/lib/tests/CMakeLists.txt @@ -6,10 +6,11 @@ create_test_sourcelist(${PARENT_DIR}_tests test_suite.c bitmap_test.c btree_test.c crc32_test.c + hash_test.c md5_test.c sha3_test.c shm_rbuff_test.c - time_utils_test.c + time_test.c ) add_executable(${PARENT_DIR}_test EXCLUDE_FROM_ALL ${${PARENT_DIR}_tests}) @@ -19,7 +20,11 @@ target_link_libraries(${PARENT_DIR}_test ouroboros-common) add_dependencies(check ${PARENT_DIR}_test) set(tests_to_run ${${PARENT_DIR}_tests}) -remove(tests_to_run test_suite.c) +if(CMAKE_VERSION VERSION_LESS "3.29.0") + remove(tests_to_run test_suite.c) +else () + list(POP_FRONT tests_to_run) +endif() foreach (test ${tests_to_run}) get_filename_component(test_name ${test} NAME_WE) diff --git a/src/lib/tests/bitmap_test.c b/src/lib/tests/bitmap_test.c index fdacdce8..4dbd6653 100644 --- a/src/lib/tests/bitmap_test.c +++ b/src/lib/tests/bitmap_test.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Test of the bitmap * - * 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 @@ -27,7 +27,8 @@ #define BITMAP_SIZE 200 -int bitmap_test(int argc, char ** argv) +int bitmap_test(int argc, + char ** argv) { struct bmp * bmp; ssize_t bits = BITMAP_SIZE; @@ -60,27 +61,23 @@ int bitmap_test(int argc, char ** argv) if (!bmp_is_id_valid(bmp, id)) { if (i < BITMAP_SIZE + offset) { printf("Failed valid ID %d (%zd).\n", i, id); - bmp_destroy(bmp); - return -1; + goto fail; } if (id >= offset && id < bits + offset) { printf("Valid ID %zd returned invalid.\n", id); - bmp_destroy(bmp); - return -1; + goto fail; } continue; } if (!bmp_is_id_used(bmp, id)) { printf("ID not marked in use.\n"); - bmp_destroy(bmp); - return -1; + goto fail; } if (id != i) { printf("Wrong ID returned.\n"); - bmp_destroy(bmp); - return -1; + goto fail; } } @@ -89,20 +86,24 @@ int bitmap_test(int argc, char ** argv) if (bmp_release(bmp, r)) { printf("Failed to release ID.\n"); - return -1; + goto fail; } id = bmp_allocate(bmp); if (!bmp_is_id_valid(bmp, id)) continue; + if (id != r) { printf("Wrong prev ID returned.\n"); - bmp_destroy(bmp); - return -1; + goto fail; } } bmp_destroy(bmp); return 0; + + fail: + bmp_destroy(bmp); + return -1; } diff --git a/src/lib/tests/btree_test.c b/src/lib/tests/btree_test.c index 00b3cde1..8bd30370 100644 --- a/src/lib/tests/btree_test.c +++ b/src/lib/tests/btree_test.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Test of the B-tree implementation * - * 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 diff --git a/src/lib/tests/crc32_test.c b/src/lib/tests/crc32_test.c index 35f8a1cd..a26c8220 100644 --- a/src/lib/tests/crc32_test.c +++ b/src/lib/tests/crc32_test.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Test of the CRC32 function * - * 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 diff --git a/src/lib/tests/hash_test.c b/src/lib/tests/hash_test.c new file mode 100644 index 00000000..970d9185 --- /dev/null +++ b/src/lib/tests/hash_test.c @@ -0,0 +1,202 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2024 + * + * Test of the hashing functions + * + * 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/. + */ + +#include <ouroboros/hash.h> +#include <ouroboros/test.h> + +#include <stdlib.h> +#include <stdint.h> +#include <assert.h> +#include <string.h> +#include <stdio.h> + +/* + * Test vectors calculated at + * https://www.lammertbies.nl/comm/info/crc-calculation.html + */ + +struct vec_entry { + char * in; + char * out; +}; + +static int test_crc32(void) +{ + int ret = 0; + + struct vec_entry vec [] = { + { "0", "f4dbdf21" }, + { "123456789", "cbf43926" }, + { "987654321", "015f0201" }, + { NULL, NULL } + }; + + struct vec_entry * cur = vec; + + TEST_START(); + + while (cur->in != NULL) { + uint8_t crc[4]; + char res[9]; + + str_hash(HASH_CRC32, crc, cur->in); + + sprintf(res, HASH_FMT32, HASH_VAL32(crc)); + if (strcmp(res, cur->out) != 0) { + printf("Hash failed %s != %s.\n", res, cur->out); + ret |= -1; + } + + ++cur; + } + + TEST_END(ret); + + return ret; +} + +static int test_md5(void) +{ + int ret = 0; + + struct vec_entry vec [] = {{ + "abc", + "900150983cd24fb0d6963f7d28e17f72" + }, { + "The quick brown fox jumps over the lazy dog", + "9e107d9d372bb6826bd81d3542a419d6" + }, { + "abcdbcdecdefdefgefghfghighijhijkijkljklmklmnlmnomnopnopq", + "8215ef0796a20bcaaae116d3876c664a" + }, { + NULL, + NULL + }}; + + struct vec_entry * cur = vec; + + TEST_START(); + + + while (cur->in != NULL) { + uint8_t md5[16]; + char res[33]; + + str_hash(HASH_MD5, md5, cur->in); + + sprintf(res, HASH_FMT128, HASH_VAL128(md5)); + if (strcmp(res, cur->out) != 0) { + printf("Hash failed %s != %s.\n", res, cur->out); + ret |= -1; + } + + ++cur; + } + + TEST_END(ret); + + return ret; +} + +static int test_sha3(void) +{ + int ret = 0; + + uint8_t sha3[64]; + char res[129]; + + char * in = "abc"; + + char * out = + "e642824c3f8cf24ad09234ee7d3c766f" + "c9a3a5168d0c94ad73b46fdf"; + + TEST_START(); + + str_hash(HASH_SHA3_224, sha3, in); + + sprintf(res, HASH_FMT224, HASH_VAL224(sha3)); + if (strcmp(res, out) != 0) { + printf("SHA3-224 failed %s != %s", res, out); + ret |= -1; + } + + out = + "3a985da74fe225b2045c172d6bd390bd" + "855f086e3e9d525b46bfe24511431532"; + + str_hash(HASH_SHA3_256, sha3, in); + + sprintf(res, HASH_FMT256, HASH_VAL256(sha3)); + if (strcmp(res, out) != 0) { + printf("SHA3-256 failed %s != %s.\n", res, out); + ret |= -1; + } + + out = + "ec01498288516fc926459f58e2c6ad8d" + "f9b473cb0fc08c2596da7cf0e49be4b2" + "98d88cea927ac7f539f1edf228376d25"; + + str_hash(HASH_SHA3_384, sha3, in); + + sprintf(res, HASH_FMT384, HASH_VAL384(sha3)); + if (strcmp(res, out) != 0) { + printf("SHA3-384failed %s != %s.'n", res, out); + ret |= -1; + } + + out = + "b751850b1a57168a5693cd924b6b096e" + "08f621827444f70d884f5d0240d2712e" + "10e116e9192af3c91a7ec57647e39340" + "57340b4cf408d5a56592f8274eec53f0"; + + str_hash(HASH_SHA3_512, sha3, in); + + sprintf(res, HASH_FMT512, HASH_VAL512(sha3)); + if (strcmp(res, out) != 0) { + printf("SHA3-512 failed %s != %s.\n", res, out); + ret |= -1; + } + + TEST_END(ret); + + return ret; +} + +int hash_test(int argc, + char ** argv) +{ + int ret = 0; + + (void) argc; + (void) argv; + + ret |= test_crc32(); + + ret |= test_md5(); + + ret |= test_sha3(); + + return ret; +} diff --git a/src/lib/tests/md5_test.c b/src/lib/tests/md5_test.c index 00497ae7..28e8f42f 100644 --- a/src/lib/tests/md5_test.c +++ b/src/lib/tests/md5_test.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Test of the MD5 function * - * 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 diff --git a/src/lib/tests/sha3_test.c b/src/lib/tests/sha3_test.c index 945402ee..82b4ef0d 100644 --- a/src/lib/tests/sha3_test.c +++ b/src/lib/tests/sha3_test.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Test of the SHA3 function * - * 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 diff --git a/src/lib/tests/shm_rbuff_test.c b/src/lib/tests/shm_rbuff_test.c index 1d16a09d..e36c3229 100644 --- a/src/lib/tests/shm_rbuff_test.c +++ b/src/lib/tests/shm_rbuff_test.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Test of the shm_rbuff * - * 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 diff --git a/src/lib/tests/time_utils_test.c b/src/lib/tests/time_test.c index c384d272..65f896bb 100644 --- a/src/lib/tests/time_utils_test.c +++ b/src/lib/tests/time_test.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Test of the time utilities * - * 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 @@ -22,7 +22,7 @@ #define _POSIX_C_SOURCE 200809L -#include <ouroboros/time_utils.h> +#include <ouroboros/time.h> #include <stdio.h> @@ -66,8 +66,8 @@ static int tv_check(struct timeval * v, return v->tv_sec == sec && v->tv_usec == usec; } -int time_utils_test(int argc, - char ** argv) +int time_test(int argc, + char ** argv) { struct timespec s0; struct timespec s1; diff --git a/src/lib/timerwheel.c b/src/lib/timerwheel.c new file mode 100644 index 00000000..96f4ac47 --- /dev/null +++ b/src/lib/timerwheel.c @@ -0,0 +1,414 @@ +/* + * Ouroboros - Copyright (C) 2016 - 2024 + * + * Timerwheel + * + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> + * + * 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/. + */ + +#include <ouroboros/list.h> + +/* Overflow limits range to about 6 hours. */ +#define ts_to_ns(ts) (ts.tv_sec * BILLION + ts.tv_nsec) +#define ts_to_rxm_slot(ts) (ts_to_ns(ts) >> RXMQ_RES) +#define ts_to_ack_slot(ts) (ts_to_ns(ts) >> ACKQ_RES) + +struct rxm { + struct list_head next; + uint32_t seqno; +#ifndef RXM_BUFFER_ON_HEAP + struct shm_du_buff * sdb; +#endif + struct frct_pci * pkt; + size_t len; + time_t t0; /* Time when original was sent (us). */ + struct frcti * frcti; + int fd; + int flow_id; /* Prevent rtx when fd reused. */ +}; + +struct ack { + struct list_head next; + struct frcti * frcti; + int fd; + int flow_id; +}; + +struct { + /* + * At a 1 ms min resolution, every level bumps the + * resolution by a factor of 16. + */ + struct list_head rxms[RXMQ_LVLS][RXMQ_SLOTS]; + + struct list_head acks[ACKQ_SLOTS]; + bool map[ACKQ_SLOTS][PROG_MAX_FLOWS]; + + size_t prv_rxm[RXMQ_LVLS]; /* Last processed rxm slots. */ + size_t prv_ack; /* Last processed ack slot. */ + pthread_mutex_t lock; +} rw; + +static void timerwheel_fini(void) +{ + size_t i; + size_t j; + struct list_head * p; + struct list_head * h; + + pthread_mutex_lock(&rw.lock); + + for (i = 0; i < RXMQ_LVLS; ++i) { + for (j = 0; j < RXMQ_SLOTS; j++) { + list_for_each_safe(p, h, &rw.rxms[i][j]) { + struct rxm * rxm; + rxm = list_entry(p, struct rxm, next); + list_del(&rxm->next); +#ifdef RXM_BUFFER_ON_HEAP + free(rxm->pkt); +#else + shm_du_buff_ack(rxm->sdb); + ipcp_sdb_release(rxm->sdb); +#endif + free(rxm); + } + } + } + + for (i = 0; i < ACKQ_SLOTS; ++i) { + list_for_each_safe(p, h, &rw.acks[i]) { + struct ack * a = list_entry(p, struct ack, next); + list_del(&a->next); + free(a); + } + } + + pthread_mutex_unlock(&rw.lock); + + pthread_mutex_destroy(&rw.lock); +} + +static int timerwheel_init(void) +{ + struct timespec now; + size_t i; + size_t j; + + if (pthread_mutex_init(&rw.lock, NULL)) + return -1; + + clock_gettime(PTHREAD_COND_CLOCK, &now); + + for (i = 0; i < RXMQ_LVLS; ++i) { + rw.prv_rxm[i] = (ts_to_rxm_slot(now) - 1); + rw.prv_rxm[i] >>= (RXMQ_BUMP * i); + rw.prv_rxm[i] &= (RXMQ_SLOTS - 1); + for (j = 0; j < RXMQ_SLOTS; ++j) + list_head_init(&rw.rxms[i][j]); + } + + rw.prv_ack = (ts_to_ack_slot(now) - 1) & (ACKQ_SLOTS - 1); + for (i = 0; i < ACKQ_SLOTS; ++i) + list_head_init(&rw.acks[i]); + + return 0; +} + +static void timerwheel_move(void) +{ + struct timespec now; + struct list_head * p; + struct list_head * h; + size_t rxm_slot; + size_t ack_slot; + size_t i; + size_t j; + + pthread_mutex_lock(&rw.lock); + + pthread_cleanup_push(__cleanup_mutex_unlock, &rw.lock); + + clock_gettime(PTHREAD_COND_CLOCK, &now); + + rxm_slot = ts_to_rxm_slot(now); + + for (i = 0; i < RXMQ_LVLS; ++i) { + size_t j_max_slot = rxm_slot & (RXMQ_SLOTS - 1); + j = rw.prv_rxm[i]; + if (j_max_slot < j) + j_max_slot += RXMQ_SLOTS; + while (j++ < j_max_slot) { + list_for_each_safe(p, h, + &rw.rxms[i][j & (RXMQ_SLOTS - 1)]) { + struct rxm * r; + struct frct_cr * snd_cr; + struct frct_cr * rcv_cr; + size_t slot; + size_t rslot; + ssize_t idx; + struct shm_du_buff * sdb; + struct frct_pci * pci; + struct flow * f; + uint32_t snd_lwe; + uint32_t rcv_lwe; + size_t lvl = 0; + + r = list_entry(p, struct rxm, next); + + list_del(&r->next); + + snd_cr = &r->frcti->snd_cr; + rcv_cr = &r->frcti->rcv_cr; + f = &ai.flows[r->fd]; +#ifndef RXM_BUFFER_ON_HEAP + shm_du_buff_ack(r->sdb); +#endif + if (f->frcti == NULL + || f->info.id != r->flow_id) + goto cleanup; + + pthread_rwlock_rdlock(&r->frcti->lock); + + snd_lwe = snd_cr->lwe; + rcv_lwe = rcv_cr->lwe; + + pthread_rwlock_unlock(&r->frcti->lock); + + /* Has been ack'd, remove. */ + if (before(r->seqno, snd_lwe)) + goto cleanup; + + /* Check for r-timer expiry. */ + if (ts_to_ns(now) - r->t0 > r->frcti->r) + goto flow_down; + + pthread_rwlock_wrlock(&r->frcti->lock); + + if (r->seqno == r->frcti->rttseq) { + r->frcti->rto += + r->frcti->rto >> RTO_DIV; + r->frcti->probe = false; + } +#ifdef PROC_FLOW_STATS + r->frcti->n_rtx++; +#endif + rslot = r->frcti->rto >> RXMQ_RES; + + pthread_rwlock_unlock(&r->frcti->lock); + + /* Schedule at least in the next time slot. */ + slot = ts_to_ns(now) >> RXMQ_RES; + + while (rslot >= RXMQ_SLOTS) { + ++lvl; + rslot >>= RXMQ_BUMP; + slot >>= RXMQ_BUMP; + } + + if (lvl >= RXMQ_LVLS) /* Can't reschedule */ + goto flow_down; + + rslot = (rslot + slot + 1) & (RXMQ_SLOTS - 1); +#ifdef RXM_BLOCKING + if (ipcp_sdb_reserve(&sdb, r->len) < 0) +#else + if (shm_rdrbuff_alloc(ai.rdrb, r->len, NULL, + &sdb) < 0) +#endif + goto reschedule; /* rdrbuff full */ + + pci = (struct frct_pci *) shm_du_buff_head(sdb); + memcpy(pci, r->pkt, r->len); +#ifndef RXM_BUFFER_ON_HEAP + ipcp_sdb_release(r->sdb); + r->sdb = sdb; + r->pkt = pci; + shm_du_buff_wait_ack(sdb); +#endif + idx = shm_du_buff_get_idx(sdb); + + /* Retransmit the copy. */ + pci->ackno = hton32(rcv_lwe); +#ifdef RXM_BLOCKING + if (shm_rbuff_write_b(f->tx_rb, idx, NULL) < 0) +#else + if (shm_rbuff_write(f->tx_rb, idx) < 0) +#endif + goto flow_down; + shm_flow_set_notify(f->set, f->info.id, + FLOW_PKT); + reschedule: + list_add(&r->next, &rw.rxms[lvl][rslot]); + continue; + + flow_down: + shm_rbuff_set_acl(f->tx_rb, ACL_FLOWDOWN); + shm_rbuff_set_acl(f->rx_rb, ACL_FLOWDOWN); + cleanup: +#ifdef RXM_BUFFER_ON_HEAP + free(r->pkt); +#else + ipcp_sdb_release(r->sdb); +#endif + free(r); + } + } + rw.prv_rxm[i] = rxm_slot & (RXMQ_SLOTS - 1); + /* Move up a level in the wheel. */ + rxm_slot >>= RXMQ_BUMP; + } + + ack_slot = ts_to_ack_slot(now) & (ACKQ_SLOTS - 1) ; + + j = rw.prv_ack; + + if (ack_slot < j) + ack_slot += ACKQ_SLOTS; + + while (j++ < ack_slot) { + list_for_each_safe(p, h, &rw.acks[j & (ACKQ_SLOTS - 1)]) { + struct ack * a; + struct flow * f; + + a = list_entry(p, struct ack, next); + + list_del(&a->next); + + f = &ai.flows[a->fd]; + + rw.map[j & (ACKQ_SLOTS - 1)][a->fd] = false; + + if (f->info.id == a->flow_id && f->frcti != NULL) + send_frct_pkt(a->frcti); + + free(a); + } + } + + rw.prv_ack = ack_slot & (ACKQ_SLOTS - 1); + + pthread_cleanup_pop(true); +} + +static int timerwheel_rxm(struct frcti * frcti, + uint32_t seqno, + struct shm_du_buff * sdb) +{ + struct timespec now; + struct rxm * r; + size_t slot; + size_t lvl = 0; + time_t rto_slot; + + r = malloc(sizeof(*r)); + if (r == NULL) + return -ENOMEM; + + clock_gettime(PTHREAD_COND_CLOCK, &now); + + r->t0 = ts_to_ns(now); + r->seqno = seqno; + r->frcti = frcti; + r->len = shm_du_buff_len(sdb); +#ifdef RXM_BUFFER_ON_HEAP + r->pkt = malloc(r->len); + if (r->pkt == NULL) { + free(r); + return -ENOMEM; + } + memcpy(r->pkt, shm_du_buff_head(sdb), r->len); +#else + r->sdb = sdb; + r->pkt = (struct frct_pci *) shm_du_buff_head(sdb); +#endif + pthread_rwlock_rdlock(&r->frcti->lock); + + rto_slot = frcti->rto >> RXMQ_RES; + slot = r->t0 >> RXMQ_RES; + + r->fd = frcti->fd; + r->flow_id = ai.flows[r->fd].info.id; + + pthread_rwlock_unlock(&r->frcti->lock); + + while (rto_slot >= RXMQ_SLOTS) { + ++lvl; + rto_slot >>= RXMQ_BUMP; + slot >>= RXMQ_BUMP; + } + + if (lvl >= RXMQ_LVLS) { /* Out of timerwheel range. */ +#ifdef RXM_BUFFER_ON_HEAP + free(r->pkt); +#endif + free(r); + return -EPERM; + } + + slot = (slot + rto_slot + 1) & (RXMQ_SLOTS - 1); + + pthread_mutex_lock(&rw.lock); + + list_add_tail(&r->next, &rw.rxms[lvl][slot]); +#ifndef RXM_BUFFER_ON_HEAP + shm_du_buff_wait_ack(sdb); +#endif + pthread_mutex_unlock(&rw.lock); + + return 0; +} + +static int timerwheel_delayed_ack(int fd, + struct frcti * frcti) +{ + struct timespec now; + struct ack * a; + size_t slot; + + a = malloc(sizeof(*a)); + if (a == NULL) + return -ENOMEM; + + clock_gettime(PTHREAD_COND_CLOCK, &now); + + pthread_rwlock_rdlock(&frcti->lock); + + slot = (((ts_to_ns(now) + (TICTIME << 1)) >> ACKQ_RES) + 1) + & (ACKQ_SLOTS - 1); + + pthread_rwlock_unlock(&frcti->lock); + + a->fd = fd; + a->frcti = frcti; + a->flow_id = ai.flows[fd].info.id; + + pthread_mutex_lock(&rw.lock); + + if (rw.map[slot][fd]) { + pthread_mutex_unlock(&rw.lock); + free(a); + return 0; + } + + rw.map[slot][fd] = true; + + list_add_tail(&a->next, &rw.acks[slot]); + + pthread_mutex_unlock(&rw.lock); + + return 0; +} diff --git a/src/lib/tpm.c b/src/lib/tpm.c index ff0a22df..0ef1fda8 100644 --- a/src/lib/tpm.c +++ b/src/lib/tpm.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Threadpool management * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License @@ -26,12 +26,12 @@ #include <ouroboros/errno.h> #include <ouroboros/list.h> -#include <ouroboros/time_utils.h> +#include <ouroboros/time.h> #include <ouroboros/tpm.h> +#include <assert.h> #include <pthread.h> #include <stdlib.h> -#include <assert.h> #define TPM_TIMEOUT 1000 @@ -117,8 +117,7 @@ static void tpm_kill(struct tpm * tpm) static void * tpmgr(void * o) { struct timespec dl; - struct timespec to = {(TPM_TIMEOUT / 1000), - (TPM_TIMEOUT % 1000) * MILLION}; + struct timespec to = TIMESPEC_INIT_MS(TPM_TIMEOUT); struct tpm * tpm = (struct tpm *) o; while (true) { @@ -239,12 +238,12 @@ void tpm_stop(struct tpm * tpm) tpm->state = TPM_NULL; pthread_mutex_unlock(&tpm->lock); + + pthread_join(tpm->mgr, NULL); } void tpm_destroy(struct tpm * tpm) { - pthread_join(tpm->mgr, NULL); - pthread_mutex_destroy(&tpm->lock); pthread_cond_destroy(&tpm->cond); diff --git a/src/lib/utils.c b/src/lib/utils.c index e3024283..fdbcd9d9 100644 --- a/src/lib/utils.c +++ b/src/lib/utils.c @@ -1,10 +1,10 @@ /* - * Ouroboros - Copyright (C) 2016 - 2020 + * Ouroboros - Copyright (C) 2016 - 2024 * * Handy utilities * - * Dimitri Staessens <dimitri.staessens@ugent.be> - * Sander Vrijders <sander.vrijders@ugent.be> + * Dimitri Staessens <dimitri@ouroboros.rocks> + * Sander Vrijders <sander@ouroboros.rocks> * * This library is free software; you can redistribute it and/or * modify it under the terms of the GNU Lesser General Public License @@ -20,6 +20,10 @@ * Foundation, Inc., http://www.fsf.org/about/contact/. */ +#define _POSIX_C_SOURCE 200809L + +#include <ouroboros/utils.h> + #include <stdlib.h> #include <string.h> @@ -35,14 +39,14 @@ int n_digits(unsigned i) return n; } -char * path_strip(char * src) +char * path_strip(const char * src) { - char * dst = NULL; + char * dst; if (src == NULL) return NULL; - dst = src + strlen(src); + dst = (char *) src + strlen(src); while (dst > src && *dst != '/') --dst; @@ -52,3 +56,58 @@ char * path_strip(char * src) return dst; } + +size_t argvlen(const char ** argv) +{ + size_t argc = 0; + + if (argv == NULL) + return 0; + + while (*argv++ != NULL) + argc++; + + return argc; +} + +void argvfree(char ** argv) +{ + char ** argv_dup; + + if (argv == NULL) + return; + + argv_dup = argv; + while (*argv_dup != NULL) + free(*(argv_dup++)); + + free(argv); +} + +char ** argvdup(char ** argv) +{ + int argc = 0; + char ** argv_dup = argv; + int i; + + if (argv == NULL) + return NULL; + + while (*(argv_dup++) != NULL) + argc++; + + argv_dup = malloc((argc + 1) * sizeof(*argv_dup)); + if (argv_dup == NULL) + return NULL; + + for (i = 0; i < argc; ++i) { + argv_dup[i] = strdup(argv[i]); + if (argv_dup[i] == NULL) { + argvfree(argv_dup); + return NULL; + } + } + + argv_dup[argc] = NULL; + return argv_dup; +} |