fix(controller): offload demotion convergence and catch-up streaming by default (bug-1450, bug-1451)
f88e4ef4 moved ha_demotion_convergence_fn's WAL truncate + self-restart
and ha_catchup_continuation_fn's WAL streaming off the delaymgr thread
(which drives heartbeats and election timeouts cluster-wide) onto the
slow lane, but left both flags OFF: flipping them ON hung pre-existing
unit fixtures that demote a primary or trigger catch-up without ever
starting c->slowlane -- the work was queued in a lane with no workers and
never ran.
- slowlane_is_started(): true while the lane has live workers.
- Both dispatchers run inline when the flag is OFF *or* the lane was never
started. Production always starts the lane, so it always offloads; the
lightweight fixtures keep their old inline behaviour, and neither
one-shot action can be silently dropped.
- g_drive_async_demotion_offload / g_drive_async_catchup_offload now
default to 1. The offload tests save and restore the previous value
instead of forcing 0.
New test_catchup_offload_without_started_lane_runs_inline (RED before: the
catch-up body never ran, the stream gate was never reached). Controller
suite 1239/0 with both offloads ON by default.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
88c4310e11
commit
7aac018c00
6 changed files with 72 additions and 17 deletions
|
|
@ -2851,8 +2851,13 @@ typedef struct {
|
||||||
* same kind of follow-up audit, out of scope for this fix (bug-1451,
|
* same kind of follow-up audit, out of scope for this fix (bug-1451,
|
||||||
* buglog.json). Compiled in ALL builds; the setter is the test seam to
|
* buglog.json). Compiled in ALL builds; the setter is the test seam to
|
||||||
* flip RED<->GREEN.
|
* flip RED<->GREEN.
|
||||||
|
*
|
||||||
|
* bug-1451 follow-up (2026-09-27): now ON by default. The fixture hang above
|
||||||
|
* is avoided by running inline whenever the slow lane was never started
|
||||||
|
* (slowlane_is_started), which is exactly those lightweight fixtures;
|
||||||
|
* production always starts the lane.
|
||||||
*/
|
*/
|
||||||
static _Atomic int g_drive_async_catchup_offload = 0;
|
static _Atomic int g_drive_async_catchup_offload = 1;
|
||||||
|
|
||||||
void ha_set_catchup_offload_for_test(int on)
|
void ha_set_catchup_offload_for_test(int on)
|
||||||
{
|
{
|
||||||
|
|
@ -2951,10 +2956,12 @@ int ha_catchup_continuation_fn(void *arg)
|
||||||
cluster_t *c = ctx->c;
|
cluster_t *c = ctx->c;
|
||||||
int peer_idx = ctx->peer_idx;
|
int peer_idx = ctx->peer_idx;
|
||||||
|
|
||||||
if (!ha_catchup_offload_enabled()) {
|
if (!ha_catchup_offload_enabled() || !slowlane_is_started(&c->slowlane)) {
|
||||||
/* Legacy inline path (flag OFF) — blocks the calling (delaymgr)
|
/* Inline path -- blocks the calling (delaymgr) thread. Taken with
|
||||||
* thread. Retained solely so the RED test can show the pre-change
|
* the flag OFF (kept so the RED test can show the pre-change stall)
|
||||||
* stall; production runs with the flag ON. */
|
* or when the slow lane was never started (lightweight fixtures):
|
||||||
|
* work submitted there would never run and the standby would never
|
||||||
|
* be caught up (bug-1451). */
|
||||||
catchup_lane_fn(ctx);
|
catchup_lane_fn(ctx);
|
||||||
return 0;
|
return 0;
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -203,7 +203,7 @@ typedef struct {
|
||||||
* delaymgr thread that also drives heartbeats/election timeouts cluster-wide
|
* delaymgr thread that also drives heartbeats/election timeouts cluster-wide
|
||||||
* never blocks on the truncation I/O or the tlog_exit()+execv() inside
|
* never blocks on the truncation I/O or the tlog_exit()+execv() inside
|
||||||
* ha_self_restart) or runs it INLINE on the calling (delaymgr) thread (OFF,
|
* ha_self_restart) or runs it INLINE on the calling (delaymgr) thread (OFF,
|
||||||
* the PRODUCTION DEFAULT for now — see below). This is the SAME established
|
* formerly the production default -- see below). This is the SAME established
|
||||||
* slow-lane pattern already used for snapshot send
|
* slow-lane pattern already used for snapshot send
|
||||||
* (ha_snapshot_send_dispatch/snapshot_lane_fn), config-push
|
* (ha_snapshot_send_dispatch/snapshot_lane_fn), config-push
|
||||||
* (ha_config_push_offloadable/configpush_lane_fn), and peer connect
|
* (ha_config_push_offloadable/configpush_lane_fn), and peer connect
|
||||||
|
|
@ -232,8 +232,13 @@ typedef struct {
|
||||||
* the cluster-wide default to ON is the same kind of follow-up audit, out of
|
* the cluster-wide default to ON is the same kind of follow-up audit, out of
|
||||||
* scope for this fix (bug-1450, buglog.json). Compiled in ALL builds; the
|
* scope for this fix (bug-1450, buglog.json). Compiled in ALL builds; the
|
||||||
* setter is the test seam to flip RED<->GREEN.
|
* setter is the test seam to flip RED<->GREEN.
|
||||||
|
*
|
||||||
|
* bug-1450 follow-up (2026-09-27): now ON by default. The fixture hang above
|
||||||
|
* is avoided by running inline whenever the slow lane was never started
|
||||||
|
* (slowlane_is_started), which is exactly those lightweight fixtures;
|
||||||
|
* production always starts the lane.
|
||||||
*/
|
*/
|
||||||
static _Atomic int g_drive_async_demotion_offload = 0;
|
static _Atomic int g_drive_async_demotion_offload = 1;
|
||||||
|
|
||||||
void ha_set_demotion_offload_for_test(int on)
|
void ha_set_demotion_offload_for_test(int on)
|
||||||
{
|
{
|
||||||
|
|
@ -335,10 +340,12 @@ int ha_demotion_convergence_fn(void *arg)
|
||||||
}
|
}
|
||||||
cluster_t *c = ctx->c;
|
cluster_t *c = ctx->c;
|
||||||
|
|
||||||
if (!ha_demotion_offload_enabled()) {
|
if (!ha_demotion_offload_enabled() || !slowlane_is_started(&c->slowlane)) {
|
||||||
/* Legacy inline path (flag OFF) — blocks the calling (delaymgr)
|
/* Inline path -- blocks the calling (delaymgr) thread. Taken with
|
||||||
* thread. Retained solely so the RED test can show the pre-change
|
* the flag OFF (kept so the RED test can show the pre-change stall)
|
||||||
* stall; production runs with the flag ON. */
|
* or when the slow lane was never started (lightweight fixtures):
|
||||||
|
* work submitted there would never run, and this one-shot safety
|
||||||
|
* action must never be dropped (bug-1450). */
|
||||||
int rc = ha_demotion_convergence_do_run(c);
|
int rc = ha_demotion_convergence_do_run(c);
|
||||||
free(ctx);
|
free(ctx);
|
||||||
return rc;
|
return rc;
|
||||||
|
|
|
||||||
|
|
@ -88,6 +88,12 @@ int slowlane_init(slowlane_t *sl, int n_workers, int max_inflight);
|
||||||
*/
|
*/
|
||||||
int slowlane_submit(slowlane_t *sl, uint64_t key, slowlane_fn_t fn, void *arg);
|
int slowlane_submit(slowlane_t *sl, uint64_t key, slowlane_fn_t fn, void *arg);
|
||||||
|
|
||||||
|
/* Non-zero once slowlane_init() has started worker threads (and before
|
||||||
|
* slowlane_fini()). A submit to a lane that was never started is queued but
|
||||||
|
* never runs; callers that must not drop work fall back to running it inline
|
||||||
|
* (bug-1450/1451). */
|
||||||
|
int slowlane_is_started(slowlane_t *sl);
|
||||||
|
|
||||||
/*
|
/*
|
||||||
* Drain the pending queue, stop and join all workers. Shutdown only.
|
* Drain the pending queue, stop and join all workers. Shutdown only.
|
||||||
* After return the lane is dormant; re-init to reuse.
|
* After return the lane is dormant; re-init to reuse.
|
||||||
|
|
|
||||||
|
|
@ -170,6 +170,18 @@ int slowlane_init(slowlane_t *sl, int n_workers, int max_inflight)
|
||||||
return 0;
|
return 0;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
int slowlane_is_started(slowlane_t *sl)
|
||||||
|
{
|
||||||
|
int started;
|
||||||
|
|
||||||
|
if (sl == NULL)
|
||||||
|
return 0;
|
||||||
|
pthread_mutex_lock(&sl->lock);
|
||||||
|
started = (sl->started > 0) && !sl->stopping;
|
||||||
|
pthread_mutex_unlock(&sl->lock);
|
||||||
|
return started;
|
||||||
|
}
|
||||||
|
|
||||||
int slowlane_submit(slowlane_t *sl, uint64_t key, slowlane_fn_t fn, void *arg)
|
int slowlane_submit(slowlane_t *sl, uint64_t key, slowlane_fn_t fn, void *arg)
|
||||||
{
|
{
|
||||||
if (sl == NULL || fn == NULL)
|
if (sl == NULL || fn == NULL)
|
||||||
|
|
|
||||||
|
|
@ -7071,8 +7071,11 @@ static int cc_poll_flag(_Atomic int *flag, int timeout_ms)
|
||||||
return atomic_load(flag);
|
return atomic_load(flag);
|
||||||
}
|
}
|
||||||
|
|
||||||
static void cc_run_offload_probe(int offload_on, const char *label)
|
static void cc_run_offload_probe_ex(int offload_on, int start_lane,
|
||||||
|
const char *label)
|
||||||
{
|
{
|
||||||
|
int saved_offload = ha_catchup_offload_enabled();
|
||||||
|
|
||||||
char tmpdir[256];
|
char tmpdir[256];
|
||||||
snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_cc_%s_XXXXXX",
|
snprintf(tmpdir, sizeof(tmpdir), "/tmp/xce_cc_%s_XXXXXX",
|
||||||
offload_on ? "on" : "off");
|
offload_on ? "on" : "off");
|
||||||
|
|
@ -7103,7 +7106,8 @@ static void cc_run_offload_probe(int offload_on, const char *label)
|
||||||
.max_promotion_retries = 5,
|
.max_promotion_retries = 5,
|
||||||
};
|
};
|
||||||
TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &hcfg, 0));
|
TEST_ASSERT_EQUAL_INT(0, ha_init(&c.ha, &hcfg, 0));
|
||||||
TEST_ASSERT_EQUAL_INT(0, slowlane_init(&c.slowlane, 2, 4));
|
if (start_lane)
|
||||||
|
TEST_ASSERT_EQUAL_INT(0, slowlane_init(&c.slowlane, 2, 4));
|
||||||
TEST_ASSERT_EQUAL_INT(0, delaymgr_init(&c.delaymgr, 5, cc_run_delay_task));
|
TEST_ASSERT_EQUAL_INT(0, delaymgr_init(&c.delaymgr, 5, cc_run_delay_task));
|
||||||
TEST_ASSERT_EQUAL_INT(0, delaymgr_start(&c.delaymgr));
|
TEST_ASSERT_EQUAL_INT(0, delaymgr_start(&c.delaymgr));
|
||||||
|
|
||||||
|
|
@ -7126,7 +7130,7 @@ static void cc_run_offload_probe(int offload_on, const char *label)
|
||||||
TEST_ASSERT_EQUAL_INT(0,
|
TEST_ASSERT_EQUAL_INT(0,
|
||||||
delay_task_ms(&c.delaymgr, 1, 1, cc_probe_fn, &probe));
|
delay_task_ms(&c.delaymgr, 1, 1, cc_probe_fn, &probe));
|
||||||
|
|
||||||
if (offload_on) {
|
if (offload_on && start_lane) {
|
||||||
TEST_ASSERT_TRUE_MESSAGE(cc_poll_flag(&probe.fired, 1000),
|
TEST_ASSERT_TRUE_MESSAGE(cc_poll_flag(&probe.fired, 1000),
|
||||||
"GREEN: with catch-up offloaded to the slow lane, the delaymgr "
|
"GREEN: with catch-up offloaded to the slow lane, the delaymgr "
|
||||||
"worker thread must be free to run the NEXT queued task "
|
"worker thread must be free to run the NEXT queued task "
|
||||||
|
|
@ -7154,11 +7158,12 @@ static void cc_run_offload_probe(int offload_on, const char *label)
|
||||||
"the probe task must eventually fire once the gate is released");
|
"the probe task must eventually fire once the gate is released");
|
||||||
|
|
||||||
ha_repl_stream_wal_hook = NULL;
|
ha_repl_stream_wal_hook = NULL;
|
||||||
ha_set_catchup_offload_for_test(0); /* restore production default */
|
ha_set_catchup_offload_for_test(saved_offload);
|
||||||
cc_gate_destroy(&g_cc_gate);
|
cc_gate_destroy(&g_cc_gate);
|
||||||
delaymgr_stop(&c.delaymgr);
|
delaymgr_stop(&c.delaymgr);
|
||||||
delaymgr_fini(&c.delaymgr);
|
delaymgr_fini(&c.delaymgr);
|
||||||
slowlane_fini(&c.slowlane);
|
if (start_lane)
|
||||||
|
slowlane_fini(&c.slowlane);
|
||||||
ha_fini(&c.ha);
|
ha_fini(&c.ha);
|
||||||
persist_stop(&c.persist);
|
persist_stop(&c.persist);
|
||||||
persist_fini(&c.persist);
|
persist_fini(&c.persist);
|
||||||
|
|
@ -7166,6 +7171,21 @@ static void cc_run_offload_probe(int offload_on, const char *label)
|
||||||
printf("%s: PASS\n", label);
|
printf("%s: PASS\n", label);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
static void cc_run_offload_probe(int offload_on, const char *label)
|
||||||
|
{
|
||||||
|
cc_run_offload_probe_ex(offload_on, /*start_lane=*/1, label);
|
||||||
|
}
|
||||||
|
|
||||||
|
/* bug-1451: with the offload ON (now the production default) and a slow
|
||||||
|
* lane that was never started -- the lightweight fixtures that made the
|
||||||
|
* offload hang when it was first flipped ON -- the catch-up body must still
|
||||||
|
* run (inline), never sit forever in a lane with no workers. */
|
||||||
|
static void test_catchup_offload_without_started_lane_runs_inline(void)
|
||||||
|
{
|
||||||
|
cc_run_offload_probe_ex(1, /*start_lane=*/0,
|
||||||
|
"test_catchup_offload_without_started_lane_runs_inline");
|
||||||
|
}
|
||||||
|
|
||||||
static void test_catchup_offload_probe_runs_promptly(void)
|
static void test_catchup_offload_probe_runs_promptly(void)
|
||||||
{
|
{
|
||||||
cc_run_offload_probe(1, "test_catchup_offload_probe_runs_promptly");
|
cc_run_offload_probe(1, "test_catchup_offload_probe_runs_promptly");
|
||||||
|
|
@ -7556,6 +7576,7 @@ void test_config_repl(void)
|
||||||
/* xce-audit-invariants controller pass, thread-role violation 2 fix
|
/* xce-audit-invariants controller pass, thread-role violation 2 fix
|
||||||
* (2026-07-18): catch-up continuation slow-lane offload */
|
* (2026-07-18): catch-up continuation slow-lane offload */
|
||||||
RUN_TEST(test_catchup_offload_probe_runs_promptly);
|
RUN_TEST(test_catchup_offload_probe_runs_promptly);
|
||||||
|
RUN_TEST(test_catchup_offload_without_started_lane_runs_inline);
|
||||||
RUN_TEST(test_catchup_inline_probe_blocks_RED);
|
RUN_TEST(test_catchup_inline_probe_blocks_RED);
|
||||||
/* bug-1476: concurrent load_primary_config() UAF/double-free race */
|
/* bug-1476: concurrent load_primary_config() UAF/double-free race */
|
||||||
RUN_TEST(test_bug1476_concurrent_load_primary_config_no_uaf_or_double_free);
|
RUN_TEST(test_bug1476_concurrent_load_primary_config_no_uaf_or_double_free);
|
||||||
|
|
|
||||||
|
|
@ -926,6 +926,8 @@ static void *dcb_run_thread(void *p)
|
||||||
|
|
||||||
static void dcb_run_offload(int offload_on, const char *label)
|
static void dcb_run_offload(int offload_on, const char *label)
|
||||||
{
|
{
|
||||||
|
int saved_offload = ha_demotion_offload_enabled();
|
||||||
|
|
||||||
dc_setup("blocking", /*n_peers=*/3, /*n_standbys=*/2,
|
dc_setup("blocking", /*n_peers=*/3, /*n_standbys=*/2,
|
||||||
/*appended_lsn=*/5, /*n_vcpus=*/8);
|
/*appended_lsn=*/5, /*n_vcpus=*/8);
|
||||||
atomic_store(&g_dc.persist.committed_lsn, 3);
|
atomic_store(&g_dc.persist.committed_lsn, 3);
|
||||||
|
|
@ -979,7 +981,7 @@ static void dcb_run_offload(int offload_on, const char *label)
|
||||||
/* Let the async completion (offload ON) finish before teardown. */
|
/* Let the async completion (offload ON) finish before teardown. */
|
||||||
dc_poll_restart_calls(1, 2000);
|
dc_poll_restart_calls(1, 2000);
|
||||||
|
|
||||||
ha_set_demotion_offload_for_test(0); /* restore production default */
|
ha_set_demotion_offload_for_test(saved_offload);
|
||||||
dcb_gate_destroy(&g_dcb_gate);
|
dcb_gate_destroy(&g_dcb_gate);
|
||||||
dc_teardown();
|
dc_teardown();
|
||||||
printf("%s: PASS\n", label);
|
printf("%s: PASS\n", label);
|
||||||
|
|
|
||||||
Loading…
Add table
Add a link
Reference in a new issue