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.
This commit is contained in:
parent
d91f446e9e
commit
3bc6fd14de
3 changed files with 113 additions and 7 deletions
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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 |
|
||||
|
|
|
|||
|
|
@ -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 <errno.h>
|
||||
#include <stdatomic.h>
|
||||
#include <string.h>
|
||||
#include <time.h>
|
||||
|
||||
#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;
|
||||
}
|
||||
|
||||
/* ============================================================
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue