From 3bc6fd14de94930bd50dd0b237b678c5352b26f2 Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 11 Jun 2026 15:24:30 +0000 Subject: [PATCH] feat(ha-commit-engine): WAIT mode, admission cap, writable_lsn gate (task 2.2) commit_submit(WAIT): n_vcpus-relative admission cap with fast retriable NOT_READY (admission-cap-only per corrected contract), CLOCK_REALTIME timedwait loop on the per-LSN + gate predicate, post-timeout predicate re-check, n_peers<=1 self-quorum through the single advancement site. Inflight counter accurate under engine lock for the full blocking interval with XCE_TESTING asserts. 16/16 group tests green unmodified; full suite at the 3-failure baseline. --- .kiro/specs/ha-commit-engine/tasks.md | 2 +- .wolf/memory.md | 5 ++ src/controller/commit.c | 113 ++++++++++++++++++++++++-- 3 files changed, 113 insertions(+), 7 deletions(-) diff --git a/.kiro/specs/ha-commit-engine/tasks.md b/.kiro/specs/ha-commit-engine/tasks.md index 2adb8497..1c1d15de 100644 --- a/.kiro/specs/ha-commit-engine/tasks.md +++ b/.kiro/specs/ha-commit-engine/tasks.md @@ -23,7 +23,7 @@ - Observable completion: new cases FAIL (red) against the task-1 engine - _Requirements: 1.1, 4.3, 4.4, 6.3, 7.1, 7.2, 7.3_ -- [ ] 2.2 Implement WAIT mode, admission counter, and writable_lsn gate to green +- [x] 2.2 Implement WAIT mode, admission counter, and writable_lsn gate to green - `commit_submit(WAIT)`: admission check (`inflight_waits >= max(1, n_vcpus - COMMIT_WAIT_RESERVED)` → `-XCE_ERR_NOT_READY`, distinct log message), `pthread_cond_timedwait` loop with per-LSN predicate (`committed_lsn >= lsn && (writable_lsn == 0 || committed_lsn >= writable_lsn)`), `-ETIMEDOUT` on deadline; `n_peers <= 1` immediate self-quorum success advancing `committed_lsn`; `commit_set_writable_lsn` stores the gate (no caller in this spec) - XCE_TESTING assert: inflight counter within cap and never negative - Observable completion: all task 2.1 tests PASS; cap is expressed only relative to the `n_vcpus` init argument (no absolute constant other than the reserved headroom) diff --git a/.wolf/memory.md b/.wolf/memory.md index 7fe6119b..795a3e06 100644 --- a/.wolf/memory.md +++ b/.wolf/memory.md @@ -880,3 +880,8 @@ | 14:47 | task 2.1 RED tests: 9 WAIT/admission/gate cases added (16 total, 7 pass / 9 RED -ENOSYS) | src/controller/testsuite/test_commit_engine.c | RED verified, ASAN clean | ~9k | | 07:56 | Edited .kiro/specs/ha-commit-engine/tasks.md | inline fix | ~20 | | 07:56 | Edited .kiro/specs/ha-commit-engine/tasks.md | 1→2 lines | ~122 | +| 07:59 | Edited src/controller/commit.c | modified mutex() | ~257 | +| 07:59 | Edited src/controller/commit.c | added 1 condition(s) | ~360 | +| 08:00 | Edited src/controller/commit.c | added 5 condition(s) | ~684 | +| 15:20 | ha-commit-engine task 2.2: implemented COMMIT_MODE_WAIT (admission cap max(1,n_vcpus-COMMIT_WAIT_RESERVED) -> -XCE_ERR_NOT_READY, CLOCK_REALTIME pthread_cond_timedwait loop with per-LSN+writable_lsn predicate, n_peers<=1 self-quorum via commit_advance_locked) | src/controller/commit.c | 16/16 test_commit_engine PASS, full suite = 3 baseline failures only (bug-411), no sanitizer output | ~28k | +| 08:24 | Edited .kiro/specs/ha-commit-engine/tasks.md | inline fix | ~22 | diff --git a/src/controller/commit.c b/src/controller/commit.c index 147367d9..a660872e 100644 --- a/src/controller/commit.c +++ b/src/controller/commit.c @@ -17,13 +17,23 @@ * beyond the engine mutex (plus the registry slot mutex, which the same * iomgr path already takes in repl_on_ack): no I/O, no condvar waits. * - * COMMIT_MODE_WAIT (blocking wait + admission cap) lands in task 2.2. + * COMMIT_MODE_WAIT (task 2.2) blocks the caller until its per-LSN + * predicate holds — own entry committed AND the own-term gate (when set) + * reached — or the deadline expires (-ETIMEDOUT). Admission control caps + * concurrent blocking waits at max(1, n_vcpus - COMMIT_WAIT_RESERVED); + * cap exhaustion is the ONLY cause of -XCE_ERR_NOT_READY (an unmet gate + * blocks within the timeout instead — CORRECTED contract, design + * §Commit Engine). n_peers <= 1 self-quorum advances committed_lsn + * through the single advancement path and completes immediately. * - * Requirements: 1.2, 2.1, 2.2, 2.3, 5.1 + * Requirements: 1.1, 1.2, 2.1, 2.2, 2.3, 4.3, 4.4, 5.1, 6.3, 7.1, 7.2, 7.3 */ #include #include #include +#include + +#include "xce/errnox.h" #include "commit.h" #include "mylock.h" @@ -140,6 +150,40 @@ 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). */ +static int wait_cap(const commit_engine_t* ce) +{ + int cap = ce->n_vcpus - COMMIT_WAIT_RESERVED; + + return (cap < 1) ? 1 : cap; +} + +/* 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) +{ + uint64_t c = atomic_load(&ce->p->committed_lsn); + + return c >= lsn && (ce->writable_lsn == 0 || c >= ce->writable_lsn); +} + +/* Absolute deadline now + timeout_ms on CLOCK_REALTIME — the condvar was + * initialized with default attrs (task 1.2), so pthread_cond_timedwait + * measures against CLOCK_REALTIME. */ +static void wait_deadline(struct timespec* dl, uint32_t timeout_ms) +{ + clock_gettime(CLOCK_REALTIME, dl); + dl->tv_sec += timeout_ms / 1000u; + dl->tv_nsec += (long)(timeout_ms % 1000u) * 1000000L; + if (dl->tv_nsec >= 1000000000L) { + dl->tv_sec += 1; + dl->tv_nsec -= 1000000000L; + } +} + /* ============================================================ * Lifecycle * ============================================================ */ @@ -177,7 +221,10 @@ void commit_engine_fini(commit_engine_t* ce) int commit_submit(commit_engine_t* ce, uint64_t lsn, commit_mode_t mode, uint32_t timeout_ms) { - (void)timeout_ms; /* RETAIN never honors the timeout; WAIT uses it (2.2) */ + struct timespec dl; + uint64_t cur, gate; + int rc = 0; + int cap, wrc; if (ce == NULL || ce->p == NULL || lsn == 0) { return -EINVAL; @@ -187,12 +234,66 @@ int commit_submit(commit_engine_t* ce, uint64_t lsn, commit_mode_t mode, /* Commit state is fully derived from the registry watermarks at * ACK time, so RETAIN registration has nothing to record per * entry: return immediately, never block; the entry commits later - * via commit_on_ack (Req 2.1). */ + * via commit_on_ack (Req 2.1). RETAIN never honors the timeout. */ return 0; } - /* COMMIT_MODE_WAIT: blocking wait + admission cap land in task 2.2. */ - return -ENOSYS; + /* COMMIT_MODE_WAIT — blocking wait with admission control. The + * deadline is absolute so spurious wakeups never extend the wait. */ + wait_deadline(&dl, timeout_ms); + + pthread_mutex_lock(&ce->lock); + + /* 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) { + 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; + } + + ce->inflight_waits++; +#ifdef XCE_TESTING + assert(ce->inflight_waits > 0 && ce->inflight_waits <= cap); +#endif + + /* Standalone / single-member: self constitutes the quorum (Req 6.3). + * Advance through the single advancement path — never a second + * committed_lsn store site — then complete via the same predicate. */ + if (ce->n_peers <= 1) { + commit_advance_locked(ce); + } + + while (!wait_done_locked(ce, lsn)) { + wrc = pthread_cond_timedwait(&ce->cond, &ce->lock, &dl); + if (wrc != 0) { + /* Deadline expired (or hard condvar error): one last + * predicate check closes the wakeup/timeout race. */ + if (!wait_done_locked(ce, lsn)) { + rc = -ETIMEDOUT; + } + break; + } + } + + ce->inflight_waits--; +#ifdef XCE_TESTING + assert(ce->inflight_waits >= 0); +#endif + cur = atomic_load(&ce->p->committed_lsn); + gate = ce->writable_lsn; + pthread_mutex_unlock(&ce->lock); + + if (rc == -ETIMEDOUT) { + WARN("commit: WAIT deadline expired without commit " + "(lsn=%lu committed_lsn=%lu writable_lsn=%lu timeout_ms=%u)", + lsn, cur, gate, timeout_ms); + } + return rc; } /* ============================================================