lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(raft-drive-async): LivenessAdmission pool reservation + commit fold (tasks 1.6/1.7)

Browse source 
vtaskmgr gains vtask_class_t (LIVENESS/LIVE_BLOCK/REQ_BLOCK) + a pool-level
blocking reservation: REQ_BLOCK fast-rejects -XCE_ERR_NOT_READY past
n_vcpus-R; LIVENESS never gated (deadlock-freedom). commit_submit folds its
inflight_waits onto the pool via commit_set_pool; production wired in
cluster.c phase2 (vtaskmgr_set_liveness_reserved(R) + commit_set_pool).
Lock order p->lock->ce->lock->blk_lock. Dormant until 3.1. Commit suites
38/30/22/38 green; lib 282/0, controller 196/0, ASAN clean. Phase 1 done.
This commit is contained in:
Rongsong Shen 2026-06-26 07:43:19 +00:00
parent 0781c6c027
commit a36234a103
12 changed files with 721 additions and 22 deletions
Download patch file Download diff file Expand all files Collapse all files

5
.kiro/specs/raft-drive-async/tasks.md
View file

@ -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 <target>`). 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.

9
src/controller/cluster.c
View file

@ -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

80
src/controller/commit.c
View file

@ -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

23
src/controller/include/commit.h
View file

@ -12,8 +12,13 @@
#include <stdint.h>
#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 =

1
src/controller/meson.build
View file

@ -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',

2
src/controller/testsuite/controller_test.c
View file

@ -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();
}

217
src/controller/testsuite/test_commit_admission_pool.c Normal file
View file

@ -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 <pthread.h>
#include <stdatomic.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <unity/unity.h>
#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);
}

55
src/include/vtaskmgr.h
View file

@ -3,6 +3,7 @@
#define _VTASKMGR_H_ 1
#include <pthread.h>
#include <stdatomic.h>
#include <stddef.h>
#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*);

1
src/lib/meson.build
View file

@ -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',
]

2
src/lib/testsuite/test.c
View file

@ -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();
}

221
src/lib/testsuite/test_vtask_admission.c Normal file
View file

@ -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 <pthread.h>
#include <stdatomic.h>
#include <stdint.h>
#include <time.h>
#include <unistd.h>
#include <unity/unity.h>
#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);
}

127
src/lib/vtaskmgr.c
View file

@ -3,6 +3,8 @@
#include <stdlib.h>
#include <unistd.h>
#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;