diff --git a/.kiro/specs/raft-drive-async/tasks.md b/.kiro/specs/raft-drive-async/tasks.md index 589e3bac..c36840f2 100644 --- a/.kiro/specs/raft-drive-async/tasks.md +++ b/.kiro/specs/raft-drive-async/tasks.md @@ -39,7 +39,7 @@ Tasks mirror the design's 4-phase migration. The slow lane is foundational (ever - _Boundary: BoundedConnect_ - _Depends: 1.3_ -- [ ] 1.6 LivenessAdmission — pool-level task-class reservation in vtaskmgr +- [x] 1.6 LivenessAdmission — pool-level task-class reservation in vtaskmgr - Add `vtask_class_t` (`VTASK_LIVENESS` / `VTASK_LIVE_BLOCK` / `VTASK_REQ_BLOCK`) + a classified submit; the dispatcher reserves the `VTASK_LIVENESS_RESERVED` value **resolved in 1.1** (single source of truth — consume it, do not recompute the clamp/WARN here) and fast-rejects a `VTASK_REQ_BLOCK` with `-XCE_ERR_NOT_READY` when admitting it would breach the reservation (never parks a worker, never silently queues). - Fold `commit.c`'s `inflight_waits`/`COMMIT_WAIT_RESERVED` into this — `commit_submit` submits a `VTASK_REQ_BLOCK`; the pool owns the single reservation. - Dormant infra in Phase 1: no behavior change until peer_ctrl dispatches through it (3.1). **Hard prerequisite of 3.1** — without it, moving ack/vote handling onto the pool is not deadlock-safe (vote-fsync/config-ack waits, invisible to the old commit-only counter, could starve the ack that wakes commit-waiters). @@ -48,7 +48,7 @@ Tasks mirror the design's 4-phase migration. The slow lane is foundational (ever - _Boundary: LivenessAdmission_ - _Depends: 1.1_ -- [ ] 1.7 (P) LivenessAdmission unit test +- [x] 1.7 (P) LivenessAdmission unit test - Assert: a `VTASK_LIVENESS` task is dispatched while `*_BLOCK` tasks hold `n_vcpus − VTASK_LIVENESS_RESERVED` vcpus; `VTASK_REQ_BLOCK` over the cap returns `-XCE_ERR_NOT_READY` without parking; folding commit-wait through the new path preserves the existing reject-at-cap behavior. Reverting the reservation makes a liveness task starve behind blockers. - _Requirements: 3.1, 3.2, 3.3, 4.5_ - _Boundary: LivenessAdmission_ @@ -137,3 +137,4 @@ Tasks mirror the design's 4-phase migration. The slow lane is foundational (ever - **Build/test env**: builds MUST run in the nix dev shell (`nix develop --command meson compile -C src/xce.build `). The IDE clang reports false errors (missing openssl/unity headers, unknown `-fvar-tracking`) because it isn't in the nix shell — ignore those. Controller tests: `rm -rf /tmp/xce_*` first (bug-924 overlayfs), run `./src/xce.build/controller/controller_unit_test`, set `ASAN_OPTIONS=detect_leaks=0`. - **Pre-existing leak (not ours)**: the default-options full controller suite exits non-zero due to a LeakSanitizer leak in `test_commit_barrier_reply.c` / `test_worker_join.c` / `test_egress_filter.c` (~174KB). Unrelated to this spec. Run with `detect_leaks=0` (or a `-t` filter) so it doesn't mask raft-drive-async regressions. - **1.1**: `drive_async_liveness_reserved(n_peers, n_vcpus) = max(1, min(n_peers−1, n_vcpus−1))` in `ha.c` is the single source of truth for `VTASK_LIVENESS_RESERVED` (floor applied LAST). Consumed by 1.6. `myconf_t` CLI override fields for the new `DEFAULT_*` constants were deferred to the consumer tasks (1.2 connect-timeout, 1.3/1.6) per YAGNI — add each field with its reader. +- **1.6**: pool reservation lives in vtaskmgr (`vtask_class_t` + `blocking_inflight`/`liveness_reserved` + `vtaskmgr_blocking_acquire/release`); commit folds in via opt-in `commit_set_pool` (kept `commit_engine_init(n_vcpus)` signature — the `n_vcpus` local cap is the genuine pool==NULL fallback for unit fixtures). **Production wiring is in `cluster.c` phase2** (once-guarded, right after `commit_engine_init`): `vtaskmgr_set_liveness_reserved(R)` + `commit_set_pool` — DO NOT remove; without it the reservation is dead. Lock order `p->lock → ce->lock → blk_lock` (cycle-free). New-cap = `n_vcpus−R` (tighter than old `n_vcpus−2` for 5/7-node — intended). LOW deferred: `vtaskmgr_blocking_cap()` reads `liveness_reserved` without `blk_lock` (init-time set-once, benign). When 3.1 dispatches peer_ctrl, REQ_BLOCK slot is released in `run_task` — ensure no double-count vs the commit path. diff --git a/src/controller/cluster.c b/src/controller/cluster.c index 83ed2114..1e1d5dde 100644 --- a/src/controller/cluster.c +++ b/src/controller/cluster.c @@ -1353,6 +1353,15 @@ int cluster_init_phase2(cluster_t* c) "commit/REQ_BLOCK cap=%d)", liveness_reserved, c->ha.n_peers, c->vtaskmgr.n_vcpus, c->vtaskmgr.n_vcpus - liveness_reserved); + + /* raft-drive-async task 1.6: install the pool-level LivenessAdmission + * reservation and fold commit-wait admission onto it. The pool now owns + * the single reservation; commit_submit(WAIT) acquires a pool blocking + * slot (cap = n_vcpus - liveness_reserved) instead of the local + * COMMIT_WAIT_RESERVED cap. Dormant beyond the commit cap-value change + * until peer_ctrl dispatches classified tasks (task 3.1). */ + vtaskmgr_set_liveness_reserved(&c->vtaskmgr, liveness_reserved); + commit_set_pool(&c->commit, &c->vtaskmgr); if (c->vtaskmgr.n_vcpus < c->ha.n_peers) { /* Under-provisioned: the n_vcpus ceiling clamped the reservation * below the ideal n_peers-1. Still correct (deadlock-free via fast diff --git a/src/controller/commit.c b/src/controller/commit.c index 1d718b10..721dbfcf 100644 --- a/src/controller/commit.c +++ b/src/controller/commit.c @@ -191,10 +191,11 @@ static void commit_advance_locked(commit_engine_t* ce) pthread_cond_broadcast(&ce->cond); } -/* Admission cap for concurrent blocking waits, expressed only relative to - * the n_vcpus init argument: keep COMMIT_WAIT_RESERVED vtasks free of - * blocking commit waits, floored at 1 so one wait is always admittable - * (Req 7.1, 7.2). */ +/* Admission cap for concurrent blocking waits, LOCAL (no-pool) fallback: + * keep COMMIT_WAIT_RESERVED vtasks free of blocking commit waits, floored at 1 + * so one wait is always admittable (Req 7.1, 7.2). Used only when no vtaskmgr + * pool is wired (commit_set_pool); when a pool is wired the pool owns the + * single reservation (cap = n_vcpus - VTASK_LIVENESS_RESERVED). */ static int wait_cap(const commit_engine_t* ce) { int cap = ce->n_vcpus - COMMIT_WAIT_RESERVED; @@ -202,6 +203,40 @@ static int wait_cap(const commit_engine_t* ce) return (cap < 1) ? 1 : cap; } +/* raft-drive-async task 1.6: acquire a blocking-commit-wait admission slot. + * When a pool is wired, delegate to the pool's single LivenessAdmission + * reservation (the pool owns the cap); otherwise apply the local cap. Returns + * 0 (admitted) or -XCE_ERR_NOT_READY (cap exhausted — fast reject, never + * blocks). Caller holds ce->lock. inflight_waits is maintained in BOTH modes + * as the observable in-flight commit-wait count. */ +static int wait_admit_locked(commit_engine_t* ce) +{ + if (ce->pool != NULL) { + int rc = vtaskmgr_blocking_acquire(ce->pool); + if (rc != 0) { + return rc; /* -E_XCE_NOT_READY == -XCE_ERR_NOT_READY */ + } + ce->inflight_waits++; + return 0; + } + + /* Local fallback: cap = max(1, n_vcpus - COMMIT_WAIT_RESERVED). */ + if (ce->inflight_waits >= wait_cap(ce)) { + return -XCE_ERR_NOT_READY; + } + ce->inflight_waits++; + return 0; +} + +/* Release a blocking-commit-wait admission slot. Caller holds ce->lock. */ +static void wait_release_locked(commit_engine_t* ce) +{ + ce->inflight_waits--; + if (ce->pool != NULL) { + vtaskmgr_blocking_release(ce->pool); + } +} + /* WAIT completion predicate (Req 1.1): own entry committed AND the * own-term gate (when set) reached. Caller holds ce->lock. */ static int wait_done_locked(const commit_engine_t* ce, uint64_t lsn) @@ -270,6 +305,7 @@ int commit_engine_init(commit_engine_t* ce, persist_t* p, int n_peers, int n_vcp ce->p = p; ce->n_peers = n_peers; ce->n_vcpus = n_vcpus; + ce->pool = NULL; /* local cap until commit_set_pool wires the pool */ ce->inflight_waits = 0; ce->writable_lsn = 0; /* no own-term gate until set by promotion */ ce->raft_commit_index = 0; /* advisory record only; does not gate commit (optA) */ @@ -283,6 +319,20 @@ int commit_engine_init(commit_engine_t* ce, persist_t* p, int n_peers, int n_vcp return 0; } +/* raft-drive-async task 1.6: fold commit-wait admission onto the vtaskmgr + * pool's single LivenessAdmission reservation. Set-once at wiring time; safe + * to set NULL to restore the local fallback. The pool must outlive the engine + * (controller lifecycle guarantees this — vtaskmgr_fini after commit_engine_fini). */ +void commit_set_pool(commit_engine_t* ce, vtaskmgr_t* pool) +{ + if (ce == NULL) { + return; + } + pthread_mutex_lock(&ce->lock); + ce->pool = pool; + pthread_mutex_unlock(&ce->lock); +} + void commit_engine_fini(commit_engine_t* ce) { if (ce == NULL) { @@ -311,7 +361,7 @@ int commit_submit(commit_engine_t* ce, uint64_t lsn, commit_mode_t mode, struct timespec dl; uint64_t cur, gate; int rc = 0; - int cap, wrc; + int wrc, arc; if (ce == NULL || ce->p == NULL || lsn == 0) { return -EINVAL; @@ -333,19 +383,19 @@ int commit_submit(commit_engine_t* ce, uint64_t lsn, commit_mode_t mode, /* Admission check: cap exhaustion is the ONLY -XCE_ERR_NOT_READY * cause (Req 7.1, 7.2; CORRECTED contract). Retriable fast-reject — - * never blocks. */ - cap = wait_cap(ce); - if (ce->inflight_waits >= cap) { + * never blocks. When a vtaskmgr pool is wired (raft-drive-async task 1.6) + * the cap is the pool's single LivenessAdmission reservation; otherwise + * the local COMMIT_WAIT_RESERVED cap. */ + arc = wait_admit_locked(ce); + if (arc != 0) { pthread_mutex_unlock(&ce->lock); WARN("commit: WAIT rejected, admission cap exhausted " - "(cap=%d n_vcpus=%d lsn=%lu)", - cap, ce->n_vcpus, lsn); - return -XCE_ERR_NOT_READY; + "(pool=%s n_vcpus=%d lsn=%lu)", + ce->pool ? "pool" : "local", ce->n_vcpus, lsn); + return arc; } - - ce->inflight_waits++; #ifdef XCE_TESTING - assert(ce->inflight_waits > 0 && ce->inflight_waits <= cap); + assert(ce->inflight_waits > 0); #endif /* Standalone / single-member: self constitutes the quorum (Req 6.3). @@ -367,7 +417,7 @@ int commit_submit(commit_engine_t* ce, uint64_t lsn, commit_mode_t mode, } } - ce->inflight_waits--; + wait_release_locked(ce); #ifdef XCE_TESTING assert(ce->inflight_waits >= 0); #endif diff --git a/src/controller/include/commit.h b/src/controller/include/commit.h index 91550d8e..69700b0b 100644 --- a/src/controller/include/commit.h +++ b/src/controller/include/commit.h @@ -12,8 +12,13 @@ #include #include "persist.h" /* persist_t (owns committed_lsn atomic + replicator_t registry) */ +#include "vtaskmgr.h" /* vtaskmgr_t — LivenessAdmission blocking-admission seam */ -/* Headroom kept free of blocking commit waits; cap = max(1, n_vcpus - reserved) */ +/* Headroom kept free of blocking commit waits; cap = max(1, n_vcpus - reserved). + * Used ONLY by the local (no-pool) admission fallback. When a vtaskmgr pool is + * wired (commit_set_pool, raft-drive-async task 1.6), the pool owns the single + * reservation (cap = max(1, n_vcpus - VTASK_LIVENESS_RESERVED)) and this + * constant is not consulted. */ #define COMMIT_WAIT_RESERVED 2 typedef enum { COMMIT_MODE_WAIT, COMMIT_MODE_RETAIN } commit_mode_t; @@ -29,7 +34,13 @@ typedef struct commit_engine { persist_t* p; /* back-reference: committed_lsn store + repl registry */ int n_peers; int n_vcpus; - int inflight_waits; /* <= max(1, n_vcpus - COMMIT_WAIT_RESERVED) */ + /* raft-drive-async task 1.6: when non-NULL, commit-wait admission is folded + * onto the vtaskmgr pool's single LivenessAdmission reservation (the pool + * owns the cap). When NULL (commit-only unit fixtures), the legacy local + * cap max(1, n_vcpus - COMMIT_WAIT_RESERVED) is used. inflight_waits is + * maintained in BOTH modes as the observable in-flight commit-wait count. */ + vtaskmgr_t* pool; + int inflight_waits; /* in-flight blocking commit waits */ uint64_t writable_lsn; /* own-term gate; 0 = no gate */ /* ha-term-safety commit-ceiling-optA: the raft library's commit_index, * recorded for OBSERVABILITY / TRACE / conformance-oracle ONLY. It does NOT @@ -48,6 +59,14 @@ typedef struct commit_engine { int commit_engine_init(commit_engine_t* ce, persist_t* p, int n_peers, int n_vcpus); void commit_engine_fini(commit_engine_t* ce); +/* raft-drive-async task 1.6: fold commit-wait admission onto the vtaskmgr + * pool's single LivenessAdmission reservation. After this, commit_submit(WAIT) + * acquires a pool blocking-admission slot (cap = n_vcpus - VTASK_LIVENESS_RESERVED) + * instead of using the local COMMIT_WAIT_RESERVED cap. pool == NULL restores + * the local fallback. Call once at wiring time (cluster_init_phase2), after the + * pool's liveness reservation is set. */ +void commit_set_pool(commit_engine_t* ce, vtaskmgr_t* pool); + /* Register entry at append time. RETAIN returns 0 immediately. * WAIT blocks up to timeout_ms: 0 = committed; -ETIMEDOUT = not committed * (entry remains in log; outcome resolves per umbrella 2.5); -XCE_ERR_NOT_READY = diff --git a/src/controller/meson.build b/src/controller/meson.build index 59b77dd2..e0bd2e02 100644 --- a/src/controller/meson.build +++ b/src/controller/meson.build @@ -142,6 +142,7 @@ controller_unit_test_src = [ 'testsuite/test_term_safety_adapter.c', 'testsuite/test_drive_async_config.c', 'testsuite/test_slowlane.c', + 'testsuite/test_commit_admission_pool.c', 'testsuite/test_waldump.c', 'testsuite/waldump.c', 'testsuite/test_walcompare.c', diff --git a/src/controller/testsuite/controller_test.c b/src/controller/testsuite/controller_test.c index f3523b69..38889f27 100644 --- a/src/controller/testsuite/controller_test.c +++ b/src/controller/testsuite/controller_test.c @@ -68,6 +68,7 @@ extern void test_waldump(void); extern void test_walcompare(void); extern void test_drive_async_config(void); extern void test_slowlane(void); +extern void test_commit_admission_pool(void); void setUp(void) { return; } @@ -205,6 +206,7 @@ int main(int argc, char* argv[]) TEST(test_walcompare); TEST(test_drive_async_config); TEST(test_slowlane); + TEST(test_commit_admission_pool); return UNITY_END(); } diff --git a/src/controller/testsuite/test_commit_admission_pool.c b/src/controller/testsuite/test_commit_admission_pool.c new file mode 100644 index 00000000..8e46bac3 --- /dev/null +++ b/src/controller/testsuite/test_commit_admission_pool.c @@ -0,0 +1,217 @@ +/* + * test_commit_admission_pool.c — raft-drive-async task 1.6/1.7. + * + * The commit-wait admission cap is FOLDED onto the vtaskmgr pool's single + * LivenessAdmission reservation. Asserts: + * + * (1) with a pool wired and R = VTASK_LIVENESS_RESERVED set, commit_submit + * (WAIT) fast-rejects with -XCE_ERR_NOT_READY at the POOL cap + * (n_vcpus - R), not the old local COMMIT_WAIT_RESERVED cap. This is the + * intended cap-value change: n_peers=5 → R=4 → cap=1 (TIGHTER than the + * old n_vcpus-2=6). + * (2) commit-wait still fast-rejects (never blocks) at the cap — backpressure + * preserved. + * (3) releasing a pool slot (a waiter completing) re-admits the next wait. + * (4) no-pool fallback still uses the local COMMIT_WAIT_RESERVED cap + * (regression guard for the existing commit fixtures). + * + * Deterministic: a background thread parks in commit_submit(WAIT) on an LSN + * that cannot commit (no acks, n_peers>1) with a long timeout, holding the + * admission slot; the test polls ce->inflight_waits to confirm saturation + * before probing the next admit. No correctness-by-sleep on assertions. + * + * Requirements: 3.1, 3.2, 3.3, 4.5. + */ +#include +#include +#include +#include +#include + +#include + +#include "xce/errnox.h" + +#include "commit.h" +#include "repl.h" +#include "vtaskmgr.h" + +static persist_t g_p; +static commit_engine_t g_ce; +static vtaskmgr_t g_mgr; +static int g_eps[MAX_STANDBYS]; /* opaque repl_ep_t tokens */ + +static void pool_setup(int n_peers, int n_standbys, int n_vcpus, int reserved) +{ + char name[16]; + + memset(&g_p, 0, sizeof(g_p)); + TEST_ASSERT_EQUAL_INT(0, repl_init(&g_p.repl)); + for (int i = 0; i < n_standbys; i++) { + snprintf(name, sizeof(name), "sb%d", i); + TEST_ASSERT_EQUAL_INT( + i, repl_add_standby(&g_p.repl, &g_eps[i], name, 7000 + (uint32_t)i)); + } + /* Entry 1 appended + durable, but no acks → cannot commit (n_peers>1). */ + g_p.current_lsn = 2; + atomic_store(&g_p.committed_lsn, 0); + atomic_store(&g_p.durable_lsn, 1); + + TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&g_mgr, n_vcpus)); + vtaskmgr_set_liveness_reserved(&g_mgr, reserved); + + TEST_ASSERT_EQUAL_INT(0, commit_engine_init(&g_ce, &g_p, n_peers, n_vcpus)); + if (reserved >= 0) { + commit_set_pool(&g_ce, &g_mgr); + } +} + +static void pool_teardown(void) +{ + commit_engine_fini(&g_ce); + vtaskmgr_fini(&g_mgr); + repl_fini(&g_p.repl); +} + +/* Background waiter: parks in commit_submit(WAIT) holding a slot. */ +typedef struct { + uint64_t lsn; + uint32_t timeout_ms; + int rc; +} waiter_t; + +static void* waiter_fn(void* a) +{ + waiter_t* w = (waiter_t*)a; + w->rc = commit_submit(&g_ce, w->lsn, COMMIT_MODE_WAIT, w->timeout_ms); + return NULL; +} + +static uint64_t mono_ms(void) +{ + struct timespec ts; + clock_gettime(CLOCK_MONOTONIC, &ts); + return (uint64_t)ts.tv_sec * 1000 + ts.tv_nsec / 1000000; +} + +static int poll_inflight(int want, uint32_t deadline_ms) +{ + uint64_t t0 = mono_ms(); + int v; + + for (;;) { + pthread_mutex_lock(&g_ce.lock); + v = g_ce.inflight_waits; + pthread_mutex_unlock(&g_ce.lock); + if (v >= want || mono_ms() - t0 >= deadline_ms) { + return v; + } + usleep(500); + } +} + +/* (1)+(2): the fold uses the POOL cap (n_vcpus - R), which differs from the + * old local COMMIT_WAIT_RESERVED cap. n_vcpus=4, R=3 → POOL cap = 4-3 = 1 + * (TIGHTER than the old local n_vcpus-2 = 2). One parked waiter saturates the + * pool slot; the next commit_submit(WAIT) fast-rejects without blocking. + * (MAX_STANDBYS = 3 bounds the cluster, so a small cap is exercised here; the + * cap-value formula is what matters, not the absolute count.) */ +static void test_commit_fold_uses_pool_cap(void) +{ + const int N_VCPUS = 4; + const int RESERVED = 3; /* pool cap = 4 - 3 = 1 */ + const int POOL_CAP = N_VCPUS - RESERVED; /* 1 */ + const int OLD_LOCAL_CAP = N_VCPUS - COMMIT_WAIT_RESERVED; /* 2 — what the + old local cap would have allowed */ + pthread_t tids[1]; + waiter_t wargs[1]; + int i; + + /* Sanity: the pool cap is genuinely TIGHTER than the old local cap — this + * is the intended behavior change the fold introduces. */ + TEST_ASSERT_TRUE(POOL_CAP < OLD_LOCAL_CAP); + + pool_setup(/*n_peers=*/3, /*n_standbys=*/2, N_VCPUS, RESERVED); + TEST_ASSERT_EQUAL_INT_MESSAGE(POOL_CAP, vtaskmgr_blocking_cap(&g_mgr), + "fold must use the pool cap n_vcpus - VTASK_LIVENESS_RESERVED"); + + /* Saturate the pool cap with POOL_CAP parked waiters. */ + for (i = 0; i < POOL_CAP; i++) { + wargs[i].lsn = 1; + wargs[i].timeout_ms = 4000; /* long — held until we release via ack */ + wargs[i].rc = -9999; + TEST_ASSERT_EQUAL_INT(0, + pthread_create(&tids[i], NULL, waiter_fn, &wargs[i])); + } + TEST_ASSERT_EQUAL_INT_MESSAGE(POOL_CAP, poll_inflight(POOL_CAP, 3000), + "all POOL_CAP waiters must occupy admission slots"); + + /* The over-cap wait fast-rejects WITHOUT blocking (backpressure). Under the + * OLD local cap (2) this second wait would have been ADMITTED — proving the + * fold now governs admission. */ + uint64_t t0 = mono_ms(); + int rc = commit_submit(&g_ce, /*lsn=*/1, COMMIT_MODE_WAIT, /*to=*/4000); + uint64_t dt = mono_ms() - t0; + TEST_ASSERT_EQUAL_INT_MESSAGE(-XCE_ERR_NOT_READY, rc, + "over-pool-cap commit-wait must fast-reject with -XCE_ERR_NOT_READY"); + TEST_ASSERT_TRUE_MESSAGE(dt < 500, + "admission rejection must not block the caller"); + + /* Release the parked waiter: ack entry 1 from a majority so it commits. + * n_peers=3 → self + 1 standby = 2 > 3/2. */ + repl_on_ack(&g_p.repl, &g_eps[0], 1); + commit_on_ack(&g_ce, 0, 1); + for (i = 0; i < POOL_CAP; i++) { + pthread_join(tids[i], NULL); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, wargs[i].rc, + "admitted waiter must complete with success after majority ack"); + } + + /* (3) Slot released → a fresh wait is admitted again (commits immediately + * since entry 1 is now committed). */ + rc = commit_submit(&g_ce, /*lsn=*/1, COMMIT_MODE_WAIT, /*to=*/1000); + TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc, + "after the slot is released a fresh wait must be admitted (and commit)"); + + pool_teardown(); + printf("test_commit_fold_uses_pool_cap: PASS\n"); +} + +/* (4) No pool wired → local COMMIT_WAIT_RESERVED cap still governs. n_vcpus=2, + * COMMIT_WAIT_RESERVED=2 → local cap = max(1, 0) = 1. One parked waiter + * saturates; the second fast-rejects. Regression guard for existing fixtures. */ +static void test_commit_local_fallback_cap_preserved(void) +{ + pthread_t tid; + waiter_t warg; + + /* reserved = -1 sentinel → pool_setup skips commit_set_pool (no pool). */ + pool_setup(/*n_peers=*/3, /*n_standbys=*/2, /*n_vcpus=*/2, /*reserved=*/-1); + /* Confirm no pool is wired: the engine uses the local cap path. */ + + warg.lsn = 1; + warg.timeout_ms = 4000; + warg.rc = -9999; + TEST_ASSERT_EQUAL_INT(0, pthread_create(&tid, NULL, waiter_fn, &warg)); + TEST_ASSERT_EQUAL_INT_MESSAGE(1, poll_inflight(1, 3000), + "local-cap waiter must occupy the only slot (cap=1)"); + + int rc = commit_submit(&g_ce, /*lsn=*/1, COMMIT_MODE_WAIT, /*to=*/4000); + TEST_ASSERT_EQUAL_INT_MESSAGE(-XCE_ERR_NOT_READY, rc, + "over-local-cap commit-wait must fast-reject (cap = n_vcpus-RESERVED)"); + + /* Release: majority ack (n_peers=3 → self + 1 standby = 2 > 3/2). */ + repl_on_ack(&g_p.repl, &g_eps[0], 1); + commit_on_ack(&g_ce, 0, 1); + pthread_join(tid, NULL); + TEST_ASSERT_EQUAL_INT(0, warg.rc); + + pool_teardown(); + printf("test_commit_local_fallback_cap_preserved: PASS\n"); +} + +void test_commit_admission_pool(void) +{ + RUN_TEST(test_commit_fold_uses_pool_cap); + RUN_TEST(test_commit_local_fallback_cap_preserved); +} diff --git a/src/include/vtaskmgr.h b/src/include/vtaskmgr.h index 433f7748..482f2f13 100644 --- a/src/include/vtaskmgr.h +++ b/src/include/vtaskmgr.h @@ -3,6 +3,7 @@ #define _VTASKMGR_H_ 1 #include +#include #include #include "dataq.h" @@ -15,6 +16,23 @@ typedef int (*vtask_func_t)(void*); #define VTASK_IO 0x01 #define VTASK_GENERIC 0x02 +/* raft-drive-async task 1.6: LivenessAdmission task classes. + * + * VTASK_LIVENESS never parks — WAL-ack/heartbeat/vote receipt+send. + * NEVER admission-gated, so >= VTASK_LIVENESS_RESERVED + * workers are always free for it (deadlock-freedom). + * VTASK_LIVE_BLOCK bounded mandatory park — vote-grant term fsync (R4.6). + * Runs in the reserved headroom; not gated by the + * REQ_BLOCK cap (it is short and bounded). + * VTASK_REQ_BLOCK may park seconds — commit-wait / config-ack wait. + * Admission-gated: rejected with -E_XCE_NOT_READY when + * admitting it would breach the reservation. */ +typedef enum { + VTASK_LIVENESS = 0, + VTASK_LIVE_BLOCK, + VTASK_REQ_BLOCK +} vtask_class_t; + typedef struct { int vtype; union { @@ -29,6 +47,7 @@ typedef struct { vtask_data_t* data; int vcpu; vtask_func_t entry; + vtask_class_t cls; /* LivenessAdmission class (task 1.6); default LIVENESS */ } vtask_t; #define ITPTR_TO_VTASK(p) ((vtask_t*)(((char*)p) - offsetof(vtask_t, link))) @@ -55,6 +74,14 @@ typedef struct { cbuf_t wbuf; cbuf_t wtbuf; /* used to wake up vtaskmgr to check tasks */ dataq_t taskq; + /* LivenessAdmission (raft-drive-async task 1.6): pool-level reservation. + * liveness_reserved == 0 disables the gate (legacy: nothing gated). + * blocking_inflight counts in-flight *blocking-admission* holders + * (VTASK_REQ_BLOCK tasks + commit_submit waits folded onto this seam); + * capped at max(1, n_vcpus - liveness_reserved). */ + int liveness_reserved; + int blocking_inflight; + pthread_mutex_t blk_lock; /* guards blocking_inflight check-and-increment */ } vtaskmgr_t; extern vtask_t* vtask_alloc(); @@ -64,6 +91,34 @@ extern void vtask_fini(vtask_t*); extern int vtask_enq(vtaskmgr_t*, vtask_t*); extern int vtask_handle_msg(vtaskmgr_t*, vtask_func_t, void* arg); +/* LivenessAdmission (raft-drive-async task 1.6) — pool-level reservation. + * + * vtaskmgr_set_liveness_reserved: set R, the count of vcpus reserved for + * never-gated VTASK_LIVENESS work. The controller resolves R via + * drive_async_liveness_reserved() (the single source of truth) and wires it + * here at init. R == 0 disables the gate (legacy behavior). */ +extern void vtaskmgr_set_liveness_reserved(vtaskmgr_t*, int reserved); + +/* Effective blocking-admission cap = max(1, n_vcpus - liveness_reserved) + * (the cap is floored at 1 so a commit can always run); returns n_vcpus when + * the gate is disabled (reserved == 0). */ +extern int vtaskmgr_blocking_cap(const vtaskmgr_t*); + +/* Acquire/release a blocking-admission slot. acquire returns 0 when admitted + * or -E_XCE_NOT_READY when admitting would breach the reservation (fast + * reject — never blocks). When the gate is disabled (reserved == 0) acquire + * always succeeds and is uncounted. This is the seam commit.c folds onto: + * commit_submit acquires before its blocking wait and releases on wake. */ +extern int vtaskmgr_blocking_acquire(vtaskmgr_t*); +extern void vtaskmgr_blocking_release(vtaskmgr_t*); + +/* Classified submit. VTASK_REQ_BLOCK is admission-gated: returns + * -E_XCE_NOT_READY (task NOT enqueued — caller still owns it) when admitting + * would breach the reservation. VTASK_LIVENESS / VTASK_LIVE_BLOCK are never + * gated (always enqueued). On success the task is enqueued and owned by the + * pool. */ +extern int vtask_enq_classified(vtaskmgr_t*, vtask_t*, vtask_class_t); + extern int vtaskmgr_init(vtaskmgr_t*, int); extern int vtaskmgr_start(vtaskmgr_t*); extern void vtaskmgr_fini(vtaskmgr_t*); diff --git a/src/lib/meson.build b/src/lib/meson.build index e23c0ce8..e6eb0ec0 100644 --- a/src/lib/meson.build +++ b/src/lib/meson.build @@ -180,6 +180,7 @@ libxce_test_src = ['testsuite/test.c', 'testsuite/test_raft.c', 'testsuite/test_wal_wire.c', 'testsuite/test_ctrlstatus.c', + 'testsuite/test_vtask_admission.c', 'raft/raft_sim.c', ] diff --git a/src/lib/testsuite/test.c b/src/lib/testsuite/test.c index d3cef32a..4731c702 100644 --- a/src/lib/testsuite/test.c +++ b/src/lib/testsuite/test.c @@ -43,6 +43,7 @@ extern void test_request_send_propagation(void); extern void test_raft(void); extern void test_wal_wire(void); extern void test_ctrlstatus(void); +extern void test_vtask_admission(void); void setUp(void) { return; } @@ -155,6 +156,7 @@ int main(int argc, char* argv[]) TEST(test_raft); TEST(test_wal_wire); TEST(test_ctrlstatus); + TEST(test_vtask_admission); return UNITY_END(); } diff --git a/src/lib/testsuite/test_vtask_admission.c b/src/lib/testsuite/test_vtask_admission.c new file mode 100644 index 00000000..86d15fd9 --- /dev/null +++ b/src/lib/testsuite/test_vtask_admission.c @@ -0,0 +1,221 @@ +/* + * test_vtask_admission.c — raft-drive-async task 1.6/1.7: LivenessAdmission. + * + * Pool-level task-class reservation in vtaskmgr. Asserts: + * (a) under *_BLOCK saturation a VTASK_LIVENESS task still runs; + * (b) a VTASK_REQ_BLOCK over the cap returns -E_XCE_NOT_READY without + * blocking the caller (fast-reject, never parks a worker); + * (c) blocking-admission acquire/release (the seam commit.c folds onto) + * caps concurrent blocking at max(1, n_vcpus - reserved); + * (d) removing the reservation (reserved=0) lets liveness starve behind + * blockers (mutation intuition — the guarantee depends on the gate). + * + * Deterministic harness: blocking tasks park on a semaphore-equivalent + * (cbuf) gate that the test controls; a liveness task posts an atomic + * counter the test polls. No correctness-by-sleep on the assertion path. + * + * Requirements: 3.1, 3.2, 3.3, 4.5. + */ +#include +#include +#include +#include +#include + +#include + +#include "mylock.h" +#include "vtaskmgr.h" +#include "xce/errnox.h" + +/* ── Deterministic gating primitives ───────────────────────────────────── */ + +/* A blocking task parks until the test releases it. We use a cbuf as a + * counting gate: the task does a blocking cbuf_get (parks holding its vcpu); + * the test cbuf_puts one token per task it wants to release. */ +typedef struct { + cbuf_t* gate; /* park here (blocking get) */ + atomic_int* started; /* ++ when the task body begins (occupies vcpu) */ + atomic_int* finished; /* ++ when the task body returns */ +} block_arg_t; + +static int block_task_entry(void* p) +{ + vtask_t* t = (vtask_t*)p; + block_arg_t* arg = (block_arg_t*)t->data->u.generic; + int tok = 0; + + atomic_fetch_add(arg->started, 1); + cbuf_get(arg->gate, &tok); /* park, holding this vcpu, until released */ + atomic_fetch_add(arg->finished, 1); + return 0; +} + +typedef struct { + atomic_int* ran; +} live_arg_t; + +static int live_task_entry(void* p) +{ + vtask_t* t = (vtask_t*)p; + live_arg_t* arg = (live_arg_t*)t->data->u.generic; + + atomic_fetch_add(arg->ran, 1); + return 0; +} + +static vtask_t* make_task(vtask_func_t entry, void* generic) +{ + vtask_data_t* d = calloc(1, sizeof(*d)); + vtask_t* t = vtask_alloc(); + + d->vtype = VTASK_GENERIC; + d->u.generic = generic; + vtask_init(t, entry, d); + return t; +} + +/* Spin-wait (bounded) for an atomic to reach >= want; returns 1 on success. */ +static int wait_atomic_ge(atomic_int* a, int want, int max_ms) +{ + for (int i = 0; i < max_ms * 10; i++) { + if (atomic_load(a) >= want) { + return 1; + } + usleep(100); /* 0.1ms poll — liveness, not a correctness sleep */ + } + return atomic_load(a) >= want; +} + +/* ── Test (a)+(b): liveness runs under REQ_BLOCK saturation; over-cap + * REQ_BLOCK fast-rejects ───────────────────────────────────────────── */ + +static void test_liveness_runs_under_block_saturation(void) +{ + const int NVCPUS = 4; + const int RESERVED = 2; /* cap = n_vcpus - reserved = 2 */ + const int CAP = NVCPUS - RESERVED; + vtaskmgr_t mgr; + cbuf_t gate; + atomic_int started = 0; + atomic_int finished = 0; + atomic_int live_ran = 0; + + cbuf_init(&gate, NVCPUS + 4, sizeof(int)); + + TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&mgr, NVCPUS)); + vtaskmgr_set_liveness_reserved(&mgr, RESERVED); + TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&mgr)); + + /* Saturate the REQ_BLOCK cap: CAP blocking tasks, all parked on the gate. */ + block_arg_t barg = { .gate = &gate, .started = &started, .finished = &finished }; + for (int i = 0; i < CAP; i++) { + vtask_t* t = make_task(block_task_entry, &barg); + TEST_ASSERT_EQUAL_INT( + 0, vtask_enq_classified(&mgr, t, VTASK_REQ_BLOCK)); + } + /* All CAP blocking tasks must reach their body (occupy a vcpu) and park. */ + TEST_ASSERT_TRUE_MESSAGE(wait_atomic_ge(&started, CAP, 2000), + "blocking tasks did not all start"); + + /* (b) One more REQ_BLOCK is over the cap → fast-reject, NEVER parks. */ + vtask_t* over = make_task(block_task_entry, &barg); + int rc = vtask_enq_classified(&mgr, over, VTASK_REQ_BLOCK); + TEST_ASSERT_EQUAL_INT_MESSAGE(-E_XCE_NOT_READY, rc, + "REQ_BLOCK over cap must fast-reject with -E_XCE_NOT_READY"); + /* Rejected task was NOT enqueued — caller owns it. Free it. */ + vtask_fini(over); + free(over); + + /* (a) A LIVENESS task dispatched while *_BLOCK holds all CAP slots MUST + * still run — reserved headroom guarantees a free vcpu. */ + live_arg_t larg = { .ran = &live_ran }; + vtask_t* lt = make_task(live_task_entry, &larg); + TEST_ASSERT_EQUAL_INT(0, vtask_enq_classified(&mgr, lt, VTASK_LIVENESS)); + TEST_ASSERT_TRUE_MESSAGE(wait_atomic_ge(&live_ran, 1, 2000), + "LIVENESS task starved while *_BLOCK saturated the cap"); + + /* Release all parked blockers so shutdown is clean. */ + for (int i = 0; i < CAP; i++) { + int tok = 1; + cbuf_put(&gate, &tok); + } + TEST_ASSERT_TRUE(wait_atomic_ge(&finished, CAP, 2000)); + + vtaskmgr_fini(&mgr); + cbuf_fini(&gate); +} + +/* ── Test (c): blocking acquire/release seam caps concurrency at the cap and + * fast-rejects past it (this is the seam commit.c folds onto). ───────── */ + +static void test_blocking_acquire_release_cap(void) +{ + const int NVCPUS = 5; + const int RESERVED = 4; /* cap = 1 (n_peers=5 → R=4) */ + vtaskmgr_t mgr; + + TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&mgr, NVCPUS)); + vtaskmgr_set_liveness_reserved(&mgr, RESERVED); + + TEST_ASSERT_EQUAL_INT(1, vtaskmgr_blocking_cap(&mgr)); /* 5 - 4 = 1 */ + + /* First acquire admitted; second over the cap of 1 fast-rejects. */ + TEST_ASSERT_EQUAL_INT(0, vtaskmgr_blocking_acquire(&mgr)); + TEST_ASSERT_EQUAL_INT(-E_XCE_NOT_READY, vtaskmgr_blocking_acquire(&mgr)); + + /* Release frees the slot; next acquire admitted again. */ + vtaskmgr_blocking_release(&mgr); + TEST_ASSERT_EQUAL_INT(0, vtaskmgr_blocking_acquire(&mgr)); + vtaskmgr_blocking_release(&mgr); + + vtaskmgr_fini(&mgr); +} + +/* Cap is floored at 1 even when reserved >= n_vcpus (degenerate small-core). */ +static void test_blocking_cap_floored_at_one(void) +{ + vtaskmgr_t mgr; + + TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&mgr, 2)); + vtaskmgr_set_liveness_reserved(&mgr, 5); /* reserved > n_vcpus */ + TEST_ASSERT_EQUAL_INT(1, vtaskmgr_blocking_cap(&mgr)); + /* One acquire always admittable (commit can still run). */ + TEST_ASSERT_EQUAL_INT(0, vtaskmgr_blocking_acquire(&mgr)); + TEST_ASSERT_EQUAL_INT(-E_XCE_NOT_READY, vtaskmgr_blocking_acquire(&mgr)); + vtaskmgr_blocking_release(&mgr); + vtaskmgr_fini(&mgr); +} + +/* ── Test (d): MUTATION intuition — with reserved=0 the gate is disabled, + * so blocking is uncapped and a liveness task can be starved behind a + * full house of parked blockers. We assert the gate is OFF (acquire never + * rejects) which is exactly the unsafe pre-reservation behavior. ─────── */ + +static void test_reservation_disabled_does_not_gate(void) +{ + vtaskmgr_t mgr; + + TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&mgr, 4)); + vtaskmgr_set_liveness_reserved(&mgr, 0); /* gate disabled */ + + /* With the gate disabled, acquire NEVER rejects — blocking is uncapped, + * so nothing reserves a worker for liveness. This is the failure mode the + * reservation closes (mutation oracle): flip reserved back to >0 and the + * cap re-engages (covered by the tests above). */ + for (int i = 0; i < 100; i++) { + TEST_ASSERT_EQUAL_INT(0, vtaskmgr_blocking_acquire(&mgr)); + } + for (int i = 0; i < 100; i++) { + vtaskmgr_blocking_release(&mgr); + } + vtaskmgr_fini(&mgr); +} + +void test_vtask_admission(void) +{ + RUN_TEST(test_liveness_runs_under_block_saturation); + RUN_TEST(test_blocking_acquire_release_cap); + RUN_TEST(test_blocking_cap_floored_at_one); + RUN_TEST(test_reservation_disabled_does_not_gate); +} diff --git a/src/lib/vtaskmgr.c b/src/lib/vtaskmgr.c index abc2e47e..0565206f 100644 --- a/src/lib/vtaskmgr.c +++ b/src/lib/vtaskmgr.c @@ -3,6 +3,8 @@ #include #include +#include "xce/errnox.h" + #include "mylock.h" #include "netio.h" #include "utils.h" @@ -24,6 +26,7 @@ int vtask_init(vtask_t* task, vtask_func_t entry, void* data) task->vcpu = -1; task->data = data; task->owner = NULL; + task->cls = VTASK_LIVENESS; /* never-gated by default; legacy enq paths */ return 0; } @@ -68,6 +71,14 @@ void run_task(vtask_t* task) } vtaskmgr = (vtaskmgr_t*)task->owner; + + /* LivenessAdmission (task 1.6): a VTASK_REQ_BLOCK task held a + * blocking-admission slot from enqueue (vtask_enq_classified) through its + * run; release it now so the next REQ_BLOCK can be admitted. */ + if (vtaskmgr != NULL && task->cls == VTASK_REQ_BLOCK) { + vtaskmgr_blocking_release(vtaskmgr); + } + if (vtaskmgr != NULL) { /* wake up dispatcher */ vtaskmgr_wakeup(vtaskmgr); @@ -168,10 +179,14 @@ int vtaskmgr_init(vtaskmgr_t* vtaskmgr, int nvcpus) return -EINVAL; } - vtaskmgr->dispatcher = 0; - vtaskmgr->n_vcpus = nvcpus; - vtaskmgr->vcpus = (vcpu_t*)calloc(nvcpus, sizeof(vcpu_t)); + vtaskmgr->dispatcher = 0; + vtaskmgr->n_vcpus = nvcpus; + vtaskmgr->liveness_reserved = 0; /* gate disabled until wired (task 1.6) */ + vtaskmgr->blocking_inflight = 0; + pthread_mutex_init(&vtaskmgr->blk_lock, NULL); + vtaskmgr->vcpus = (vcpu_t*)calloc(nvcpus, sizeof(vcpu_t)); if (vtaskmgr->vcpus == NULL) { + pthread_mutex_destroy(&vtaskmgr->blk_lock); return -ENOMEM; } for (i = 0; i < nvcpus; i++) { @@ -335,10 +350,116 @@ void vtaskmgr_fini(vtaskmgr_t* vtaskmgr) dataq_fini(&(vtaskmgr->taskq)); cbuf_fini(&(vtaskmgr->wtbuf)); cbuf_fini(&(vtaskmgr->wbuf)); + pthread_mutex_destroy(&vtaskmgr->blk_lock); return; } +/* ============================================================ + * LivenessAdmission (raft-drive-async task 1.6) + * ============================================================ */ + +void vtaskmgr_set_liveness_reserved(vtaskmgr_t* vtaskmgr, int reserved) +{ + if (vtaskmgr == NULL) { + return; + } + if (reserved < 0) { + reserved = 0; + } + pthread_mutex_lock(&vtaskmgr->blk_lock); + vtaskmgr->liveness_reserved = reserved; + pthread_mutex_unlock(&vtaskmgr->blk_lock); +} + +int vtaskmgr_blocking_cap(const vtaskmgr_t* vtaskmgr) +{ + int cap; + + if (vtaskmgr == NULL) { + return 0; + } + if (vtaskmgr->liveness_reserved <= 0) { + /* Gate disabled: blocking is uncapped (legacy behavior). */ + return vtaskmgr->n_vcpus; + } + cap = vtaskmgr->n_vcpus - vtaskmgr->liveness_reserved; + /* Floor at 1 so a commit (REQ_BLOCK) can always run, even on a degenerate + * small-core box where the reservation would otherwise drive the cap to 0 + * (design Sizing note). */ + return (cap < 1) ? 1 : cap; +} + +int vtaskmgr_blocking_acquire(vtaskmgr_t* vtaskmgr) +{ + int rc = 0; + + if (vtaskmgr == NULL) { + return -EINVAL; + } + + pthread_mutex_lock(&vtaskmgr->blk_lock); + if (vtaskmgr->liveness_reserved <= 0) { + /* Gate disabled: admit uncounted (legacy: nothing reserved). */ + pthread_mutex_unlock(&vtaskmgr->blk_lock); + return 0; + } + int cap = vtaskmgr->n_vcpus - vtaskmgr->liveness_reserved; + if (cap < 1) { + cap = 1; + } + if (vtaskmgr->blocking_inflight >= cap) { + /* Reservation would be breached — fast reject (never parks). */ + rc = -E_XCE_NOT_READY; + } else { + vtaskmgr->blocking_inflight++; + } + pthread_mutex_unlock(&vtaskmgr->blk_lock); + return rc; +} + +void vtaskmgr_blocking_release(vtaskmgr_t* vtaskmgr) +{ + if (vtaskmgr == NULL) { + return; + } + pthread_mutex_lock(&vtaskmgr->blk_lock); + if (vtaskmgr->liveness_reserved > 0 && vtaskmgr->blocking_inflight > 0) { + vtaskmgr->blocking_inflight--; + } + pthread_mutex_unlock(&vtaskmgr->blk_lock); +} + +int vtask_enq_classified(vtaskmgr_t* vtaskmgr, vtask_t* task, vtask_class_t cls) +{ + int rc; + + if (vtaskmgr == NULL || task == NULL) { + return -EINVAL; + } + + task->cls = cls; + + if (cls == VTASK_REQ_BLOCK) { + /* Admission-gated: reserve a blocking slot BEFORE enqueue. On reject + * the task is NOT enqueued — the caller still owns it. The slot is + * released in run_task when this task completes. */ + rc = vtaskmgr_blocking_acquire(vtaskmgr); + if (rc != 0) { + return rc; + } + rc = vtask_enq(vtaskmgr, task); + if (rc < 0) { + /* Enqueue failed after acquiring — give the slot back. */ + vtaskmgr_blocking_release(vtaskmgr); + } + return rc; + } + + /* VTASK_LIVENESS / VTASK_LIVE_BLOCK are never admission-gated. */ + return vtask_enq(vtaskmgr, task); +} + int vtask_enq(vtaskmgr_t* vtaskmgr, vtask_t* task) { int rc = 0;