From 5650acd375cd177c56333d8a80900077a7b8ca4a Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Fri, 26 Jun 2026 05:24:00 +0000 Subject: [PATCH] feat(raft-drive-async): SlowLane bounded executor for long blocking I/O (task 1.2) Dedicated-pthread pool isolated from the vtask pool; non-blocking slowlane_submit (0 enqueued / -EEXIST deduped / -EAGAIN cap-full-deferred), admission cap on running tasks, per-key in-flight dedup, run-exactly-once, drain+join fini. Wired into controller lifecycle (dormant; consumers land in 2.x/3.x). 7 unit tests, mutation-verified cap+dedup; full suite 194/0 green, 15x ASAN. --- .kiro/specs/raft-drive-async/tasks.md | 2 +- src/controller/cluster.c | 25 ++ src/controller/include/slowlane.h | 95 ++++++ src/controller/meson.build | 3 + src/controller/slowlane.c | 264 +++++++++++++++++ src/controller/testsuite/controller_test.c | 2 + src/controller/testsuite/test_slowlane.c | 318 +++++++++++++++++++++ src/include/xce/cluster_def.h | 2 + 8 files changed, 710 insertions(+), 1 deletion(-) create mode 100644 src/controller/include/slowlane.h create mode 100644 src/controller/slowlane.c create mode 100644 src/controller/testsuite/test_slowlane.c diff --git a/.kiro/specs/raft-drive-async/tasks.md b/.kiro/specs/raft-drive-async/tasks.md index 2867f4f..491b14c 100644 --- a/.kiro/specs/raft-drive-async/tasks.md +++ b/.kiro/specs/raft-drive-async/tasks.md @@ -10,7 +10,7 @@ Tasks mirror the design's 4-phase migration. The slow lane is foundational (ever - Observable: a build with the new constants compiles; defaults are readable at controller startup; the resolved reservation logged at init. - _Requirements: 2.3, 3.1_ -- [ ] 1.2 Implement the SlowLane executor +- [x] 1.2 Implement the SlowLane executor - Bounded pool of dedicated worker threads, isolated from the vtask request pool; non-blocking submit that enqueues and returns immediately. - Admission cap on concurrent in-flight tasks; per-key in-flight dedup (key 0 disables dedup); init/fini wired into the controller lifecycle. - Observable: a submitted task runs exactly once on a lane worker; submit never blocks the caller; concurrent running tasks never exceed the cap; a submit with an in-flight key is a no-op. diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 6cd12e2..83ed211 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -1367,6 +1367,25 @@ int cluster_init_phase2(cluster_t* c) } } + /* raft-drive-async task 1.2: bring up the SlowLane executor — the dedicated + * bounded-thread offload lane for long blocking I/O (snapshot transfer, + * config-push, peer reconnect) that must never run on the iomgr/delaymgr + * drive threads. Dormant infra here: no production caller until the + * reconnect (2.x) / snapshot/config-push (3.x) offloads land. Sized from + * the DEFAULT_SLOWLANE_* constants (task 1.1, config.h). Guard c->slowlane + * so the legacy two-phase path (phase2 can run twice) inits exactly once. */ + if (c->slowlane.workers == NULL) { + rc = slowlane_init(&c->slowlane, DEFAULT_SLOWLANE_WORKERS, + DEFAULT_SLOWLANE_MAX_INFLIGHT); + if (rc != 0) { + ERROR("slowlane_init failed: %d (workers=%d, max_inflight=%d)", + rc, DEFAULT_SLOWLANE_WORKERS, DEFAULT_SLOWLANE_MAX_INFLIGHT); + return rc; + } + INFO("drive-async: slowlane up (workers=%d, max_inflight=%d)", + DEFAULT_SLOWLANE_WORKERS, DEFAULT_SLOWLANE_MAX_INFLIGHT); + } + /* Role set by ha_elect_evaluate_and_signal() inside cluster_run() */ if (ha_is_primary(&c->ha)) ha_promote_to_primary(c); @@ -1743,6 +1762,12 @@ void cluster_fini(cluster_t* c) /* Stop HA watchdog thread before tearing down other subsystems — P1-5 */ ha_fini(&c->ha); + /* raft-drive-async task 1.2: drain + join the SlowLane after the drive + * threads have stopped submitting (ha_fini above) and before persist/iomgr + * teardown, since lane tasks (snapshot/reconnect, once wired) touch those. + * No-op if phase2 never ran (workers==NULL guard inside slowlane_fini). */ + slowlane_fini(&c->slowlane); + /* Stop persistence before tearing down the rest */ persist_stop(&c->persist); persist_fini(&c->persist); diff --git a/src/controller/include/slowlane.h b/src/controller/include/slowlane.h new file mode 100644 index 0000000..4c3c56d --- /dev/null +++ b/src/controller/include/slowlane.h @@ -0,0 +1,95 @@ +#pragma once +/* + * slowlane.h — SlowLane executor: a bounded pool of dedicated worker threads + * that runs long blocking I/O OFF the Raft drive threads (iomgr epoll loop, + * delaymgr timer), with admission control and per-key in-flight dedup. + * + * Service interface is verbatim from + * .kiro/specs/raft-drive-async/design.md (Components -> SlowLane). + * + * The lane owns scheduling/isolation, NOT the work content: the submitted + * `fn` (peer connect, snapshot transfer, config-push body) stays in its own + * module and is responsible for clearing any caller-side in-flight state on + * completion. `slowlane_submit` is non-blocking: a drive-thread caller + * enqueues and returns immediately, never waiting for `fn` to run. + * + * raft-drive-async task 1.2 — Requirements 3.1, 3.2, 3.3, 3.4, 1.5. + */ +#include +#include + +typedef void (*slowlane_fn_t)(void *arg); + +/* + * One queued unit of work. Allocated by the lane on submit, freed by the + * worker after `fn` returns. Held on the FIFO pending queue until a worker + * picks it up. + */ +typedef struct slowlane_task { + uint64_t key; /* dedup key; 0 disables dedup */ + slowlane_fn_t fn; + void *arg; + struct slowlane_task *next; /* intrusive FIFO link */ +} slowlane_task_t; + +/* + * SlowLane executor state. All fields are lane-private and guarded by `lock`; + * callers touch the lane only through the API below. Bounded by design: + * - `n_workers` dedicated pthreads, isolated from the vtask request pool; + * - at most `max_inflight` tasks RUNNING concurrently (admission cap); + * - at most one in-flight task per non-zero key (dedup). + */ +typedef struct slowlane { + pthread_mutex_t lock; + pthread_cond_t cond; /* signalled when work or a slot frees up */ + + slowlane_task_t *q_head; /* FIFO pending queue (admission-deferred) */ + slowlane_task_t *q_tail; + + uint64_t *inflight_keys; /* running OR queued non-zero keys (dedup set) */ + int n_inflight_keys; + int cap_inflight_keys; + + int n_workers; + int max_inflight; /* admission cap on concurrently RUNNING tasks */ + int n_running; /* tasks currently executing on a worker */ + + pthread_t *workers; + int started; /* worker threads created */ + int stopping; /* fini requested: drain + join */ +} slowlane_t; + +/* + * Initialise the lane and start `n_workers` dedicated worker threads. + * `max_inflight` bounds the number of tasks RUNNING concurrently (excess is + * queued, never dropped, never run on a drive thread). + * Returns 0 on success, -EINVAL on a bad argument, -errno on a thread/alloc + * failure (lane left un-started). + */ +int slowlane_init(slowlane_t *sl, int n_workers, int max_inflight); + +/* + * Submit `fn(arg)` to run exactly once on a lane worker. NON-BLOCKING: the + * caller (which may be a drive thread) returns immediately. + * + * Returns (per design SlowLane Service Interface): + * 0 enqueued AND a worker slot was free (running < max_inflight) — + * the task can start immediately + * -EEXIST `key` (non-zero) already in flight — deduped no-op, NOT enqueued + * -EAGAIN admission cap full at submit time (running >= max_inflight) — the + * task IS still queued/deferred (never dropped, never run on a drive + * thread); the caller is told it was deferred, not that it failed + * -errno allocation failure / lane stopping + * + * In all of {0, -EAGAIN} the work is enqueued and will run exactly once. Only + * -EEXIST and -errno mean the task was not enqueued. + * + * `key == 0` disables dedup (every key-0 submit is independent work). + */ +int slowlane_submit(slowlane_t *sl, uint64_t key, slowlane_fn_t fn, void *arg); + +/* + * Drain the pending queue, stop and join all workers. Shutdown only. + * After return the lane is dormant; re-init to reuse. + */ +void slowlane_fini(slowlane_t *sl); diff --git a/src/controller/meson.build b/src/controller/meson.build index 9772228..59b77dd 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -26,6 +26,7 @@ controller_src = [ 'persist.c', 'repl.c', 'ctrl_metrics.c', + 'slowlane.c', ] incdir = include_directories('.', '../include', '../lib', @@ -140,6 +141,7 @@ controller_unit_test_src = [ 'testsuite/test_ha_vote_floor_heartbeat.c', 'testsuite/test_term_safety_adapter.c', 'testsuite/test_drive_async_config.c', + 'testsuite/test_slowlane.c', 'testsuite/test_waldump.c', 'testsuite/waldump.c', 'testsuite/test_walcompare.c', @@ -159,6 +161,7 @@ controller_unit_test_src = [ 'repl.c', 'ctrl_metrics.c', 'config_repl.c', + 'slowlane.c', ] ssl_dep = dependency('openssl') diff --git a/src/controller/slowlane.c b/src/controller/slowlane.c new file mode 100644 index 0000000..635348e --- /dev/null +++ b/src/controller/slowlane.c @@ -0,0 +1,264 @@ +#define _GNU_SOURCE +#include "slowlane.h" + +#include +#include +#include +#include + +/* + * slowlane.c — bounded, admission-capped executor with per-key in-flight + * dedup, running long blocking I/O off the Raft drive threads. + * + * raft-drive-async task 1.2 — Requirements 3.1, 3.2, 3.3, 3.4, 1.5. + * + * Concurrency model (single mutex `lock`, one condvar `cond`): + * - `q_head/q_tail` is the FIFO of admitted-but-not-yet-running tasks. + * - `n_running` counts tasks executing on a worker; admission holds + * `n_running <= max_inflight` at all times. + * - `inflight_keys[]` holds every non-zero key that is queued OR running; + * a key is added at submit and removed when its `fn` returns, so a second + * submit of the same key in that window dedups (-EEXIST). + * - Workers wait on `cond` for "work pending AND a running slot free". + * submit and task-completion both signal `cond`. + * + * Lock discipline: `fn` ALWAYS runs with `lock` released (it blocks for I/O); + * the lane never holds `lock` across user work — so a slow/stuck `fn` cannot + * stall submit() or other workers' admission decisions. + */ + +/* ── dedup key set helpers (caller holds sl->lock) ────────────────── */ + +static bool key_present_locked(const slowlane_t *sl, uint64_t key) +{ + for (int i = 0; i < sl->n_inflight_keys; i++) + if (sl->inflight_keys[i] == key) + return true; + return false; +} + +static int key_add_locked(slowlane_t *sl, uint64_t key) +{ + if (sl->n_inflight_keys == sl->cap_inflight_keys) { + int newcap = sl->cap_inflight_keys ? sl->cap_inflight_keys * 2 : 8; + uint64_t *grown = + realloc(sl->inflight_keys, (size_t)newcap * sizeof(uint64_t)); + if (grown == NULL) + return -ENOMEM; + sl->inflight_keys = grown; + sl->cap_inflight_keys = newcap; + } + sl->inflight_keys[sl->n_inflight_keys++] = key; + return 0; +} + +static void key_remove_locked(slowlane_t *sl, uint64_t key) +{ + if (key == 0) + return; + for (int i = 0; i < sl->n_inflight_keys; i++) { + if (sl->inflight_keys[i] == key) { + /* swap-remove (order irrelevant for a membership set) */ + sl->inflight_keys[i] = + sl->inflight_keys[--sl->n_inflight_keys]; + return; + } + } +} + +/* ── worker thread ────────────────────────────────────────────────── */ + +static void *slowlane_worker(void *arg) +{ + slowlane_t *sl = arg; + + pthread_mutex_lock(&sl->lock); + for (;;) { + /* Wait for: a queued task AND a free running slot. On shutdown we + * still drain the queue (so already-submitted work runs exactly once), + * but stop waiting for new arrivals. */ + while (!sl->stopping && + (sl->q_head == NULL || sl->n_running >= sl->max_inflight)) + pthread_cond_wait(&sl->cond, &sl->lock); + + if (sl->q_head == NULL) { + /* Nothing left to run. If stopping, exit; otherwise (can only be a + * spurious wake on a non-empty-but-cap-full path that then emptied) + * loop back to wait. */ + if (sl->stopping) + break; + continue; + } + + /* Even while stopping, respect the admission cap on the drain. */ + if (sl->n_running >= sl->max_inflight) { + if (sl->stopping) { + /* A free slot will appear when a running task completes and + * signals; wait for it so we can finish draining. */ + pthread_cond_wait(&sl->cond, &sl->lock); + } + continue; + } + + /* Pop FIFO head, claim a running slot. */ + slowlane_task_t *t = sl->q_head; + sl->q_head = t->next; + if (sl->q_head == NULL) + sl->q_tail = NULL; + sl->n_running++; + + pthread_mutex_unlock(&sl->lock); + /* Run user work with NO lock held (it blocks for I/O). Exactly once. */ + t->fn(t->arg); + uint64_t done_key = t->key; + free(t); + pthread_mutex_lock(&sl->lock); + + sl->n_running--; + key_remove_locked(sl, done_key); + /* A slot freed (admission) and a key cleared (dedup): wake a waiter + * that may now be admissible or a submit blocked on nothing here. */ + pthread_cond_broadcast(&sl->cond); + } + pthread_mutex_unlock(&sl->lock); + return NULL; +} + +/* ── public API ───────────────────────────────────────────────────── */ + +int slowlane_init(slowlane_t *sl, int n_workers, int max_inflight) +{ + if (sl == NULL || n_workers < 1 || max_inflight < 1) + return -EINVAL; + + memset(sl, 0, sizeof(*sl)); + sl->n_workers = n_workers; + sl->max_inflight = max_inflight; + + if (pthread_mutex_init(&sl->lock, NULL) != 0) + return -errno; + if (pthread_cond_init(&sl->cond, NULL) != 0) { + pthread_mutex_destroy(&sl->lock); + return -errno; + } + + sl->workers = calloc((size_t)n_workers, sizeof(pthread_t)); + if (sl->workers == NULL) { + pthread_cond_destroy(&sl->cond); + pthread_mutex_destroy(&sl->lock); + return -ENOMEM; + } + + for (int i = 0; i < n_workers; i++) { + int rc = pthread_create(&sl->workers[i], NULL, slowlane_worker, sl); + if (rc != 0) { + /* Roll back the partially-started pool: stop and join what exists. */ + pthread_mutex_lock(&sl->lock); + sl->stopping = 1; + pthread_cond_broadcast(&sl->cond); + pthread_mutex_unlock(&sl->lock); + for (int j = 0; j < i; j++) + pthread_join(sl->workers[j], NULL); + free(sl->workers); + sl->workers = NULL; + pthread_cond_destroy(&sl->cond); + pthread_mutex_destroy(&sl->lock); + return -rc; + } + sl->started++; + } + return 0; +} + +int slowlane_submit(slowlane_t *sl, uint64_t key, slowlane_fn_t fn, void *arg) +{ + if (sl == NULL || fn == NULL) + return -EINVAL; + + pthread_mutex_lock(&sl->lock); + + if (sl->stopping) { + pthread_mutex_unlock(&sl->lock); + return -ESHUTDOWN; + } + + /* Per-key in-flight dedup: a non-zero key already queued/running is a + * no-op (R1.5 snapshot-per-standby, R2.5 reconnect-per-peer). */ + if (key != 0 && key_present_locked(sl, key)) { + pthread_mutex_unlock(&sl->lock); + return -EEXIST; + } + + slowlane_task_t *t = calloc(1, sizeof(*t)); + if (t == NULL) { + pthread_mutex_unlock(&sl->lock); + return -ENOMEM; + } + t->key = key; + t->fn = fn; + t->arg = arg; + t->next = NULL; + + if (key != 0) { + int rc = key_add_locked(sl, key); + if (rc != 0) { + free(t); + pthread_mutex_unlock(&sl->lock); + return rc; + } + } + + /* Append to FIFO tail. */ + if (sl->q_tail != NULL) + sl->q_tail->next = t; + else + sl->q_head = t; + sl->q_tail = t; + + /* Admission status at submit time: a free running slot means the task can + * start immediately (0); a full cap means it is DEFERRED but still queued + * (-EAGAIN) — never dropped, never run on a drive thread. */ + int rc = (sl->n_running >= sl->max_inflight) ? -EAGAIN : 0; + + pthread_cond_signal(&sl->cond); + pthread_mutex_unlock(&sl->lock); + return rc; +} + +void slowlane_fini(slowlane_t *sl) +{ + if (sl == NULL || sl->workers == NULL) + return; + + /* Request drain+stop; workers finish the queued tasks, then exit. */ + pthread_mutex_lock(&sl->lock); + sl->stopping = 1; + pthread_cond_broadcast(&sl->cond); + pthread_mutex_unlock(&sl->lock); + + for (int i = 0; i < sl->started; i++) + pthread_join(sl->workers[i], NULL); + + /* All workers joined: no concurrent access remains. Free any tasks that + * were never run (only possible if a worker create failed mid-init; the + * normal drain path empties the queue). */ + slowlane_task_t *t = sl->q_head; + while (t != NULL) { + slowlane_task_t *next = t->next; + free(t); + t = next; + } + sl->q_head = sl->q_tail = NULL; + + free(sl->inflight_keys); + sl->inflight_keys = NULL; + sl->n_inflight_keys = 0; + sl->cap_inflight_keys = 0; + + free(sl->workers); + sl->workers = NULL; + sl->started = 0; + + pthread_cond_destroy(&sl->cond); + pthread_mutex_destroy(&sl->lock); +} diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index 3cc1e13..f3523b6 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -67,6 +67,7 @@ extern void test_term_safety_adapter(void); extern void test_waldump(void); extern void test_walcompare(void); extern void test_drive_async_config(void); +extern void test_slowlane(void); void setUp(void) { return; } @@ -203,6 +204,7 @@ int main(int argc, char* argv[]) TEST(test_waldump); TEST(test_walcompare); TEST(test_drive_async_config); + TEST(test_slowlane); return UNITY_END(); } diff --git a/src/controller/testsuite/test_slowlane.c b/src/controller/testsuite/test_slowlane.c new file mode 100644 index 0000000..003184b --- /dev/null +++ b/src/controller/testsuite/test_slowlane.c @@ -0,0 +1,318 @@ +#define _GNU_SOURCE +#include +#include +#include +#include +#include + +#include + +#include "slowlane.h" + +/* ==================================================================== + * raft-drive-async task 1.2 — SlowLane executor unit tests (RED first). + * + * The SlowLane is a bounded pool of dedicated worker threads that runs + * long blocking I/O off the Raft drive threads, with admission control + * and per-key in-flight dedup. These tests assert the four core + * behavioral contracts (acceptance criteria 3.1-3.4, 1.5): + * (a) slowlane_submit is non-blocking + * (b) concurrent RUNNING tasks never exceed max_inflight + * (c) a submit with an already-in-flight non-zero key returns -EEXIST + * (d) fn runs exactly once on a lane worker + * (+) -EAGAIN is returned (but work still queued) when the cap is full + * + * Determinism: every blocking task parks on a test-owned condvar gate + * (NOT a sleep), so the test controls exactly when a worker is occupied + * and when it releases. Concurrency peaks are observed via an atomic + * running counter the tasks themselves bump. + * ==================================================================== */ + +/* ── Test-owned gate: a controllable blocking task body ───────────── */ + +typedef struct { + pthread_mutex_t mu; + pthread_cond_t release_cv; /* test broadcasts to release parked tasks */ + pthread_cond_t entered_cv; /* task broadcasts when it has entered/parked */ + int gate_open; /* 0 = parked tasks must wait */ + + _Atomic int running; /* tasks currently inside fn (post-entry) */ + _Atomic int max_running; /* high-water mark of `running` */ + _Atomic int total_runs; /* how many times any task fn executed */ + _Atomic int entered; /* tasks that have reached the gate */ +} gate_t; + +static void gate_init(gate_t *g) +{ + pthread_mutex_init(&g->mu, NULL); + pthread_cond_init(&g->release_cv, NULL); + pthread_cond_init(&g->entered_cv, NULL); + g->gate_open = 0; + atomic_store(&g->running, 0); + atomic_store(&g->max_running, 0); + atomic_store(&g->total_runs, 0); + atomic_store(&g->entered, 0); +} + +static void gate_destroy(gate_t *g) +{ + pthread_mutex_destroy(&g->mu); + pthread_cond_destroy(&g->release_cv); + pthread_cond_destroy(&g->entered_cv); +} + +/* Blocking task body: bump running, record the peak, signal entry, then park + * on the gate until the test opens it. Runs on a lane worker. */ +static void gated_task(void *arg) +{ + gate_t *g = arg; + + int now = atomic_fetch_add(&g->running, 1) + 1; + int prev = atomic_load(&g->max_running); + while (now > prev && !atomic_compare_exchange_weak(&g->max_running, &prev, now)) + ; /* CAS-max */ + atomic_fetch_add(&g->total_runs, 1); + + pthread_mutex_lock(&g->mu); + atomic_fetch_add(&g->entered, 1); + pthread_cond_broadcast(&g->entered_cv); + while (!g->gate_open) + pthread_cond_wait(&g->release_cv, &g->mu); + pthread_mutex_unlock(&g->mu); + + atomic_fetch_sub(&g->running, 1); +} + +static void gate_open(gate_t *g) +{ + pthread_mutex_lock(&g->mu); + g->gate_open = 1; + pthread_cond_broadcast(&g->release_cv); + pthread_mutex_unlock(&g->mu); +} + +/* Wait (bounded) until at least `n` tasks have entered the gate. */ +static int wait_entered_at_least(gate_t *g, int n, int timeout_ms) +{ + struct timespec dl; + clock_gettime(CLOCK_REALTIME, &dl); + dl.tv_sec += timeout_ms / 1000; + dl.tv_nsec += (timeout_ms % 1000) * 1000000L; + if (dl.tv_nsec >= 1000000000L) { dl.tv_sec++; dl.tv_nsec -= 1000000000L; } + + pthread_mutex_lock(&g->mu); + while (atomic_load(&g->entered) < n) { + if (pthread_cond_timedwait(&g->entered_cv, &g->mu, &dl) == ETIMEDOUT) + break; + } + int got = atomic_load(&g->entered); + pthread_mutex_unlock(&g->mu); + return got; +} + +/* ── (a) submit is non-blocking ───────────────────────────────────── */ + +/* A submit must return promptly even when the task it submits will block + * indefinitely. We submit a gated (never-released until we open) task and + * assert submit returned without us opening the gate. */ +static void test_submit_is_non_blocking(void) +{ + slowlane_t sl; + gate_t g; + gate_init(&g); + TEST_ASSERT_EQUAL_INT(0, slowlane_init(&sl, 2, 2)); + + struct timespec t0, t1; + clock_gettime(CLOCK_MONOTONIC, &t0); + int rc = slowlane_submit(&sl, 0, gated_task, &g); + clock_gettime(CLOCK_MONOTONIC, &t1); + TEST_ASSERT_EQUAL_INT(0, rc); + + /* submit returned while the task is (or will be) parked: a blocking call + * would not have returned. Bound the wall time generously (<500ms). */ + long ms = (t1.tv_sec - t0.tv_sec) * 1000 + + (t1.tv_nsec - t0.tv_nsec) / 1000000; + TEST_ASSERT_TRUE_MESSAGE(ms < 500, "submit blocked the caller"); + + gate_open(&g); + slowlane_fini(&sl); /* drains + joins */ + gate_destroy(&g); +} + +/* ── (b) concurrent running tasks never exceed the cap ────────────── */ + +static void test_admission_cap_bounds_concurrency(void) +{ + slowlane_t sl; + gate_t g; + gate_init(&g); + const int n_workers = 4; + const int max_inflight = 2; + TEST_ASSERT_EQUAL_INT(0, slowlane_init(&sl, n_workers, max_inflight)); + + /* Submit more tasks than the cap; all must enqueue, but only `cap` may run + * concurrently even though 4 workers exist. */ + for (int i = 0; i < 6; i++) { + int rc = slowlane_submit(&sl, 0, gated_task, &g); + /* 0 while a slot is free, -EAGAIN once the cap is full (still queued). */ + TEST_ASSERT_TRUE_MESSAGE(rc == 0 || rc == -EAGAIN, "submit dropped work"); + } + + /* Let the first wave run and park at the gate. */ + int entered = wait_entered_at_least(&g, max_inflight, 2000); + TEST_ASSERT_EQUAL_INT(max_inflight, entered); + + /* Give any (incorrect) over-admission a chance to manifest, then assert the + * peak never exceeded the cap. */ + struct timespec nap = { .tv_sec = 0, .tv_nsec = 100 * 1000000L }; + nanosleep(&nap, NULL); + TEST_ASSERT_EQUAL_INT(max_inflight, atomic_load(&g.running)); + TEST_ASSERT_TRUE_MESSAGE(atomic_load(&g.max_running) <= max_inflight, + "concurrent running tasks exceeded the cap"); + + /* Release everything; all 6 must eventually run and the peak stays <= cap. */ + gate_open(&g); + slowlane_fini(&sl); + TEST_ASSERT_EQUAL_INT(6, atomic_load(&g.total_runs)); + TEST_ASSERT_TRUE_MESSAGE(atomic_load(&g.max_running) <= max_inflight, + "cap violated across the full run"); + gate_destroy(&g); +} + +/* ── (+) -EAGAIN when the cap is full at submit (work still queued) ── */ + +static void test_submit_returns_eagain_when_cap_full(void) +{ + slowlane_t sl; + gate_t g; + gate_init(&g); + const int max_inflight = 1; + TEST_ASSERT_EQUAL_INT(0, slowlane_init(&sl, 4, max_inflight)); + + /* First task takes the only running slot and parks. */ + TEST_ASSERT_EQUAL_INT(0, slowlane_submit(&sl, 0, gated_task, &g)); + TEST_ASSERT_EQUAL_INT(1, wait_entered_at_least(&g, 1, 2000)); + + /* Cap is now full: the next submit is deferred -> -EAGAIN, but queued. */ + int rc = slowlane_submit(&sl, 0, gated_task, &g); + TEST_ASSERT_EQUAL_INT_MESSAGE(-EAGAIN, rc, + "over-cap submit must return -EAGAIN"); + + gate_open(&g); + slowlane_fini(&sl); + /* The deferred task was NOT dropped: both ran. */ + TEST_ASSERT_EQUAL_INT(2, atomic_load(&g.total_runs)); + gate_destroy(&g); +} + +/* ── (c) per-key dedup returns -EEXIST for an in-flight key ────────── */ + +static void test_dedup_returns_eexist_for_inflight_key(void) +{ + slowlane_t sl; + gate_t g; + gate_init(&g); + TEST_ASSERT_EQUAL_INT(0, slowlane_init(&sl, 4, 4)); + + const uint64_t KEY = 0x51A8; + + /* First submit of KEY is accepted and runs/parks. */ + TEST_ASSERT_EQUAL_INT(0, slowlane_submit(&sl, KEY, gated_task, &g)); + TEST_ASSERT_EQUAL_INT(1, wait_entered_at_least(&g, 1, 2000)); + + /* Second submit of the SAME in-flight key is a deduped no-op. */ + TEST_ASSERT_EQUAL_INT(-EEXIST, + slowlane_submit(&sl, KEY, gated_task, &g)); + + /* A DIFFERENT key is still admitted. */ + TEST_ASSERT_EQUAL_INT(0, slowlane_submit(&sl, KEY + 1, gated_task, &g)); + TEST_ASSERT_EQUAL_INT(2, wait_entered_at_least(&g, 2, 2000)); + + gate_open(&g); + slowlane_fini(&sl); + /* Exactly 2 runs: the duplicate KEY submit did not run a third task. */ + TEST_ASSERT_EQUAL_INT(2, atomic_load(&g.total_runs)); + gate_destroy(&g); +} + +/* After an in-flight key completes, the SAME key is admittable again. */ +static void test_dedup_key_reusable_after_completion(void) +{ + slowlane_t sl; + gate_t g; + gate_init(&g); + TEST_ASSERT_EQUAL_INT(0, slowlane_init(&sl, 2, 2)); + + const uint64_t KEY = 7; + + TEST_ASSERT_EQUAL_INT(0, slowlane_submit(&sl, KEY, gated_task, &g)); + TEST_ASSERT_EQUAL_INT(1, wait_entered_at_least(&g, 1, 2000)); + /* Release and let it fully complete (running drops to 0, key cleared). */ + gate_open(&g); + + /* Re-close the gate for the second wave. */ + pthread_mutex_lock(&g.mu); + /* wait until the first task left fn */ + while (atomic_load(&g.running) != 0) { + pthread_mutex_unlock(&g.mu); + struct timespec s = { 0, 1 * 1000000L }; + nanosleep(&s, NULL); + pthread_mutex_lock(&g.mu); + } + g.gate_open = 0; + atomic_store(&g.entered, 0); + pthread_mutex_unlock(&g.mu); + + /* Same key, after completion -> admitted again (0, not -EEXIST). */ + TEST_ASSERT_EQUAL_INT(0, slowlane_submit(&sl, KEY, gated_task, &g)); + TEST_ASSERT_EQUAL_INT(1, wait_entered_at_least(&g, 1, 2000)); + + gate_open(&g); + slowlane_fini(&sl); + TEST_ASSERT_EQUAL_INT(2, atomic_load(&g.total_runs)); + gate_destroy(&g); +} + +/* ── (d) fn runs exactly once ──────────────────────────────────────── */ + +static void test_fn_runs_exactly_once(void) +{ + slowlane_t sl; + gate_t g; + gate_init(&g); + TEST_ASSERT_EQUAL_INT(0, slowlane_init(&sl, 4, 4)); + + /* The gate starts open so non-blocking tasks run straight through. */ + gate_open(&g); + + const int N = 32; + for (int i = 0; i < N; i++) + TEST_ASSERT_EQUAL_INT(0, slowlane_submit(&sl, 0, gated_task, &g)); + + slowlane_fini(&sl); /* drains + joins -> all submitted work has run */ + + /* Exactly N executions: no task dropped, none double-run. */ + TEST_ASSERT_EQUAL_INT(N, atomic_load(&g.total_runs)); + gate_destroy(&g); +} + +/* ── init/fini hardening: fini on an empty lane, double-fini safety ── */ + +static void test_fini_on_idle_lane(void) +{ + slowlane_t sl; + TEST_ASSERT_EQUAL_INT(0, slowlane_init(&sl, 2, 2)); + slowlane_fini(&sl); /* no submits — must drain/join cleanly */ + TEST_PASS(); +} + +void test_slowlane(void) +{ + RUN_TEST(test_submit_is_non_blocking); + RUN_TEST(test_admission_cap_bounds_concurrency); + RUN_TEST(test_submit_returns_eagain_when_cap_full); + RUN_TEST(test_dedup_returns_eexist_for_inflight_key); + RUN_TEST(test_dedup_key_reusable_after_completion); + RUN_TEST(test_fn_runs_exactly_once); + RUN_TEST(test_fini_on_idle_lane); +} diff --git a/src/include/xce/cluster_def.h b/src/include/xce/cluster_def.h index cbb6d71..5397d63 100644 --- a/src/include/xce/cluster_def.h +++ b/src/include/xce/cluster_def.h @@ -20,6 +20,7 @@ /* ha.h forward-declares cluster_t; including it here is safe (no circular dep) */ #include "commit.h" #include "ha.h" +#include "slowlane.h" #include "xce/types.h" #include "xlist.h" #include "zarray.h" @@ -123,6 +124,7 @@ typedef struct cluster_s { persist_t persist; /* Persistence module — P0-F */ ha_state_t ha; /* HA role management — P1-5 */ commit_engine_t commit; /* Per-entry commit engine — ha-commit-engine 5.3 */ + slowlane_t slowlane; /* Long-blocking-IO offload lane — raft-drive-async 1.2 */ _Atomic int restart_requested; /* set by OP_upgradeConfig handler */ pthread_t heartbeat_listener_tid; /* pre-Phase-2 SSL listener thread */ _Atomic int heartbeat_listener_running;