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

feat(election-iomgr-refactor): COMMITTING path in ha_elect_evaluate_and_signal + committing/election timeout watchdogs — tasks 4.1, 4.3, 4.4

Browse source
This commit is contained in:
Rongsong Shen 2026-05-30 11:37:12 +00:00
parent 6de4b3ff0e
commit f4a6726f1b
6 changed files with 646 additions and 65 deletions
Download patch file Download diff file Expand all files Collapse all files

298
src/controller/ha.c
View file

@ -927,6 +927,46 @@ int handle_request_vote(ha_state_t *ha,
return 1;
}
/* ── Type definitions needed by ha_elect_evaluate_and_signal (Task 4.1) ─────
*
* election_timeout_ctx_t and committing_timeout_ctx_t are defined here (before
* ha_elect_evaluate_and_signal) because the COMMITTING path references them.
* vote_ep_ctx_t is defined in the iomgr vote-send section below.
*/
typedef struct {
_Atomic int cancelled;
ha_state_t *ha;
cluster_t *c;
ha_elect_ctx_t *ctx;
} election_timeout_ctx_t;
/* committing_timeout_ctx_t: owns the heap allocation; freed by the callback. */
typedef struct {
_Atomic int cancelled;
ha_state_t *ha;
cluster_t *c;
} committing_timeout_ctx_t;
/* committing_timeout_fn: forward declaration — defined after ha_elect_evaluate_and_signal */
static int committing_timeout_fn(void *arg);
/* cluster_init_phase2: defined in cluster.c. Only referenced in production builds
* (XCE_TESTING uses ha_phase2_hook instead, and cluster.c is not linked into the
* unit-test binary). */
#ifndef XCE_TESTING
extern int cluster_init_phase2(cluster_t *c);
#endif
#ifdef XCE_TESTING
/*
* ha_phase2_hook: test-only seam.
* When non-NULL, ha_elect_evaluate_and_signal() calls this INSTEAD OF the real
* cluster_init_phase2() so tests can count phase2 calls without a live cluster_t.
*/
int (*ha_phase2_hook)(void *c) = NULL;
#endif /* XCE_TESTING */
/* ── Election Evaluation (Req 2.1, 2.7, 2.8) ───────────────────────────────
*
* ha_elect_evaluate_and_signal: evaluate accumulated votes; set ctx->result
@ -947,14 +987,14 @@ int handle_request_vote(ha_state_t *ha,
*/
ctrl_role_t ha_elect_evaluate_and_signal(ha_state_t *ha, ha_elect_ctx_t *ctx)
{
(void)ha; /* ha->term used for fencing in network path; not needed here */
/* Guard: called from handle_request_vote_reply under ctx->lock; called from
* unit-test path (peers==NULL) without threads — done_posted prevents double
* sem_post in the network path if two threads both hit quorum simultaneously. */
* execution if two threads both hit quorum simultaneously. */
if (ctx->done_posted)
return (ctx->result == 1) ? CTRL_PRIMARY : CTRL_STANDBY;
ctrl_role_t role_result;
/* Fix 2 (CRITICAL): zero-peer guard — single-node shortcircuit.
* n_peers==0 means no other nodes; votes[] flexible array has 0 entries,
* so accessing votes[0] would be out-of-bounds. Self-vote (votes_granted=1)
@ -963,49 +1003,156 @@ ctrl_role_t ha_elect_evaluate_and_signal(ha_state_t *ha, ha_elect_ctx_t *ctx)
ctx->result = 1;
ctx->done_posted = 1;
sem_post(&ctx->done);
return CTRL_PRIMARY;
role_result = CTRL_PRIMARY;
goto apply_role;
}
int total_nodes = ctx->n_peers + 1; /* include self */
int quorum = (total_nodes + 1) / 2; /* ceil(total/2) */
{
int total_nodes = ctx->n_peers + 1; /* include self */
int quorum = (total_nodes + 1) / 2; /* ceil(total/2) */
/* Req 2.8: fatal gate — candidate has no config but a peer does */
uint64_t my_tag = ctx->votes[0].my_tag;
uint64_t my_lsn = ctx->votes[0].my_lsn;
if (my_tag == 0 && ctx->max_resp_tag > 0) {
ctx->result = 0;
/* Req 2.8: fatal gate — candidate has no config but a peer does */
uint64_t my_tag = ctx->votes[0].my_tag;
uint64_t my_lsn = ctx->votes[0].my_lsn;
if (my_tag == 0 && ctx->max_resp_tag > 0) {
ctx->result = 0;
ctx->done_posted = 1;
sem_post(&ctx->done);
role_result = CTRL_STANDBY;
goto apply_role;
}
/* Check quorum */
if (ctx->votes_granted < quorum) {
ctx->result = 0;
ctx->done_posted = 1;
sem_post(&ctx->done);
role_result = CTRL_STANDBY;
goto apply_role;
}
/* Req 2.7: yield gate — fresher responding peer */
if (ctx->max_resp_tag > my_tag) {
ctx->result = 0;
ctx->done_posted = 1;
sem_post(&ctx->done);
role_result = CTRL_STANDBY;
goto apply_role;
}
if (ctx->max_resp_tag == my_tag && ctx->max_resp_lsn > my_lsn) {
ctx->result = 0;
ctx->done_posted = 1;
sem_post(&ctx->done);
role_result = CTRL_STANDBY;
goto apply_role;
}
/* Quorum met and no fresher peer → PRIMARY */
ctx->result = 1;
ctx->done_posted = 1;
sem_post(&ctx->done);
return CTRL_STANDBY;
role_result = CTRL_PRIMARY;
}
/* Check quorum */
if (ctx->votes_granted < quorum) {
ctx->result = 0;
ctx->done_posted = 1;
sem_post(&ctx->done);
return CTRL_STANDBY;
apply_role:
/* ── Task 4.1: apply role and call cluster_init_phase2 ─────────────────
*
* After determining the election result, update ha->role and (on first boot)
* call cluster_init_phase2. This runs once per process lifetime — the
* phase2_done flag prevents re-entry on subsequent elections (RDHUP failover).
*
* The cluster_t pointer is carried in ctx->c (NULL in unit tests).
*
* Cancel the election timeout watchdog if we won (it would be a no-op now
* but the explicit cancel avoids a spurious second evaluate call).
*/
{
cluster_t *c = ctx->c;
int is_first_boot = !ha->phase2_done;
if (role_result == CTRL_PRIMARY) {
/* Cancel election timeout watchdog (quorum already reached) */
if (ctx->timeout_ctx) {
election_timeout_ctx_t *etc =
(election_timeout_ctx_t *)ctx->timeout_ctx;
atomic_store(&etc->cancelled, 1);
}
/* First boot: call cluster_init_phase2 to load config+data (Req 9.2) */
if (is_first_boot) {
#ifdef XCE_TESTING
/* In test mode, use hook if set; skip real phase2 (cluster.c not linked) */
if (ha_phase2_hook != NULL)
ha_phase2_hook(c);
#else
if (c != NULL)
cluster_init_phase2(c);
#endif
ha->phase2_done = 1;
}
/* Enter COMMITTING: block job traffic until majority ack (Req 4.1, 8.1) */
atomic_store_explicit(&ha->role, (int32_t)CTRL_PRIMARY_COMMITTING,
memory_order_release);
INFO("HA FSM: → CTRL_PRIMARY_COMMITTING (term=%" PRIu64
", first_boot=%d)", atomic_load_explicit(&ha->term, memory_order_acquire),
is_first_boot);
/* Reset ack counter for this committing round (Req 5.2) */
atomic_store(&ha->repl_ack_count, 0);
/* Arm committing timeout watchdog (degraded-mode fallback, Req 8.4) */
if (c) {
committing_timeout_ctx_t *ct = calloc(1, sizeof(*ct));
if (ct) {
atomic_init(&ct->cancelled, 0);
ct->ha = ha;
ct->c = c;
int rc = delay_task_ms(&c->delaymgr,
(int)ha->config.committing_timeout_ms,
1, committing_timeout_fn, ct);
if (rc < 0) {
WARN("ha_elect_evaluate_and_signal: delay_task_ms "
"for committing timeout failed: %d", rc);
free(ct);
}
}
}
} else {
/* STANDBY result */
/* Cancel election timeout watchdog — it is now a no-op but explicit
* cancellation matches spec wording and avoids a spurious lock-acquire
* if the timeout fires after STANDBY is decided. */
if (ctx->timeout_ctx) {
election_timeout_ctx_t *etc =
(election_timeout_ctx_t *)ctx->timeout_ctx;
atomic_store(&etc->cancelled, 1);
}
/* First boot: call cluster_init_phase2 so structures are ready;
* config arrives via primary's replication push (Req 9.2, 7.3) */
if (is_first_boot) {
#ifdef XCE_TESTING
/* In test mode, use hook if set; skip real phase2 (cluster.c not linked) */
if (ha_phase2_hook != NULL)
ha_phase2_hook(c);
#else
if (c != NULL)
cluster_init_phase2(c);
#endif
ha->phase2_done = 1;
}
atomic_store_explicit(&ha->role, (int32_t)CTRL_STANDBY,
memory_order_release);
INFO("HA FSM: → CTRL_STANDBY (term=%" PRIu64 ", first_boot=%d)",
atomic_load_explicit(&ha->term, memory_order_acquire),
is_first_boot);
}
}
/* Req 2.7: yield gate — fresher responding peer */
if (ctx->max_resp_tag > my_tag) {
ctx->result = 0;
ctx->done_posted = 1;
sem_post(&ctx->done);
return CTRL_STANDBY;
}
if (ctx->max_resp_tag == my_tag && ctx->max_resp_lsn > my_lsn) {
ctx->result = 0;
ctx->done_posted = 1;
sem_post(&ctx->done);
return CTRL_STANDBY;
}
/* Quorum met and no fresher peer → PRIMARY */
ctx->result = 1;
ctx->done_posted = 1;
sem_post(&ctx->done);
return CTRL_PRIMARY;
return role_result;
}
/* ── Vote Reply Handler (Req 2.2–2.6) ──────────────────────────────────────
@ -1086,10 +1233,6 @@ void handle_request_vote_reply(ha_state_t *ha,
* vote_ep_ctx_t: per-peer context stored in ep->data for each outgoing vote
* endpoint. on_vote_connected fires after the TLS handshake completes (when
* endpoint_t.ssl_handshake_done transitions to 1 in io_handler).
*
* election_timeout_ctx_t: argument for the delay_task_ms watchdog that fires
* ha_elect_evaluate_and_signal when the election deadline expires without all
* peers replying. Uses an _Atomic int cancelled flag for lock-free cancellation.
*/
typedef struct {
@ -1103,13 +1246,6 @@ typedef struct {
const char *my_name;
} vote_ep_ctx_t;
typedef struct {
_Atomic int cancelled;
ha_state_t *ha;
cluster_t *c;
ha_elect_ctx_t *ctx;
} election_timeout_ctx_t;
/* xce_request: defined in request.c, not yet in any public header */
extern int xce_request(endpoint_t *ep, xce_request_t *req);
@ -1483,6 +1619,51 @@ static int election_timeout_fn(void *arg)
return 0;
}
/* ── Task 4.1: committing_timeout_fn — delaymgr callback ───────────────────
*
* Fires ha_promote_to_primary (degraded mode) if the majority ack threshold
* has not been reached within committing_timeout_ms. Req 8.4.
* Owns the committing_timeout_ctx_t — frees it on exit.
*/
static int committing_timeout_fn(void *arg)
{
committing_timeout_ctx_t *t = (committing_timeout_ctx_t *)arg;
if (atomic_load(&t->cancelled)) {
free(t);
return 0;
}
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&t->ha->role,
memory_order_acquire);
if (role == CTRL_PRIMARY_COMMITTING) {
WARN("HA: committing timeout — majority ack not received within timeout; "
"promoting to PRIMARY (degraded mode, Req 8.4)");
if (t->c)
ha_promote_to_primary(t->c);
}
free(t);
return 0;
}
/* ── Task 4.1: ha_repl_ack_received ────────────────────────────────────────
*
* Called when a standby ACKs config+WAL catch-up (OPCode_replCatchUpDone).
* Increments repl_ack_count; when majority threshold (quorum-1 peers, since
* primary counts itself) is reached, disarms the committing timeout and calls
* ha_promote_to_primary. Req 8.2.
*/
void ha_repl_ack_received(ha_state_t *ha, cluster_t *c)
{
if (!ha) return;
int acks = atomic_fetch_add(&ha->repl_ack_count, 1) + 1;
int quorum = (ha->n_peers + 2) / 2; /* majority of total cluster */
if (acks >= quorum - 1) { /* -1: primary counts itself */
INFO("HA: majority ack received (acks=%d, quorum=%d) → CTRL_PRIMARY",
acks, quorum);
if (c)
ha_promote_to_primary(c);
}
}
/* ── Task 2.4: vote_ep_err_handler ─────────────────────────────────────────
*
* Called by iomgr when the outgoing vote endpoint encounters an error
@ -1558,28 +1739,23 @@ void ha_start_election_async(ha_state_t *ha, cluster_t *c,
ctx->max_resp_lsn = 0;
ctx->result = -1;
ctx->done_posted = 0;
ctx->c = c; /* Task 4.1: pass cluster for phase2 + committing timeout */
pthread_mutex_init(&ctx->lock, NULL);
sem_init(&ctx->done, 0, 0);
/* Store candidate own tag/lsn in votes[0] for evaluate function.
* n_peers==1 → n_other==0, so votes[] has 0 entries; use ctx fields. */
/* Store candidate own tag/lsn in ctx fields (votes[] has 0 entries
* for n_peers==0 shortcircuit — votes[0] would be out-of-bounds). */
ctx->max_resp_tag = my_config_tag;
ctx->max_resp_lsn = my_lsn;
/* Ensure votes[0] fields are accessible (ctx has 0 flexible entries;
* votes[0] would be out-of-bounds — use a separate approach for the
* evaluate function's my_tag/my_lsn reads). */
/* For single-node, quorum = ceil(1/2) = 1; votes_granted=1 → wins.
* Fix 1 (CRITICAL): enter CTRL_PRIMARY_COMMITTING, not CTRL_PRIMARY.
* With no standbys the committing timeout (task 4.4) will fire and
* promote to CTRL_PRIMARY (degraded/single-node mode, Req 8.4). */
* Task 4.1: ha_elect_evaluate_and_signal now handles COMMITTING entry,
* phase2, repl_ack_count reset, and committing timeout arming. */
INFO("HA election: outcome=PRIMARY term=%" PRIu64
" votes_granted=1 quorum=1 (single-node async)", cur_term);
INFO("HA FSM: CTRL_ELECTING → CTRL_PRIMARY_COMMITTING "
"(single-node async, term=%" PRIu64 ")", cur_term);
atomic_store_explicit(&ha->role, (int32_t)CTRL_PRIMARY_COMMITTING,
memory_order_release);
/* evaluate sets ha->role = CTRL_PRIMARY_COMMITTING and calls phase2 */
ha_elect_evaluate_and_signal(ha, ctx);
pthread_mutex_destroy(&ctx->lock);
sem_destroy(&ctx->done);
@ -1604,6 +1780,7 @@ void ha_start_election_async(ha_state_t *ha, cluster_t *c,
ctx->max_resp_lsn = 0;
ctx->result = -1;
ctx->done_posted = 0;
ctx->c = c; /* Task 4.1: pass cluster for phase2 + committing timeout */
pthread_mutex_init(&ctx->lock, NULL);
sem_init(&ctx->done, 0, 0);
@ -2053,6 +2230,7 @@ ctrl_role_t am_i_primary_quorum(ha_state_t *ha,
ctx->max_resp_lsn = 0;
ctx->result = -1;
ctx->done_posted = 0;
ctx->c = NULL; /* am_i_primary_quorum has no cluster_t */
pthread_mutex_init(&ctx->lock, NULL);
sem_init(&ctx->done, 0, 0);