test(raft-apply-decoupling): genuine multi-threaded concurrency tests for commit resolve (task 1.5)
Closes out task-group 1 (the commit-engine event-driven-resolve keystone). Pure test-only: real pthreads racing the admission gate, the sweep thread, and commit_on_ack, not sequential calls dressed as concurrent. Adds the genuine two-real-thread ack-vs-sweep-thread race both task-1.3 review rounds flagged as a nice-to-have (the existing race test only proved the tie-break via single-thread white-box state manipulation). Also resolves the bug-1407 saga: task 1.4's "closed, 70/70" and this task's initial "reopened, 10/10 failing" were both correct, contradictory measurements of the identical source commit. Root cause was a stale incremental-build artifact in the shared build/ directory (ninja -C build rebuilt incrementally across 5 sequential tasks without ever a fresh meson setup) -- not a source regression. Proven by isolating each measurement in a throwaway git worktree with a fresh build. bug-1407 is genuinely closed; the incremental-build staleness gotcha is now documented in cerebrum.md for the remaining task series.
This commit is contained in:
parent
9b4b805a27
commit
2d9f1e947a
6 changed files with 586 additions and 10 deletions
|
|
@ -67,6 +67,22 @@
|
|||
* last-look-before-timeout tie-break.
|
||||
*
|
||||
* Requirements (task 1.3): 1.3, 1.4, 9.3, 9.4
|
||||
*
|
||||
* Concurrency half (task 1.5, closes task-group 1): the tests above prove
|
||||
* each detection path (majority ack, timeout, demotion, truncation) works
|
||||
* one-at-a-time, sequentially. The cases below prove the SAME mechanisms
|
||||
* hold up TOGETHER, under genuine multi-threaded concurrency: real pthreads
|
||||
* racing commit_submit_async's admission gate, and real concurrent
|
||||
* detection paths (a majority-ack-driving thread vs. the engine's own
|
||||
* dedicated sweep thread) resolving a whole burst of pending continuations
|
||||
* without any double-resolve, leak, or hang, and without the admission cap
|
||||
* ever being exceeded. The demotion- and truncation-triggered burst cases
|
||||
* live alongside their existing single/pair-continuation counterparts in
|
||||
* test_demotion_convergence.c and test_persist_truncate.c respectively,
|
||||
* reusing those files' real cluster_t / persist_t fixtures rather than
|
||||
* duplicating them here.
|
||||
*
|
||||
* Requirements (task 1.5): 9.1, 9.2, 9.3, 9.4
|
||||
*/
|
||||
#include <errno.h>
|
||||
#include <pthread.h>
|
||||
|
|
@ -795,6 +811,356 @@ static void test_commit_submit_async_registration_during_dispatch_not_lost(void)
|
|||
"PASS\n");
|
||||
}
|
||||
|
||||
/* ════════════════════════════════════════════════════════════════════════
|
||||
* Task 1.5 — concurrent in-flight commits: exactly-once resolution and
|
||||
* bounded admission under a GENUINE multi-threaded burst (real pthreads,
|
||||
* not sequential calls dressed up as concurrent).
|
||||
* ════════════════════════════════════════════════════════════════════════ */
|
||||
|
||||
/* ------------------------------------------------------------------ *
|
||||
* (14) Admission-cap burst: a genuine multi-threaded burst of *
|
||||
* concurrent registrations must never admit more than `cap` *
|
||||
* simultaneously, even under real contention on ce->lock — not *
|
||||
* just the two-sequential-call shape existing test (3) exercises. *
|
||||
* ------------------------------------------------------------------ */
|
||||
#define BURST_ADMISSION_N 16
|
||||
|
||||
typedef struct {
|
||||
uint64_t lsn;
|
||||
int rc;
|
||||
continuation_recorder_t* rec;
|
||||
} admission_racer_arg_t;
|
||||
|
||||
static void* admission_racer_thread(void* p)
|
||||
{
|
||||
admission_racer_arg_t* a = (admission_racer_arg_t*)p;
|
||||
|
||||
a->rc = commit_submit_async(&g_ce, a->lsn, recorder_fn, a->rec);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
/* Samples ce->inflight_waits at a tight interval while the racer burst is
|
||||
* in flight, recording the highest and lowest values observed — proves the
|
||||
* admission cap is never breached and inflight_waits never goes negative,
|
||||
* under REAL concurrent contention, not just at the two checkpoints a
|
||||
* sequential test would sample. */
|
||||
typedef struct {
|
||||
_Atomic int stop;
|
||||
int max_seen;
|
||||
int min_seen;
|
||||
} admission_monitor_arg_t;
|
||||
|
||||
static void* admission_monitor_thread(void* p)
|
||||
{
|
||||
admission_monitor_arg_t* m = (admission_monitor_arg_t*)p;
|
||||
|
||||
while (!atomic_load(&m->stop)) {
|
||||
pthread_mutex_lock(&g_ce.lock);
|
||||
int v = g_ce.inflight_waits;
|
||||
pthread_mutex_unlock(&g_ce.lock);
|
||||
if (v > m->max_seen) {
|
||||
m->max_seen = v;
|
||||
}
|
||||
if (v < m->min_seen) {
|
||||
m->min_seen = v;
|
||||
}
|
||||
usleep(50);
|
||||
}
|
||||
return NULL;
|
||||
}
|
||||
|
||||
static void test_commit_submit_async_concurrent_burst_respects_admission_cap(void)
|
||||
{
|
||||
continuation_recorder_t recs[BURST_ADMISSION_N] = {0};
|
||||
admission_racer_arg_t args[BURST_ADMISSION_N];
|
||||
pthread_t tids[BURST_ADMISSION_N];
|
||||
admission_monitor_arg_t mon = {0};
|
||||
pthread_t mon_tid;
|
||||
const int cap = 4; /* max(1, n_vcpus(6) - COMMIT_WAIT_RESERVED(2)) */
|
||||
|
||||
/* n_vcpus=6 -> local admission cap = 4 (same formula the existing
|
||||
* sequential test 3 exercises); this proves it holds under genuine
|
||||
* multi-threaded contention instead of two calls in program order. */
|
||||
resolve_setup(/*n_peers=*/3, /*n_standbys=*/2,
|
||||
/*appended_lsn=*/BURST_ADMISSION_N, /*n_vcpus=*/6);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(
|
||||
0, pthread_create(&mon_tid, NULL, admission_monitor_thread, &mon));
|
||||
|
||||
for (int i = 0; i < BURST_ADMISSION_N; i++) {
|
||||
args[i].lsn = (uint64_t)(i + 1);
|
||||
args[i].rc = -9999;
|
||||
args[i].rec = &recs[i];
|
||||
TEST_ASSERT_EQUAL_INT(
|
||||
0, pthread_create(&tids[i], NULL, admission_racer_thread, &args[i]));
|
||||
}
|
||||
for (int i = 0; i < BURST_ADMISSION_N; i++) {
|
||||
pthread_join(tids[i], NULL);
|
||||
}
|
||||
atomic_store(&mon.stop, 1);
|
||||
pthread_join(mon_tid, NULL);
|
||||
|
||||
int accepted = 0, rejected = 0;
|
||||
for (int i = 0; i < BURST_ADMISSION_N; i++) {
|
||||
if (args[i].rc == 0) {
|
||||
accepted++;
|
||||
} else if (args[i].rc == -XCE_ERR_NOT_READY) {
|
||||
rejected++;
|
||||
} else {
|
||||
TEST_FAIL_MESSAGE("unexpected rc from a concurrent registration");
|
||||
}
|
||||
}
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
cap, accepted,
|
||||
"exactly `cap` concurrent registrations may be admitted, no more, no "
|
||||
"fewer, even under genuine multi-threaded contention (Req 9.1/9.2)");
|
||||
TEST_ASSERT_EQUAL_INT(BURST_ADMISSION_N - cap, rejected);
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(cap, g_ce.inflight_waits,
|
||||
"inflight_waits must equal exactly the "
|
||||
"accepted count");
|
||||
TEST_ASSERT_TRUE_MESSAGE(
|
||||
mon.max_seen <= cap,
|
||||
"inflight_waits must never exceed the admission cap, even sampled "
|
||||
"mid-burst under real thread contention");
|
||||
TEST_ASSERT_TRUE_MESSAGE(mon.min_seen >= 0,
|
||||
"inflight_waits must never go negative");
|
||||
|
||||
/* Registration alone, even racing under contention, must never invoke
|
||||
* any continuation. */
|
||||
for (int i = 0; i < BURST_ADMISSION_N; i++) {
|
||||
TEST_ASSERT_EQUAL_INT(0, atomic_load(&recs[i].calls));
|
||||
}
|
||||
|
||||
resolve_teardown();
|
||||
printf("test_commit_submit_async_concurrent_burst_respects_admission_cap: "
|
||||
"PASS\n");
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ *
|
||||
* (15) Concurrent burst resolves exactly once via a MIX of the two *
|
||||
* real detection paths (majority ack + timeout sweep), racing *
|
||||
* each other on ce->lock, with dispatch bounded to a small, fixed *
|
||||
* vtaskmgr worker pool rather than growing with the burst size. *
|
||||
* ------------------------------------------------------------------ */
|
||||
|
||||
/* Decouples the continuation registry's capacity (ce_n_vcpus, sized to fit
|
||||
* the whole burst) from the vtaskmgr pool's actual OS worker-thread count
|
||||
* (pool_n_vcpus, deliberately smaller than the burst) — resolve_pool_setup
|
||||
* ties both to the same n_vcpus value, which cannot demonstrate that
|
||||
* dispatch stays bounded to a small worker set instead of growing with the
|
||||
* number of concurrently-resolving continuations (Req 9.1/9.2). */
|
||||
static void bounded_pool_setup(int n_peers, int n_standbys, uint64_t appended_lsn,
|
||||
int ce_n_vcpus, int pool_n_vcpus)
|
||||
{
|
||||
resolve_setup(n_peers, n_standbys, appended_lsn, ce_n_vcpus);
|
||||
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&g_mgr, pool_n_vcpus));
|
||||
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&g_mgr));
|
||||
commit_set_pool(&g_ce, &g_mgr);
|
||||
}
|
||||
|
||||
#define BURST_MIX_N 8
|
||||
|
||||
typedef struct {
|
||||
uint64_t lsn;
|
||||
int rc;
|
||||
continuation_recorder_t* rec;
|
||||
} mix_register_arg_t;
|
||||
|
||||
static void* mix_register_thread(void* p)
|
||||
{
|
||||
mix_register_arg_t* a = (mix_register_arg_t*)p;
|
||||
|
||||
a->rc = commit_submit_async(&g_ce, a->lsn, recorder_fn, a->rec);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
/* Count pairwise-distinct pthread_t values in threads[0..n) — proves
|
||||
* dispatch is bounded to a small worker set rather than one thread per
|
||||
* resolved continuation (pthread_t is opaque; must compare via
|
||||
* pthread_equal, never ==). */
|
||||
static int count_distinct_threads(const pthread_t* threads, int n)
|
||||
{
|
||||
int distinct = 0;
|
||||
|
||||
for (int i = 0; i < n; i++) {
|
||||
int seen_before = 0;
|
||||
|
||||
for (int j = 0; j < i; j++) {
|
||||
if (pthread_equal(threads[i], threads[j])) {
|
||||
seen_before = 1;
|
||||
break;
|
||||
}
|
||||
}
|
||||
if (!seen_before) {
|
||||
distinct++;
|
||||
}
|
||||
}
|
||||
return distinct;
|
||||
}
|
||||
|
||||
static void test_commit_resolve_concurrent_burst_majority_and_timeout_exactly_once(void)
|
||||
{
|
||||
continuation_recorder_t recs[BURST_MIX_N] = {0};
|
||||
mix_register_arg_t args[BURST_MIX_N];
|
||||
pthread_t tids[BURST_MIX_N];
|
||||
pthread_t dispatch_threads[BURST_MIX_N];
|
||||
|
||||
/* ce_n_vcpus == BURST_MIX_N: the continuation registry has exactly
|
||||
* enough slots for the whole burst, none to spare. pool_n_vcpus=3: a
|
||||
* small, fixed, REAL vtaskmgr worker pool that must service all 8
|
||||
* resolutions — if dispatch spawned a thread per resolution instead of
|
||||
* reusing the bounded pool, more than 3 distinct threads would appear
|
||||
* below. */
|
||||
bounded_pool_setup(/*n_peers=*/3, /*n_standbys=*/2,
|
||||
/*appended_lsn=*/BURST_MIX_N,
|
||||
/*ce_n_vcpus=*/BURST_MIX_N, /*pool_n_vcpus=*/3);
|
||||
|
||||
/* Arrange: register the whole burst CONCURRENTLY from real threads. */
|
||||
for (int i = 0; i < BURST_MIX_N; i++) {
|
||||
args[i].lsn = (uint64_t)(i + 1);
|
||||
args[i].rc = -9999;
|
||||
args[i].rec = &recs[i];
|
||||
TEST_ASSERT_EQUAL_INT(
|
||||
0, pthread_create(&tids[i], NULL, mix_register_thread, &args[i]));
|
||||
}
|
||||
for (int i = 0; i < BURST_MIX_N; i++) {
|
||||
pthread_join(tids[i], NULL);
|
||||
}
|
||||
for (int i = 0; i < BURST_MIX_N; i++) {
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
0, args[i].rc,
|
||||
"every concurrent registration must succeed -- the registry has "
|
||||
"exactly enough slots for this burst");
|
||||
}
|
||||
TEST_ASSERT_EQUAL_INT(BURST_MIX_N, g_ce.inflight_waits);
|
||||
|
||||
/* Act: satisfy majority for the LOWER half (lsn 1..4) via the REAL
|
||||
* ack-driven path (repl_on_ack + commit_on_ack, exactly what task 1.2
|
||||
* wired) -- committed_lsn advances contiguously to 4, resolving lsn
|
||||
* 1..4 as COMMITTED in one bulk commit_resolve_pending() scan. The
|
||||
* UPPER half (lsn 5..8) is never ACKed, so it is left entirely to the
|
||||
* real timeout-sweep thread (task 1.3). Both detection paths now race
|
||||
* each other for ce->lock concurrently: this ack-driving thread and the
|
||||
* commit engine's own dedicated sweep thread (which independently scans
|
||||
* the SAME registry and would classify lsn 1..4 as majority-satisfied
|
||||
* too, if its own scan happens to run before this thread's
|
||||
* commit_resolve_pending() call reaches them first -- commit_resolve's
|
||||
* idempotent slot-clear-under-lock is what makes that real race safe). */
|
||||
repl_on_ack(&g_p.repl, &g_eps[0], BURST_MIX_N / 2);
|
||||
commit_on_ack(&g_ce, 0, BURST_MIX_N / 2);
|
||||
|
||||
for (int i = 0; i < BURST_MIX_N; i++) {
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
1, poll_calls(&recs[i].calls, 1, DEFAULT_SYNC_TIMEOUT_MS + 2000),
|
||||
"every continuation in the burst must resolve, through one of "
|
||||
"the two real concurrent detection paths");
|
||||
}
|
||||
|
||||
for (int i = 0; i < BURST_MIX_N; i++) {
|
||||
/* Exactly-once: no double delivery for any lsn in the burst. */
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
1, atomic_load(&recs[i].calls),
|
||||
"no continuation in the burst may be resolved more than once");
|
||||
dispatch_threads[i] = recs[i].last_thread;
|
||||
|
||||
if (i < BURST_MIX_N / 2) {
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
COMMIT_OUTCOME_COMMITTED, recs[i].last_outcome,
|
||||
"the lower half must resolve COMMITTED via the real "
|
||||
"majority-ack path");
|
||||
TEST_ASSERT_EQUAL_INT(0, recs[i].last_rc);
|
||||
} else {
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
COMMIT_OUTCOME_FAILED, recs[i].last_outcome,
|
||||
"the upper half, never acked, must resolve FAILED via the "
|
||||
"real timeout-sweep path");
|
||||
TEST_ASSERT_EQUAL_INT(-ETIMEDOUT, recs[i].last_rc);
|
||||
}
|
||||
}
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
0, g_ce.inflight_waits,
|
||||
"every admission slot in the burst must be released exactly once");
|
||||
|
||||
/* Bounded worker usage (Req 9.1/9.2): all BURST_MIX_N resolutions must
|
||||
* be dispatched across no more than the pool's 3 real worker threads --
|
||||
* proving the burst is serviced by a small, fixed pool rather than one
|
||||
* thread spawned per resolved continuation. */
|
||||
int distinct = count_distinct_threads(dispatch_threads, BURST_MIX_N);
|
||||
TEST_ASSERT_TRUE_MESSAGE(
|
||||
distinct <= 3,
|
||||
"continuation dispatch for the whole burst must stay within the "
|
||||
"bounded vtaskmgr worker pool, not grow with the burst size");
|
||||
|
||||
resolve_pool_teardown();
|
||||
printf("test_commit_resolve_concurrent_burst_majority_and_timeout_exactly_"
|
||||
"once: PASS\n");
|
||||
}
|
||||
|
||||
/* ------------------------------------------------------------------ *
|
||||
* (16, nice-to-have) Genuine two-thread race: a real second thread *
|
||||
* delivers a real majority ACK right at the edge of the *
|
||||
* registration's own (unmanipulated) DEFAULT_SYNC_TIMEOUT_MS *
|
||||
* deadline, racing the commit engine's own dedicated sweep *
|
||||
* thread for ce->lock -- unlike test (12) above, which proves the *
|
||||
* same tie-break with single-thread white-box deadline *
|
||||
* manipulation, this drives it with two REAL OS threads. *
|
||||
* ------------------------------------------------------------------ */
|
||||
typedef struct {
|
||||
uint64_t lsn;
|
||||
} delayed_ack_arg_t;
|
||||
|
||||
static void* delayed_ack_thread(void* p)
|
||||
{
|
||||
delayed_ack_arg_t* a = (delayed_ack_arg_t*)p;
|
||||
|
||||
/* Sleep to just before lsn's own real deadline (150ms of margin absorbs
|
||||
* scheduler jitter under ASAN). commit_advance_locked() broadcasts
|
||||
* ce->cond on every real advance, so this ACK wakes the sweep thread
|
||||
* EARLY -- well before its own timed-out wait would naturally fire --
|
||||
* and the sweep thread's re-scan finds wait_done_locked() satisfied
|
||||
* first (majority wins, per commit_sweep_scan_locked's own check
|
||||
* order). */
|
||||
usleep((DEFAULT_SYNC_TIMEOUT_MS - 150) * 1000);
|
||||
repl_on_ack(&g_p.repl, &g_eps[0], a->lsn);
|
||||
commit_on_ack(&g_ce, 0, a->lsn);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
static void test_commit_resolve_genuine_two_thread_race_majority_wins_over_timeout(void)
|
||||
{
|
||||
continuation_recorder_t rec = {0};
|
||||
delayed_ack_arg_t arg = {.lsn = 1};
|
||||
pthread_t tid;
|
||||
|
||||
resolve_pool_setup(/*n_peers=*/3, /*n_standbys=*/2, /*appended_lsn=*/1,
|
||||
/*n_vcpus=*/4);
|
||||
TEST_ASSERT_EQUAL_INT(0, commit_submit_async(&g_ce, 1, recorder_fn, &rec));
|
||||
|
||||
TEST_ASSERT_EQUAL_INT(0, pthread_create(&tid, NULL, delayed_ack_thread, &arg));
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
1, poll_calls(&rec.calls, 1, DEFAULT_SYNC_TIMEOUT_MS + 2000),
|
||||
"the raced lsn must still resolve exactly once");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
COMMIT_OUTCOME_COMMITTED, rec.last_outcome,
|
||||
"a genuine majority ACK landing near the deadline boundary, on a "
|
||||
"real second thread racing the real sweep thread, must still win "
|
||||
"over the timeout -- exactly-once, majority-wins under REAL "
|
||||
"two-thread concurrency (not the white-box simulation in test 12)");
|
||||
TEST_ASSERT_EQUAL_INT(0, rec.last_rc);
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
1, atomic_load(&rec.calls),
|
||||
"exactly one resolution, regardless of which real thread won");
|
||||
TEST_ASSERT_EQUAL_INT(0, g_ce.inflight_waits);
|
||||
|
||||
pthread_join(tid, NULL);
|
||||
resolve_pool_teardown();
|
||||
printf("test_commit_resolve_genuine_two_thread_race_majority_wins_over_"
|
||||
"timeout: PASS\n");
|
||||
}
|
||||
|
||||
void test_commit_resolve(void)
|
||||
{
|
||||
RUN_TEST(test_commit_submit_async_returns_immediately_before_commit);
|
||||
|
|
@ -812,4 +1178,7 @@ void test_commit_resolve(void)
|
|||
RUN_TEST(test_commit_submit_async_resolves_failed_on_timeout);
|
||||
RUN_TEST(test_commit_submit_async_majority_wins_race_at_expired_deadline);
|
||||
RUN_TEST(test_commit_submit_async_registration_during_dispatch_not_lost);
|
||||
RUN_TEST(test_commit_submit_async_concurrent_burst_respects_admission_cap);
|
||||
RUN_TEST(test_commit_resolve_concurrent_burst_majority_and_timeout_exactly_once);
|
||||
RUN_TEST(test_commit_resolve_genuine_two_thread_race_majority_wins_over_timeout);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -458,6 +458,88 @@ static void test_demotion_from_primary_resolves_pending_continuations_failed(voi
|
|||
"checked\n");
|
||||
}
|
||||
|
||||
/* raft-apply-decoupling task 1.5 (Req 9.1-9.4): a CONCURRENT burst of
|
||||
* pending continuations, registered from real pthreads racing each other
|
||||
* for ce->lock, must ALL resolve exactly once (FAILED/-XCE_ERR_STANDBY)
|
||||
* through the SAME real demotion detection path the test above exercises
|
||||
* for a fixed pair -- proving commit_resolve_pending_above's
|
||||
* snapshot-then-resolve loop holds against a registry populated under
|
||||
* genuine multi-threaded contention, not just two sequential registrations. */
|
||||
#define DC_BURST_N 10
|
||||
|
||||
typedef struct {
|
||||
uint64_t lsn;
|
||||
int rc;
|
||||
dc_continuation_recorder_t* rec;
|
||||
} dc_burst_arg_t;
|
||||
|
||||
static void* dc_burst_register_thread(void* p)
|
||||
{
|
||||
dc_burst_arg_t* a = (dc_burst_arg_t*)p;
|
||||
|
||||
a->rc = commit_submit_async(&g_dc.commit, a->lsn, dc_recorder_fn, a->rec);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
static void test_demotion_concurrent_burst_resolves_all_pending_exactly_once(void)
|
||||
{
|
||||
dc_continuation_recorder_t recs[DC_BURST_N] = {0};
|
||||
dc_burst_arg_t args[DC_BURST_N];
|
||||
pthread_t tids[DC_BURST_N];
|
||||
|
||||
dc_setup("burst", /*n_peers=*/3, /*n_standbys=*/2,
|
||||
/*appended_lsn=*/DC_BURST_N + 4, /*n_vcpus=*/16);
|
||||
atomic_store(&g_dc.persist.committed_lsn, 4);
|
||||
|
||||
/* Register a burst of DC_BURST_N continuations CONCURRENTLY from real
|
||||
* threads, all for LSNs above the committed floor (4) -- majority is
|
||||
* unreachable in this fixture (standbys never ACK), so every one stays
|
||||
* genuinely pending, not already resolved, by the time demotion fires. */
|
||||
for (int i = 0; i < DC_BURST_N; i++) {
|
||||
args[i].lsn = (uint64_t)(5 + i);
|
||||
args[i].rc = -9999;
|
||||
args[i].rec = &recs[i];
|
||||
TEST_ASSERT_EQUAL_INT(
|
||||
0, pthread_create(&tids[i], NULL, dc_burst_register_thread, &args[i]));
|
||||
}
|
||||
for (int i = 0; i < DC_BURST_N; i++) {
|
||||
pthread_join(tids[i], NULL);
|
||||
}
|
||||
for (int i = 0; i < DC_BURST_N; i++) {
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
0, args[i].rc,
|
||||
"every concurrent registration ahead of the demotion must succeed");
|
||||
}
|
||||
TEST_ASSERT_EQUAL_INT(DC_BURST_N, g_dc.commit.inflight_waits);
|
||||
|
||||
/* Act: a single real demotion edge. The burst above is already fully
|
||||
* registered by the time this fires -- this proves the BULK resolve
|
||||
* path handles a registry populated by real concurrent registrants
|
||||
* correctly, not just the pre-existing two-continuation case above. */
|
||||
raft_cb_become_follower_with_cluster_for_test(&g_dc, /*term=*/9);
|
||||
|
||||
for (int i = 0; i < DC_BURST_N; i++) {
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
1, atomic_load(&recs[i].calls),
|
||||
"every continuation in the burst must resolve exactly once on "
|
||||
"demotion");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
(int)COMMIT_OUTCOME_FAILED, (int)recs[i].last_outcome,
|
||||
"demotion must resolve every pending continuation as FAILED");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(-XCE_ERR_STANDBY, recs[i].last_rc,
|
||||
"demotion must use the existing -XCE_ERR_STANDBY vocabulary");
|
||||
TEST_ASSERT_EQUAL_UINT64_MESSAGE((uint64_t)(5 + i), recs[i].last_lsn,
|
||||
"the recorder must observe the lsn it was registered for");
|
||||
}
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
0, g_dc.commit.inflight_waits,
|
||||
"every admission slot in the burst must be released exactly once");
|
||||
|
||||
dc_teardown();
|
||||
printf("test_demotion_concurrent_burst_resolves_all_pending_exactly_once: "
|
||||
"checked\n");
|
||||
}
|
||||
|
||||
/* Regression: demotion from STANDBY must NOT truncate, NOT restart, NOT disable
|
||||
* hooks beyond today's behavior — byte-equivalent to the pre-spec path. A
|
||||
* STANDBY has no registered hooks and no convergence work. */
|
||||
|
|
@ -897,6 +979,7 @@ void test_demotion_convergence(void)
|
|||
|
||||
RUN_TEST(test_demotion_from_primary_disables_hooks);
|
||||
RUN_TEST(test_demotion_from_primary_resolves_pending_continuations_failed);
|
||||
RUN_TEST(test_demotion_concurrent_burst_resolves_all_pending_exactly_once);
|
||||
RUN_TEST(test_demotion_schedules_convergence_truncate_and_restart);
|
||||
RUN_TEST(test_self_restart_failure_invokes_fatal_hook);
|
||||
RUN_TEST(test_demotion_from_standby_no_truncation_no_restart);
|
||||
|
|
|
|||
|
|
@ -10,6 +10,7 @@
|
|||
#include <errno.h>
|
||||
#include <fcntl.h>
|
||||
#include <inttypes.h>
|
||||
#include <pthread.h>
|
||||
#include <stdatomic.h>
|
||||
#include <stdio.h>
|
||||
#include <stdlib.h>
|
||||
|
|
@ -375,6 +376,118 @@ static void test_truncate_noop_does_not_resolve_pending_continuations(void)
|
|||
printf("test_truncate_noop_does_not_resolve_pending_continuations: PASS\n");
|
||||
}
|
||||
|
||||
/* ── Concurrent burst (raft-apply-decoupling task 1.5, Req 9.1-9.4) ────────
|
||||
* Many continuations, registered from real threads racing each other for
|
||||
* ce->lock and split across the truncation cutoff, must ALL resolve exactly
|
||||
* once through the SAME real persist_truncate_wal_above detection path
|
||||
* exercised above: entries above the new tail FAILED/-ENOENT, entries at/
|
||||
* below left untouched -- proving the bulk resolve holds under genuine
|
||||
* concurrent registration, not just the fixed handful the tests above use. */
|
||||
#define TRUNC_BURST_ABOVE 6
|
||||
#define TRUNC_BURST_BELOW 3
|
||||
#define TRUNC_BURST_TOTAL (TRUNC_BURST_ABOVE + TRUNC_BURST_BELOW)
|
||||
|
||||
typedef struct {
|
||||
commit_engine_t* ce;
|
||||
uint64_t lsn;
|
||||
int rc;
|
||||
trunc_continuation_recorder_t* rec;
|
||||
} trunc_burst_arg_t;
|
||||
|
||||
static void* trunc_burst_register_thread(void* p)
|
||||
{
|
||||
trunc_burst_arg_t* a = (trunc_burst_arg_t*)p;
|
||||
|
||||
a->rc = commit_submit_async(a->ce, a->lsn, trunc_recorder_fn, a->rec);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
static void test_truncate_concurrent_burst_resolves_above_cutoff_exactly_once(void)
|
||||
{
|
||||
char tmpdir[] = "/tmp/xce-trunc-burst-XXXXXX";
|
||||
TEST_ASSERT_NOT_NULL(mkdtemp(tmpdir));
|
||||
|
||||
persist_t *p = make_fresh_persist(tmpdir);
|
||||
TEST_ASSERT_NOT_NULL(p);
|
||||
|
||||
commit_engine_t ce;
|
||||
TEST_ASSERT_EQUAL_INT(0,
|
||||
commit_engine_init(&ce, p, /*n_peers=*/3, /*n_vcpus=*/TRUNC_BURST_TOTAL + 4));
|
||||
|
||||
int dummy = 0;
|
||||
for (int i = 0; i < TRUNC_BURST_TOTAL + 2; i++)
|
||||
persist_wal_sync_append(p, WAL_JOB_SUBMIT, &dummy, sizeof(dummy), NULL);
|
||||
|
||||
/* Cutoff placed so lsn 1..TRUNC_BURST_BELOW sit at/below the new tail
|
||||
* and lsn (TRUNC_BURST_BELOW+1)..TRUNC_BURST_TOTAL sit above it. */
|
||||
const uint64_t cutoff = TRUNC_BURST_BELOW;
|
||||
|
||||
trunc_continuation_recorder_t recs[TRUNC_BURST_TOTAL] = {0};
|
||||
trunc_burst_arg_t args[TRUNC_BURST_TOTAL];
|
||||
pthread_t tids[TRUNC_BURST_TOTAL];
|
||||
|
||||
/* Register the whole burst -- below-cutoff and above-cutoff LSNs alike
|
||||
* -- CONCURRENTLY from real threads. */
|
||||
for (int i = 0; i < TRUNC_BURST_TOTAL; i++) {
|
||||
args[i].ce = &ce;
|
||||
args[i].lsn = (uint64_t)(i + 1);
|
||||
args[i].rc = -9999;
|
||||
args[i].rec = &recs[i];
|
||||
TEST_ASSERT_EQUAL_INT(
|
||||
0, pthread_create(&tids[i], NULL, trunc_burst_register_thread, &args[i]));
|
||||
}
|
||||
for (int i = 0; i < TRUNC_BURST_TOTAL; i++) {
|
||||
pthread_join(tids[i], NULL);
|
||||
}
|
||||
for (int i = 0; i < TRUNC_BURST_TOTAL; i++) {
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
0, args[i].rc,
|
||||
"every concurrent registration ahead of the truncation must succeed");
|
||||
}
|
||||
TEST_ASSERT_EQUAL_INT(TRUNC_BURST_TOTAL, ce.inflight_waits);
|
||||
|
||||
/* Act: a single real truncation above cutoff. The burst above is
|
||||
* already fully registered by the time this fires -- this proves the
|
||||
* BULK resolve path handles a registry populated by real concurrent
|
||||
* registrants correctly, not just the fixed three-continuation case
|
||||
* above. */
|
||||
int rc = persist_truncate_wal_above(p, cutoff);
|
||||
TEST_ASSERT_EQUAL_INT(0, rc);
|
||||
TEST_ASSERT_EQUAL_UINT64(cutoff + 1, p->current_lsn);
|
||||
|
||||
for (int i = 0; i < TRUNC_BURST_TOTAL; i++) {
|
||||
uint64_t lsn = (uint64_t)(i + 1);
|
||||
|
||||
if (lsn > cutoff) {
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
1, atomic_load(&recs[i].calls),
|
||||
"every continuation above the new tail must resolve exactly "
|
||||
"once on truncation, even when the whole burst was "
|
||||
"registered concurrently");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
(int)COMMIT_OUTCOME_FAILED, (int)recs[i].last_outcome,
|
||||
"truncation must resolve as FAILED");
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(-ENOENT, recs[i].last_rc,
|
||||
"truncation must use the existing -ENOENT vocabulary");
|
||||
TEST_ASSERT_EQUAL_UINT64(lsn, recs[i].last_lsn);
|
||||
} else {
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
0, atomic_load(&recs[i].calls),
|
||||
"a continuation at/below the new tail must NOT be resolved "
|
||||
"by truncation, even under concurrent registration");
|
||||
}
|
||||
}
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(
|
||||
TRUNC_BURST_BELOW, ce.inflight_waits,
|
||||
"only the above-cutoff admission slots are released by truncation; "
|
||||
"the below-cutoff ones remain pending");
|
||||
|
||||
commit_engine_fini(&ce);
|
||||
teardown_persist(p);
|
||||
printf("test_truncate_concurrent_burst_resolves_above_cutoff_exactly_once: "
|
||||
"PASS\n");
|
||||
}
|
||||
|
||||
/* ── Test group entry point ────────────────────────────────────────────────── */
|
||||
|
||||
void test_persist_truncate(void)
|
||||
|
|
@ -387,5 +500,6 @@ void test_persist_truncate(void)
|
|||
RUN_TEST(test_truncate_noop_at_current_lsn);
|
||||
RUN_TEST(test_truncate_above_resolves_pending_continuations_as_failed);
|
||||
RUN_TEST(test_truncate_noop_does_not_resolve_pending_continuations);
|
||||
RUN_TEST(test_truncate_concurrent_burst_resolves_above_cutoff_exactly_once);
|
||||
UNITY_END();
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue