wdgzyjx/myxce
1
0
Fork 0
forked from lei.chen/myxce
Code Pull requests Activity

feat(raft-apply-decoupling): fire command continuation post-apply (task 6.2)

Browse source 
Wire persist_apply_pump_run() to call persist_take_pending_apply()
immediately after each entry's apply_committed_entry() call,
invoking the registered continuation strictly after the mutation.
Adds jobsubmit_reply_success_once() (job.c) so a successful job
submission finally sends a client reply once every array element
applies -- closes the standing gap from task 3.2 where committed
submissions never got a response.

Fixes bug-1415: commit_resolve_pending() dispatches the commit
continuation and the apply-pump kick as two independent
VTASK_LIVE_BLOCK vtasks with no completion-order guarantee. On any
real multi-vcpu pool (the production default, n_vcpus=8), if the
apply pump won the race and advanced last_applied_lsn past an LSN
before its continuation registered, the registration was permanently
orphaned -- the job would be constructed and visible, but the client
would hang forever. Untested by every existing fixture in this spec
(all use pool=NULL, which trivially serializes and can't reproduce
the race).

Fixed with a self-healing check in persist_register_pending_apply():
if the LSN is already <= last_applied_lsn, invoke the callback
immediately instead of registering for later pickup. Verified by the
spec's first genuine multi-vcpu regression test (real vtaskmgr_t,
n_vcpus=4), independently RED->GREEN reproduced by two review rounds.
This commit is contained in:
Rongsong Shen 2026-07-05 00:38:02 +00:00
parent 2fa955b92e
commit 88577f4658
6 changed files with 719 additions and 25 deletions
Download patch file Download diff file Expand all files Collapse all files

357
src/controller/testsuite/test_apply_pump.c
View file

@ -52,16 +52,26 @@
* TEST_ASSERT — a failed assertion's longjmp must never skip teardown, or it
* hangs the next test in this binary (the known longjmp-past-teardown
* pathology from task 3.1's round-2 lesson, tasks.md Implementation Notes).
*
* Test 5 (bug-1415, task 6.2 round 2) is a deliberate exception to the
* "commit_set_pool is never called" note above: it is the one test in this
* file that wires a REAL multi-vcpu vtaskmgr_t pool, in its own dedicated
* fixture (tpr_setup/tpr_teardown, disjoint globals from tap_setup's) that
* leaves every other test in this file untouched. See that test's own
* header comment for why a real pool is required to reproduce bug-1415 at
* all.
*/
#include <rtti.h>
#include <capnp_c.h>
#include <errno.h>
#include <pthread.h>
#include <signal.h>
#include <stdbool.h>
#include <stdio.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <unity/unity.h>
@ -73,6 +83,7 @@
#include "jobj.h"
#include "persist.h"
#include "repl.h"
#include "vtaskmgr.h"
#include "wal_types.h"
#include "xce/cluster_def.h"
#include "xce/types.h"
@ -702,6 +713,351 @@ static void test_apply_pump_fail_racing_finish_merges_against_current_state(
"state: PASS\n");
}
/* ── 5. bug-1415 regression: late pending-apply registration self-heals ──
*
* Every test above (and every other test in this spec's test files)
* deliberately keeps ce->pool == NULL, so both commit.c's
* dispatch_continuation and persist_apply_pump_kick fall back to their
* documented inline/synchronous behavior — the continuation-registration
* path and the apply-pump walk can never actually run on two different
* threads in that mode, so the race bug-1415 describes cannot manifest
* there by construction (this is exactly why round 1's review found it
* untested by every fixture in this spec).
*
* This test wires a REAL vtaskmgr_t (n_vcpus > 1) via commit_set_pool, in
* its own dedicated fixture (tpr_setup/tpr_teardown, disjoint globals from
* tap_setup's) — commit_resolve_pending()'s two independent dispatches, the
* commit continuation (commit.c's dispatch_continuation) and the apply-pump
* kick (persist_apply_pump_kick), now run as REAL vtasks bound to separate
* idle vcpu worker threads by vtaskmgr's own dispatcher (vdispatcher_entry),
* exactly as production does (n_vcpus=8 by default, cluster_init_phase2).
* vtaskmgr.c has no completion-order guarantee between the two, so this
* test forces the specific bad interleaving deterministically instead of
* relying on natural timing luck: the test's own commit-continuation
* callback (tpr_commit_continuation, standing in for
* jobsubmit_commit_continuation — this test drives commit.c/persist.c
* directly, the actual layer bug-1415 lives in, rather than routing through
* job.c) blocks, polling persist_last_applied_lsn(), until the apply pump
* has ALREADY advanced past this entry's lsn — i.e. until
* persist_take_pending_apply() has already run (and found nothing, since
* nothing is registered yet) for this exact lsn — before ever calling
* persist_register_pending_apply(). This is a genuine barrier tied to the
* actual race condition, not a timing guess: as long as the pump eventually
* applies a durably committed entry (guaranteed here), the poll always
* terminates with the pump strictly ahead, reproducing the exact
* interleaving bug-1415 describes on every run, not just occasionally.
*
* Against the pre-fix persist_register_pending_apply(), this reproduces the
* permanent hang byte for byte: the registration links onto pending_applies
* after the walk has already passed this lsn for good (the walk visits each
* lsn exactly once), so nothing ever consumes it — the hand-off (this
* test's stand-in for jobsubmit_apply_complete / the client reply) never
* fires, and the stand-in resource (this test's stand-in for the leaked
* jobsubmit_async_ctx_t / endpoint ref) is never released. Against the
* self-healing fix, persist_register_pending_apply() detects the
* already-applied lsn and fires the hand-off immediately and synchronously
* instead of linking it, so the test passes: the hand-off fires exactly
* once, the resource is released exactly once, and no registration is left
* dangling on pending_applies. */
#define TPR_BARRIER_TIMEOUT_MS 3000
#define TPR_WAIT_TIMEOUT_MS 8000
typedef struct {
persist_t* p;
pthread_mutex_t lock;
pthread_cond_t cond;
int saw_failed_outcome;
int pump_won_race; /* 1 once confirmed the walk
* applied lsn before registration
* ran */
int apply_complete_fires; /* must end up == 1 */
uint64_t apply_complete_lsn;
void* fake_resource; /* stands in for
* jobsubmit_async_ctx_t / the
* endpoint ref -- released exactly
* once by the hand-off firing */
int resource_released; /* must end up == 1 */
} tpr_race_ctx_t;
static cluster_t g_tpr_c;
static commit_engine_t g_tpr_hce;
static vtaskmgr_t g_tpr_pool;
static int g_tpr_hep_a, g_tpr_hep_b;
static char g_tpr_tmpdir[256];
static job_submit_data_t g_tpr_seed_submit;
static job_t* tpr_seed_job(uint64_t jobid, uint32_t state)
{
xce_job_t jdef;
memset(&g_tpr_seed_submit, 0, sizeof(g_tpr_seed_submit));
g_tpr_seed_submit.queue = (char*)"q";
g_tpr_seed_submit.user = (char*)"tester";
memset(&jdef, 0, sizeof(jdef));
jdef.jobid = jobid;
jdef.state = state;
jdef.submit = &g_tpr_seed_submit;
return makeJob(g_tpr_c.jobmgr, &jdef);
}
static void tpr_setup(int n_vcpus)
{
kvmgr_params_t str_params = { .key_type = KVMGR_KEY_STRING,
.nslots = 32,
.delta = 16 };
kvmgr_params_t int_params = { .key_type = KVMGR_KEY_INT,
.nslots = 32,
.delta = 16 };
if (makeString((char*)"tpr-probe") == NULL)
xstring_table_init(64);
if (!g_tap_jobobj_ready) {
TEST_ASSERT_EQUAL_INT(0, job_obj_init(32, 32));
g_tap_jobobj_ready = 1;
}
memset(&g_tpr_c, 0, sizeof(g_tpr_c));
g_tpr_c.myhost = &g_tpr_c.myhost_info;
g_tpr_c.myhost_info.name = (char*)"ctrl-test-race";
g_tpr_c.myhost_info.key = (char*)"";
TEST_ASSERT_EQUAL_INT(0,
xce_hook_init(&g_tpr_c.hooksys, XCE_CONTROLLER_HOOK_MAX));
g_tpr_c.jobmgr = makeInstance(&xkvmgr_type, &int_params);
g_tpr_c.qmgr = makeInstance(&xkvmgr_type, &str_params);
g_tpr_c.workermgr = makeInstance(&xkvmgr_type, &str_params);
g_tpr_c.resmgr = makeInstance(&xkvmgr_type, &str_params);
TEST_ASSERT_NOT_NULL(g_tpr_c.jobmgr);
TEST_ASSERT_NOT_NULL(g_tpr_c.qmgr);
TEST_ASSERT_NOT_NULL(g_tpr_c.workermgr);
TEST_ASSERT_NOT_NULL(g_tpr_c.resmgr);
zarray_init(&g_tpr_c.sharedinst, 16, NULL);
persist_config_t cfg = persist_config_defaults();
cfg.wal_sync_ms = 5000;
cfg.sync_timeout_ms = 2000;
snprintf(g_tpr_tmpdir, sizeof(g_tpr_tmpdir), "/tmp/xce_tpr62_XXXXXX");
TEST_ASSERT_NOT_NULL(mkdtemp(g_tpr_tmpdir));
persist_set_workdir(&g_tpr_c.persist, g_tpr_tmpdir, "ctrl");
TEST_ASSERT_EQUAL_INT(0, persist_init(&g_tpr_c.persist, &cfg, &g_tpr_c));
TEST_ASSERT_EQUAL_INT(0, persist_start(&g_tpr_c.persist));
/* 3-member cluster: 2 stub standbys, engine wired, ACKs withheld until
* this test explicitly delivers one via repl_on_ack — mirrors
* tap_setup exactly, up to this point. */
TEST_ASSERT_TRUE(
repl_add_standby(&g_tpr_c.persist.repl, &g_tpr_hep_a, "sb_a", 7600) >= 0);
TEST_ASSERT_TRUE(
repl_add_standby(&g_tpr_c.persist.repl, &g_tpr_hep_b, "sb_b", 7601) >= 0);
TEST_ASSERT_EQUAL_INT(0,
commit_engine_init(&g_tpr_hce, &g_tpr_c.persist, 3, n_vcpus));
repl_set_commit_engine(&g_tpr_c.persist.repl, &g_tpr_hce);
repl_broadcast_send_hook = tap_drop_send_hook;
/* The one deliberate divergence from every other fixture in this file:
* a REAL multi-vcpu pool, started (vtaskmgr_start -- unlike
* test_commit_admission_pool.c's admission-only fixture, this test
* needs actual vtask execution on real worker threads), wired via
* commit_set_pool so commit_resolve_pending()'s two dispatches run as
* genuine concurrent vtasks instead of the inline fallback. */
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&g_tpr_pool, n_vcpus));
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&g_tpr_pool));
commit_set_pool(&g_tpr_hce, &g_tpr_pool);
}
static void tpr_teardown(void)
{
repl_broadcast_send_hook = NULL;
commit_engine_fini(&g_tpr_hce);
vtaskmgr_fini(&g_tpr_pool);
persist_stop(&g_tpr_c.persist);
persist_fini(&g_tpr_c.persist);
zarray_fini(&g_tpr_c.sharedinst);
if (g_tpr_c.resmgr) dereference(g_tpr_c.resmgr);
if (g_tpr_c.workermgr) dereference(g_tpr_c.workermgr);
if (g_tpr_c.qmgr) dereference(g_tpr_c.qmgr);
if (g_tpr_c.jobmgr) dereference(g_tpr_c.jobmgr);
xce_hook_fini(&g_tpr_c.hooksys);
}
static void tpr_wait_durable(uint64_t lsn, uint32_t deadline_ms)
{
for (uint32_t waited = 0; waited < deadline_ms; waited += 5) {
if (atomic_load(&g_tpr_c.persist.durable_lsn) >= lsn)
return;
usleep(5000);
}
}
/* Task B's hand-off target: this test's stand-in for jobsubmit_apply_complete.
* Fires exactly once when persist_register_pending_apply()'s self-healing
* check invokes it inline (the fix), or would fire never (pre-fix hang) if
* the walk had already orphaned the registration. */
static void tpr_apply_complete(void* ctx, uint64_t lsn)
{
tpr_race_ctx_t* rctx = (tpr_race_ctx_t*)ctx;
pthread_mutex_lock(&rctx->lock);
rctx->apply_complete_fires++;
rctx->apply_complete_lsn = lsn;
if (rctx->fake_resource != NULL) {
free(rctx->fake_resource);
rctx->fake_resource = NULL;
rctx->resource_released++;
}
pthread_cond_broadcast(&rctx->cond);
pthread_mutex_unlock(&rctx->lock);
}
/* Task A stand-in for jobsubmit_commit_continuation: deterministically
* forces the exact bad interleaving bug-1415 describes (see this test's
* header comment) by blocking, on this vtask's OWN worker thread, until the
* apply pump (Task B, running concurrently on a separate vcpu) has already
* applied `lsn` -- THEN registering, reproducing a genuinely late
* persist_register_pending_apply() call. */
static void tpr_commit_continuation(void* ctx, uint64_t lsn,
commit_outcome_t outcome, int rc)
{
tpr_race_ctx_t* rctx = (tpr_race_ctx_t*)ctx;
(void)rc;
if (outcome != COMMIT_OUTCOME_COMMITTED) {
pthread_mutex_lock(&rctx->lock);
rctx->saw_failed_outcome = 1;
pthread_cond_broadcast(&rctx->cond);
pthread_mutex_unlock(&rctx->lock);
return;
}
uint32_t waited_ms = 0;
while (persist_last_applied_lsn(rctx->p) < lsn &&
waited_ms < TPR_BARRIER_TIMEOUT_MS) {
usleep(2000);
waited_ms += 2;
}
pthread_mutex_lock(&rctx->lock);
rctx->pump_won_race = (persist_last_applied_lsn(rctx->p) >= lsn);
pthread_mutex_unlock(&rctx->lock);
/* The exact call bug-1415 describes as arriving too late -- the apply
* pump has already visited (and will never revisit) this lsn. */
persist_register_pending_apply(rctx->p, lsn, tpr_apply_complete, ctx);
}
static void test_apply_pump_late_pending_apply_registration_self_heals(void)
{
const int N_VCPUS = 4; /* > 1 is the whole point -- see header comment */
tpr_setup(N_VCPUS);
job_t* j = tpr_seed_job(950, JOB_STAT_PEND);
int seeded_ok = (j != NULL);
tpr_race_ctx_t rctx;
memset(&rctx, 0, sizeof(rctx));
pthread_mutex_init(&rctx.lock, NULL);
pthread_cond_init(&rctx.cond, NULL);
rctx.p = &g_tpr_c.persist;
rctx.fake_resource = malloc(64);
int resource_alloc_ok = (rctx.fake_resource != NULL);
int rc1 = persist_submit_job_fact(&g_tpr_c.persist, WAL_JOB_START, 950,
JOB_STAT_RUN);
uint64_t lsn = g_tpr_c.persist.current_lsn - 1;
/* Register the commit-outcome continuation directly against the commit
* engine -- the same mechanism jobsubmit_commit_continuation uses
* (commit_submit_async), standing in for it at this layer since this
* test's boundary is commit.c/persist.c, not job.c. */
int async_rc = commit_submit_async(&g_tpr_hce, lsn, tpr_commit_continuation,
&rctx);
tpr_wait_durable(lsn, 2000);
/* Majority ACK (self + 1 of 2 standbys = 2 of 3 peers) drives
* commit_on_ack -> commit_advance_locked -> commit_resolve_pending,
* which dispatches BOTH the continuation (Task A) and the apply-pump
* kick (Task B) as real vtasks onto the pool wired above. */
repl_on_ack(&g_tpr_c.persist.repl, &g_tpr_hep_a, lsn);
/* Bounded poll for the hand-off -- the key assertion: the client-facing
* hand-off must still fire even though the walk (Task B) won the race
* and applied lsn before the continuation (Task A) ever registered. */
struct timespec deadline;
clock_gettime(CLOCK_REALTIME, &deadline);
deadline.tv_sec += (TPR_WAIT_TIMEOUT_MS / 1000);
deadline.tv_nsec += (long)(TPR_WAIT_TIMEOUT_MS % 1000) * 1000000L;
if (deadline.tv_nsec >= 1000000000L) {
deadline.tv_sec += 1;
deadline.tv_nsec -= 1000000000L;
}
pthread_mutex_lock(&rctx.lock);
while (rctx.apply_complete_fires == 0) {
if (pthread_cond_timedwait(&rctx.cond, &rctx.lock, &deadline) != 0)
break;
}
int fires = rctx.apply_complete_fires;
int released = rctx.resource_released;
int pump_won_race = rctx.pump_won_race;
int saw_failed = rctx.saw_failed_outcome;
uint64_t fired_lsn = rctx.apply_complete_lsn;
pthread_mutex_unlock(&rctx.lock);
int has_pending_after = persist_has_pending_apply(&g_tpr_c.persist, lsn);
if (rctx.fake_resource != NULL) {
/* Only reachable pre-fix (orphaned forever, never released by
* tpr_apply_complete) -- free here so THIS test never leaks
* regardless of which way the assertions below come out. */
free(rctx.fake_resource);
rctx.fake_resource = NULL;
}
pthread_mutex_destroy(&rctx.lock);
pthread_cond_destroy(&rctx.cond);
tpr_teardown();
TEST_ASSERT_TRUE(seeded_ok);
TEST_ASSERT_TRUE(resource_alloc_ok);
TEST_ASSERT_EQUAL_INT(0, rc1);
TEST_ASSERT_EQUAL_INT(0, async_rc);
TEST_ASSERT_FALSE_MESSAGE(saw_failed,
"the entry must genuinely commit -- a FAILED outcome would mean "
"this test never exercises the race it targets");
TEST_ASSERT_TRUE_MESSAGE(pump_won_race,
"the forced barrier must confirm the apply pump actually applied "
"this lsn BEFORE the continuation registered -- otherwise this test "
"is not proving what it claims");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, fires,
"bug-1415: a persist_register_pending_apply() call that arrives "
"after the apply pump has already applied its lsn must still fire "
"its hand-off exactly once (self-healing) -- 0 means a permanent "
"hang (the orphaned-registration bug), >1 means a double-fire");
TEST_ASSERT_EQUAL_UINT64_MESSAGE(lsn, fired_lsn,
"the hand-off must fire for the correct lsn");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, released,
"the hand-off's resource (standing in for jobsubmit_async_ctx_t / "
"the endpoint ref) must be released exactly once -- no leak, no "
"double-free");
TEST_ASSERT_EQUAL_INT_MESSAGE(0, has_pending_after,
"no orphaned registration may remain linked on pending_applies "
"after the hand-off fires");
printf("test_apply_pump_late_pending_apply_registration_self_heals: "
"PASS\n");
}
/* ── Test group entry point ───────────────────────────────────────────── */
void test_apply_pump(void)
@ -711,5 +1067,6 @@ void test_apply_pump(void)
RUN_TEST(test_apply_pump_run_applies_mixed_submit_and_fact_entries);
RUN_TEST(test_apply_pump_guard_does_not_break_job_ctrl_live_append);
RUN_TEST(test_apply_pump_fail_racing_finish_merges_against_current_state);
RUN_TEST(test_apply_pump_late_pending_apply_registration_self_heals);
UNITY_END();
}