From 11e2bf77bf5b3301efb2518c141ab4bd821cb8c0 Mon Sep 17 00:00:00 2001 From: theMackabu Date: Fri, 1 May 2026 12:38:51 -0700 Subject: [PATCH] add dynamic route table --- include/routes.h | 43 +++++++++++ src/routes.c | 173 ++++++++++++++++++++++++++++++++++++++++++++ tests/test_routes.c | 42 +++++++++++ 3 files changed, 258 insertions(+) create mode 100644 include/routes.h create mode 100644 src/routes.c create mode 100644 tests/test_routes.c diff --git a/include/routes.h b/include/routes.h new file mode 100644 index 0000000..221f27c --- /dev/null +++ b/include/routes.h @@ -0,0 +1,43 @@ +#ifndef RPC_ROUTES_H +#define RPC_ROUTES_H + +#include "rpc/server.h" + +#include +#include + +#define RPC_ROUTE_PAGE_BITS 16u +#define RPC_ROUTE_PAGE_SIZE (1u << RPC_ROUTE_PAGE_BITS) +#define RPC_ROUTE_PAGE_MASK (RPC_ROUTE_PAGE_SIZE - 1u) +#define RPC_ROUTE_ROOT_SIZE (1u << (32u - RPC_ROUTE_PAGE_BITS)) + +typedef struct rpc_route { + uint32_t proc_id; + rpc_handler_fn handler; + void *user_data; +} rpc_route; + +typedef struct rpc_retired_route { + rpc_route *route; + struct rpc_retired_route *next; +} rpc_retired_route; + +typedef _Atomic(rpc_route *) rpc_route_slot; + +typedef struct rpc_routes { + _Atomic(rpc_route_slot *) *pages; + size_t root_bytes; + size_t page_bytes; + atomic_uint active_readers; + pthread_mutex_t mutate_lock; + rpc_retired_route *retired; +} rpc_routes; + +int rpc_routes_init(rpc_routes *routes); +void rpc_routes_destroy(rpc_routes *routes); +int rpc_routes_add(rpc_routes *routes, uint32_t proc_id, rpc_handler_fn handler, + void *user_data); +int rpc_routes_remove(rpc_routes *routes, uint32_t proc_id); +int rpc_routes_lookup(rpc_routes *routes, uint32_t proc_id, rpc_route *out); + +#endif diff --git a/src/routes.c b/src/routes.c new file mode 100644 index 0000000..b55b7c2 --- /dev/null +++ b/src/routes.c @@ -0,0 +1,173 @@ +#include "routes.h" + +#include +#include +#include +#include + +static void *route_mmap(size_t bytes) { + void *p = mmap(NULL, bytes, PROT_READ | PROT_WRITE, MAP_PRIVATE | MAP_ANON, -1, 0); + return p == MAP_FAILED ? NULL : p; +} + +static rpc_route_slot *route_page_alloc(const rpc_routes *routes) { + return route_mmap(routes->page_bytes); +} + +static void free_retired(rpc_routes *routes) { + if (atomic_load_explicit(&routes->active_readers, memory_order_acquire) != 0) return; + + rpc_retired_route *node = routes->retired; + routes->retired = NULL; + + while (node) { + rpc_retired_route *next = node->next; + free(node->route); + free(node); + node = next; + } +} + +static int retire_route(rpc_routes *routes, rpc_route *route) { + if (!route) return 0; + + rpc_retired_route *node = calloc(1, sizeof(*node)); + if (!node) return -1; + + node->route = route; + node->next = routes->retired; + routes->retired = node; + free_retired(routes); + + return 0; +} + +int rpc_routes_init(rpc_routes *routes) { + if (!routes) return -1; + + memset(routes, 0, sizeof(*routes)); + routes->root_bytes = RPC_ROUTE_ROOT_SIZE * sizeof(*routes->pages); + routes->page_bytes = RPC_ROUTE_PAGE_SIZE * sizeof(rpc_route_slot); + routes->pages = route_mmap(routes->root_bytes); + + if (!routes->pages) return -1; + + if (pthread_mutex_init(&routes->mutate_lock, NULL) != 0) { + munmap(routes->pages, routes->root_bytes); + memset(routes, 0, sizeof(*routes)); + return -1; + } + + atomic_init(&routes->active_readers, 0); + + return 0; +} + +void rpc_routes_destroy(rpc_routes *routes) { + if (!routes) return; + + pthread_mutex_lock(&routes->mutate_lock); + if (!routes->pages) goto retired; + + for (size_t page_idx = 0; page_idx < RPC_ROUTE_ROOT_SIZE; ++page_idx) { + rpc_route_slot *page = atomic_load_explicit(&routes->pages[page_idx], memory_order_relaxed); + if (!page) continue; + + for (size_t i = 0; i < RPC_ROUTE_PAGE_SIZE; ++i) { + rpc_route *route = atomic_load_explicit(&page[i], memory_order_relaxed); + free(route); + } + munmap(page, routes->page_bytes); + } + munmap(routes->pages, routes->root_bytes); + +retired: + rpc_retired_route *node = routes->retired; + + while (node) { + rpc_retired_route *next = node->next; + free(node->route); + free(node); + node = next; + } + + routes->retired = NULL; + pthread_mutex_unlock(&routes->mutate_lock); + pthread_mutex_destroy(&routes->mutate_lock); +} + +int rpc_routes_add(rpc_routes *routes, uint32_t proc_id, rpc_handler_fn handler, void *user_data) { + if (!routes || !handler) return -1; + + rpc_route *route = calloc(1, sizeof(*route)); + if (!route) return -1; + + *route = (rpc_route){ + .proc_id = proc_id, + .handler = handler, + .user_data = user_data + }; + + pthread_mutex_lock(&routes->mutate_lock); + uint32_t page_idx = proc_id >> RPC_ROUTE_PAGE_BITS; + uint32_t slot_idx = proc_id & RPC_ROUTE_PAGE_MASK; + rpc_route_slot *page = atomic_load_explicit(&routes->pages[page_idx], memory_order_acquire); + if (!page) { + page = route_page_alloc(routes); + if (!page) { + pthread_mutex_unlock(&routes->mutate_lock); + free(route); + return -1; + } + atomic_store_explicit(&routes->pages[page_idx], page, memory_order_release); + } + rpc_route *old = atomic_exchange_explicit( + &page[slot_idx], + route, memory_order_acq_rel + ); + + int rc = retire_route(routes, old); + pthread_mutex_unlock(&routes->mutate_lock); + + return rc; +} + +int rpc_routes_remove(rpc_routes *routes, uint32_t proc_id) { + if (!routes) return -1; + + pthread_mutex_lock(&routes->mutate_lock); + uint32_t page_idx = proc_id >> RPC_ROUTE_PAGE_BITS; + uint32_t slot_idx = proc_id & RPC_ROUTE_PAGE_MASK; + rpc_route_slot *page = atomic_load_explicit(&routes->pages[page_idx], memory_order_acquire); + rpc_route *old = page + ? atomic_exchange_explicit(&page[slot_idx], NULL, memory_order_acq_rel) + : NULL; + + int rc = old ? retire_route(routes, old) : -1; + pthread_mutex_unlock(&routes->mutate_lock); + + return rc; +} + +int rpc_routes_lookup(rpc_routes *routes, uint32_t proc_id, rpc_route *out) { + if (!routes || !out) return -1; + + atomic_fetch_add_explicit(&routes->active_readers, 1u, memory_order_acquire); + uint32_t page_idx = proc_id >> RPC_ROUTE_PAGE_BITS; + uint32_t slot_idx = proc_id & RPC_ROUTE_PAGE_MASK; + rpc_route_slot *page = atomic_load_explicit(&routes->pages[page_idx], memory_order_acquire); + rpc_route *route = page + ? atomic_load_explicit(&page[slot_idx], memory_order_acquire) + : NULL; + if (route) { + *out = *route; + } + unsigned old = + atomic_fetch_sub_explicit(&routes->active_readers, 1u, memory_order_release); + if (old == 1u) { + pthread_mutex_lock(&routes->mutate_lock); + free_retired(routes); + pthread_mutex_unlock(&routes->mutate_lock); + } + return route ? 0 : -1; +} diff --git a/tests/test_routes.c b/tests/test_routes.c new file mode 100644 index 0000000..56a8a89 --- /dev/null +++ b/tests/test_routes.c @@ -0,0 +1,42 @@ +#include "routes.h" + +#include + +static int handler_a(rpc_ctx *ctx, const rpc_value *args, size_t argc, + rpc_writer *out, void *user_data) { + (void)ctx; + (void)args; + (void)argc; + (void)out; + (void)user_data; + return 0; +} + +static int handler_b(rpc_ctx *ctx, const rpc_value *args, size_t argc, + rpc_writer *out, void *user_data) { + (void)ctx; + (void)args; + (void)argc; + (void)out; + (void)user_data; + return 0; +} + +int main(void) { + rpc_routes routes; + rpc_route route; + assert(rpc_routes_init(&routes) == 0); + assert(rpc_routes_lookup(&routes, 10, &route) != 0); + assert(rpc_routes_add(&routes, 10, handler_a, (void *)1) == 0); + assert(rpc_routes_lookup(&routes, 10, &route) == 0); + assert(route.handler == handler_a); + assert(route.user_data == (void *)1); + assert(rpc_routes_add(&routes, 10, handler_b, (void *)2) == 0); + assert(rpc_routes_lookup(&routes, 10, &route) == 0); + assert(route.handler == handler_b); + assert(route.user_data == (void *)2); + assert(rpc_routes_remove(&routes, 10) == 0); + assert(rpc_routes_lookup(&routes, 10, &route) != 0); + rpc_routes_destroy(&routes); + return 0; +} -- 2.51.2