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

feat(raft-drive-async): peer-control vtask dispatch shim, dormant (task 3.1)

Browse source 
Lands the PeerCtrlDispatch machinery behind a DEFAULT-OFF flag
(g_drive_async_peerctrl_dispatch_enabled); production behavior is
UNCHANGED. peer_ctrl_msg_handler becomes a thin shim that takes
ownership of ep->fsm_in.pkt and either runs the routing body inline
(flag OFF, byte-for-byte the prior path) or dispatches it to the vtask
pool via vtask_enq_classified (flag ON). The 3 routers now take pkt as
an explicit param (no cross-thread ep->fsm_in.pkt re-read). Per-endpoint
arrival order is enforced by bind_vtask's endpoint_block_tasks.

Packet ownership: shim clears ep->fsm_in.pkt at dispatch (sole-owner
transfer); peer_ctrl_vtask_entry NULLs iod->pkt before the body frees
it once. New vtask_t.owns_io_pkt (default 0) lets taskq_item_killer free
the carried packet for an un-executed queued task at vtaskmgr_fini WITHOUT
double-freeing the libxce endpoint-owned path (bug-1199).

Adversarial review found two FLAG-ON hazards, documented with loud
comments (flag is OFF so production is unaffected), to be resolved before
task 3.4 flips the flag ON:
  - bug-1197 (CRITICAL): raft_cb_become_follower endpoint_fini()s peer
    vote-eps from a pool worker, racing the io_worker's read of ep->data;
    needs iomgr_unmonitor before endpoint_fini in every pool-reachable
    teardown.
  - bug-1198 (HIGH): peer_ctrl_classify tags all work VTASK_LIVENESS, but
    the CTRL_PRIMARY repl-ack path can run a blocking snapshot checkpoint
    inline; resolved by task 3.2 (offload snapshot to SlowLane).

Per spec, 3.1 is not 'done' until the task 3.4 deadlock gate is green.
Tests: 3 new peerctrl tests; controller 196/196, lib 282/0; 0 net leaks.
This commit is contained in:
Rongsong Shen 2026-06-26 11:12:04 +00:00
parent 21e26a17b6
commit e17a6a613c
7 changed files with 1678 additions and 31 deletions
Download patch file Download diff file Expand all files Collapse all files

1087
.wolf/buglog.json
View file

File diff suppressed because it is too large Load diff

5
src/controller/cluster.c
View file

@ -1362,6 +1362,11 @@ int cluster_init_phase2(cluster_t* c)
* until peer_ctrl dispatches classified tasks (task 3.1). */
vtaskmgr_set_liveness_reserved(&c->vtaskmgr, liveness_reserved);
commit_set_pool(&c->commit, &c->vtaskmgr);
/* raft-drive-async task 3.1: give ha the pool handle so
* peer_ctrl_msg_handler can dispatch peer-control messages onto it when
* the dispatch flag is ON. Dormant until the flag flips (default OFF);
* stored as void* (ha.h does not pull vtaskmgr.h). */
c->ha.pool = &c->vtaskmgr;
if (c->vtaskmgr.n_vcpus < c->ha.n_peers) {
/* Under-provisioned: the n_vcpus ceiling clamped the reservation
* below the ideal n_peers-1. Still correct (deadlock-free via fast

285
src/controller/ha.c
View file

@ -1263,7 +1263,14 @@ static void raft_cb_become_follower(void *ctx, uint64_t term, int leader_idx)
/* Close all outgoing vote endpoints (loser path).
* The primary will push config and ctrlhello via promoted connections;
* the standby waits passively on its listener for the inbound connection. */
* the standby waits passively on its listener for the inbound connection.
*
* FLAG-ON HAZARD (raft-drive-async finding-1; bug-1197): when
* g_drive_async_peerctrl_dispatch_enabled is ON this loop runs on a POOL
* worker and frees ep->data while the io_worker may still deliver a queued
* peer-control event on another ep → UAF. Before task 3.4 flips the flag
* ON, each endpoint_fini below MUST be preceded by iomgr_unmonitor(ep). See
* the full hazard note at peer_ctrl_dispatch_to_pool. Safe today (flag OFF). */
{
ha_elect_ctx_t *ectx = (ha_elect_ctx_t *)atomic_load_explicit(
&ha->elect_ctx, memory_order_acquire);
@ -3795,6 +3802,35 @@ int ha_reconnect_offload_enabled(void)
memory_order_acquire);
}
/* raft-drive-async task 3.1: PeerCtrlDispatch feature flag.
*
* DEFAULT OFF (0): the registered peer-control handler runs its routing body
* INLINE on the iomgr io_worker thread — byte-for-byte the pre-task behavior,
* so THIS COMMIT DOES NOT CHANGE PRODUCTION BEHAVIOR. ON (1): the handler
* hands the decoded packet to the vtask pool and returns promptly. The flag
* stays OFF until the task 3.4 deadlock gate proves the dispatch path safe.
*
* Compiled in ALL builds (production reads it on every peer-control message);
* the setter is the test seam to flip RED<->GREEN. */
static _Atomic int g_drive_async_peerctrl_dispatch_enabled = 0;
void ha_set_peerctrl_dispatch_for_test(int on)
{
atomic_store_explicit(&g_drive_async_peerctrl_dispatch_enabled, on ? 1 : 0,
memory_order_release);
}
int ha_peerctrl_dispatch_enabled(void)
{
return atomic_load_explicit(&g_drive_async_peerctrl_dispatch_enabled,
memory_order_acquire);
}
/* raft-drive-async task 3.1: test-only observation seam (plain global, NULL in
* production — zero cost, same idiom as ha_nop_req_hook). Invoked at the top
* of peer_ctrl_msg_handler_inline before any role routing. */
void (*ha_peerctrl_body_probe)(endpoint_t *ep) = NULL;
/* ── Test-only hook shims (XCE_TESTING only) ──────────────────────────────
* Type definitions (election_timeout_ctx_t), extern declarations
* (cluster_init_phase2, push_config_transport_only,
@ -4259,11 +4295,13 @@ void handle_request_vote_reply(ha_state_t *ha,
* and never reassigned — it routes by ha->role at dispatch time.
*/
static int peer_ctrl_msg_handler(void *ep_raw);
static int vote_reply_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx);
static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx);
static int vote_reply_decode_and_dispatch(vote_ep_ctx_t *vctx, io_packet_t *pkt);
static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx,
io_packet_t *pkt);
/* ha-prevote task 5.1: preVoteReply decode/route; defined alongside the
* real-vote decoder, feeds the later-defined pre_vote_reply_dispatch. */
static int pre_vote_reply_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx);
static int pre_vote_reply_decode_and_dispatch(vote_ep_ctx_t *vctx,
io_packet_t *pkt);
static void pre_vote_reply_dispatch(ha_state_t *ha, cluster_t *c,
int peer_idx, int granted);
@ -4433,22 +4471,23 @@ static void vote_reply_dispatch(ha_state_t *ha, ha_elect_ctx_t *ctx, int idx,
* vote_reply_decode_and_dispatch: decode a RequestVoteReply packet and dispatch
* to vote_reply_dispatch. Factored from vote_reply_msg_handler so that
* peer_ctrl_msg_handler can invoke the same logic without code duplication.
* Packet ownership: caller passes ep->fsm_in.pkt; this function frees it.
* ep->fsm_in.pkt is cleared HERE, before dispatch — vote_reply_dispatch can
* reach raft_cb_become_follower → endpoint_fini(ep) (loser path), which
* would otherwise double-free the packet via the stale fsm_in.pkt pointer.
* Packet ownership (raft-drive-async task 3.1): the caller
* (peer_ctrl_msg_handler) has ALREADY taken `pkt` off ep->fsm_in.pkt and
* cleared ep->fsm_in.pkt to NULL — `pkt` is passed in explicitly and this
* function frees it on every path. Clearing in the caller (not here) is what
* makes the off-iomgr dispatch path race-free: the body never touches
* ep->fsm_in.pkt, so a concurrent io_worker read on the same endpoint cannot
* collide with the in-flight packet. vote_reply_dispatch can still reach
* raft_cb_become_follower → endpoint_fini(ep) (loser path) safely because
* ep->fsm_in.pkt is already NULL.
*/
static int vote_reply_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx)
static int vote_reply_decode_and_dispatch(vote_ep_ctx_t *vctx, io_packet_t *pkt)
{
ha_state_t *ha = vctx->ha;
ha_elect_ctx_t *ctx = vctx->ctx;
int idx = vctx->peer_idx;
io_packet_t *pkt = ep->fsm_in.pkt;
int ok = 0;
ep->fsm_in.pkt = NULL; /* take ownership (same pattern as
* repl_ack_decode_and_dispatch) */
uint64_t vr_term = 0;
int vr_granted = 0;
uint64_t vr_tag = 0;
@ -4527,20 +4566,19 @@ done:
* real-vote tally, and deliberately performs NO term adoption (a PreVote reply
* never changes our term; Req 1.2). vctx->peer_idx is the ABSOLUTE cluster
* index that raft_on_pre_vote_reply expects (set in ha_send_pre_vote_request).
* Packet ownership: takes ep->fsm_in.pkt and frees it.
* Packet ownership (raft-drive-async task 3.1): `pkt` is passed in explicitly
* (already taken off ep->fsm_in.pkt by peer_ctrl_msg_handler); this function
* frees it on every path and never touches ep->fsm_in.pkt.
*/
static int pre_vote_reply_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx)
static int pre_vote_reply_decode_and_dispatch(vote_ep_ctx_t *vctx,
io_packet_t *pkt)
{
ha_state_t *ha = vctx->ha;
cluster_t *c = vctx->c;
int idx = vctx->peer_idx; /* ABSOLUTE cluster index */
io_packet_t *pkt = ep->fsm_in.pkt;
int ok = 0;
int vr_granted = 0;
ep->fsm_in.pkt = NULL; /* take ownership (same pattern as
* vote_reply_decode_and_dispatch) */
if (!pkt || !pkt->data || pkt->sz == 0)
goto done;
@ -4603,12 +4641,21 @@ done:
* - OPCode_nop or OPCode_reply with any other body → silently ignored
* - anything else → WARN + drop
*/
static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx)
static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx,
io_packet_t *pkt)
{
ha_state_t *ha = vctx->ha;
cluster_t *c = vctx->c;
io_packet_t *pkt = ep->fsm_in.pkt;
/* Packet ownership (raft-drive-async task 3.1): `pkt` is passed in
* explicitly (peer_ctrl_msg_handler already cleared ep->fsm_in.pkt); this
* function frees it and never touches ep->fsm_in.pkt.
*
* raft-drive-async tasks 3.2/3.3 (FUTURE): the long ops reachable from the
* stateReport branch below — ha_dispatch_repl_action_for_standby's
* snapshot send / config push — are where the SlowLane offload lands. For
* task 3.1 the body only relocates onto the pool and still runs bounded
* decode/route work inline in the vtask entry. */
if (!pkt || !pkt->data || pkt->sz == 0) {
if (pkt) { free_packet(pkt); free(pkt); }
return 0;
@ -4621,7 +4668,6 @@ static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx)
(size_t)pkt->sz, 0);
free_packet(pkt);
free(pkt);
ep->fsm_in.pkt = NULL;
if (capn_ok != 0)
return 0;
@ -4795,13 +4841,24 @@ static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx)
}
/*
* peer_ctrl_msg_handler: state-aware iomgr handler set once in on_vote_connected.
* peer_ctrl_msg_handler_inline: the state-aware peer-control routing body.
* Routes by ha->role at dispatch time — no explicit handler reassignment needed.
* CTRL_ELECTING → vote_reply_decode_and_dispatch
* CTRL_PRIMARY → repl_ack_decode_and_dispatch
* CTRL_STANDBY + PreVote → pre_vote_reply_decode_and_dispatch (task 5.1)
* any other role → drop packet (WARN)
*
* Packet ownership (raft-drive-async task 3.1): `pkt` is passed in explicitly —
* the caller (peer_ctrl_msg_handler) has already taken it off ep->fsm_in.pkt and
* cleared that slot to NULL. This body and the routers it calls free `pkt` on
* every path and NEVER touch ep->fsm_in.pkt, which is what lets this body run
* off the iomgr thread (on the vtask pool) without colliding with a concurrent
* io_worker read on the same endpoint.
*
* This body is run either INLINE on the io_worker thread (dispatch flag OFF —
* the production default; identical to the pre-task behavior) or on a vtask
* pool worker (flag ON; via peer_ctrl_vtask_entry).
*
* PreVote note (ha-prevote task 5.1): a preVoteReply arrives on the candidate's
* OUTBOUND PreVote endpoint while the candidate's role is still CTRL_STANDBY
* (a PreVote round performs no role change — Req 1.1). Without this branch the
@ -4809,33 +4866,199 @@ static int repl_ack_decode_and_dispatch(endpoint_t *ep, vote_ep_ctx_t *vctx)
* times out. We gate on raft.pre_vote_in_progress so a STANDBY node with no
* active round still drops unexpected traffic.
*/
static int peer_ctrl_msg_handler(void *ep_raw)
static int peer_ctrl_msg_handler_inline(endpoint_t *ep, io_packet_t *pkt)
{
endpoint_t *ep = (endpoint_t *)ep_raw;
vote_ep_ctx_t *vctx = (vote_ep_ctx_t *)ep->data;
ha_state_t *ha = vctx->ha;
/* raft-drive-async task 3.1: test-only observation seam — fires before any
* routing so a test can count/timestamp/park the body. NULL in production. */
if (ha_peerctrl_body_probe)
ha_peerctrl_body_probe(ep);
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(&ha->role,
memory_order_acquire);
/* PreVote replies land while CTRL_STANDBY with a round in progress. */
if (role == CTRL_STANDBY && ha->raft.pre_vote_in_progress)
return pre_vote_reply_decode_and_dispatch(ep, vctx);
return pre_vote_reply_decode_and_dispatch(vctx, pkt);
switch (role) {
case CTRL_ELECTING:
return vote_reply_decode_and_dispatch(ep, vctx);
return vote_reply_decode_and_dispatch(vctx, pkt);
case CTRL_PRIMARY:
return repl_ack_decode_and_dispatch(ep, vctx);
return repl_ack_decode_and_dispatch(ep, vctx, pkt);
default: {
io_packet_t *pkt = ep->fsm_in.pkt;
if (pkt) { free_packet(pkt); free(pkt); ep->fsm_in.pkt = NULL; }
default:
if (pkt) { free_packet(pkt); free(pkt); }
WARN("peer_ctrl_msg_handler: unexpected role %d, packet dropped", (int)role);
return 0;
}
}
/*
* peer_ctrl_vtask_entry (raft-drive-async task 3.1): vtask pool entry that runs
* the peer-control routing body off the iomgr thread. Extracts (ep, pkt) from
* the VTASK_IO task data — the SAME io_task_data_t shape the lib vtask path
* uses — takes ownership of pkt out of the task data (so vtask_fini, which only
* frees the data struct, cannot reach it) and hands it to the body, which frees
* it. No double-free: pkt is freed exactly once inside the body; run_task then
* frees the task + data struct. endpoint_unblock_tasks (per-endpoint
* serialization, Req 4.1/4.2) is driven by run_task after this returns.
*/
static int peer_ctrl_vtask_entry(vtask_t *task)
{
io_task_data_t *iod = &task->data->u.iodata;
endpoint_t *ep = iod->ep;
io_packet_t *pkt = iod->pkt;
iod->pkt = NULL; /* body owns pkt now; nothing else may reference it */
if (ep == NULL) {
if (pkt) { free_packet(pkt); free(pkt); }
return -EINVAL;
}
return peer_ctrl_msg_handler_inline(ep, pkt);
}
/*
* peer_ctrl_classify (raft-drive-async task 3.1): assign the LivenessAdmission
* task class for an inbound peer-control message. This handler decodes vote
* replies, PreVote replies and WAL/config ACKs — all RECEIPT decode + commit
* eval, which is bounded liveness work. Per the design's deadlock-freedom
* argument (R4.5), ack/commit-advance decode MUST be VTASK_LIVENESS so it is
* NEVER admission-gated: a quorum of acks can always find a reserved worker to
* advance committed_lsn even when the pool is saturated by *_BLOCK tasks.
*
* (The vote-GRANT term fsync — VTASK_LIVE_BLOCK, R4.6 — is handled on the
* INBOUND RequestVote handler, not on these reply/ack endpoints; and the long
* config-push/snapshot ops are offloaded to the SlowLane in tasks 3.2/3.3. So
* everything this handler dispatches is liveness work today.)
*
* ┌─ HIGH FLAG-ON HAZARD (raft-drive-async finding-2; bug-1198) ─────────────┐
* │ LIVENESS is correct ONLY once task 3.2 offloads the snapshot send off the │
* │ pool. TODAY the CTRL_PRIMARY repl-ack path can reach │
* │ HA_REPL_SEND_SNAPSHOT → persist_checkpoint (a BLOCKING fdatasync) INLINE │
* │ in repl_ack_decode_and_dispatch → ha_dispatch_repl_action_for_standby. │
* │ Running that blocking op while tagged VTASK_LIVENESS (never gated) breaks │
* │ the deadlock-freedom reservation — a burst of snapshot sends could occupy │
* │ every liveness worker and starve the acks that advance committed_lsn. │
* │ Task 3.2 offloads HA_REPL_SEND_SNAPSHOT to the SlowLane → the repl-ack │
* │ body becomes bounded → LIVENESS becomes correct. │
* │ DO NOT flip g_drive_async_peerctrl_dispatch_enabled ON until task 3.2 │
* │ has landed (and task 3.4's deadlock gate is green). │
* └──────────────────────────────────────────────────────────────────────────┘
*/
static vtask_class_t peer_ctrl_classify(ha_state_t *ha, io_packet_t *pkt)
{
(void)ha;
(void)pkt;
return VTASK_LIVENESS;
}
/*
* peer_ctrl_dispatch_to_pool (raft-drive-async task 3.1): hand the decoded
* peer-control packet to the vtask pool and return promptly. Per-endpoint
* arrival-order serialization is automatic — bind_vtask calls
* endpoint_block_tasks(ep) and the dispatcher won't bind a second task for the
* same endpoint until this one completes (Req 4.1/4.2).
*
* Controllers NEVER self-terminate (CLAUDE.md): on any failure to enqueue
* (pool unwired, OOM, or admission reject) we fail loud and run the body INLINE
* rather than drop a consensus message — staying alive and correct.
*
* ┌─ CRITICAL FLAG-ON HAZARD (raft-drive-async finding-1; bug-1197) ──────────┐
* │ Enabling this dispatch (g_drive_async_peerctrl_dispatch_enabled == 1) │
* │ introduces an endpoint use-after-free that does NOT exist while the body │
* │ runs inline. raft_cb_become_follower (ha.c, see its endpoint_fini loop) │
* │ runs on a POOL worker and calls endpoint_fini() on ALL peer vote-eps — │
* │ freeing each ep->data (vctx) via FREEUP. Meanwhile the single io_worker │
* │ thread can be concurrently inside peer_ctrl_msg_handler_inline(ep_j, ...) │
* │ for a same-batch queued event on another ep, reading ep_j->data → races a │
* │ freed vctx (null-deref / UAF). Safe TODAY only because the flag is OFF and │
* │ the io_worker processes peer-control sequentially. │
* │ FIX REQUIRED BEFORE task 3.4 FLIPS THE FLAG ON: every pool-reachable │
* │ teardown that frees a peer ep (raft_cb_become_follower's endpoint_fini │
* │ loop) MUST iomgr_unmonitor(ep) BEFORE endpoint_fini(ep) so the io_worker │
* │ can never deliver a new event on a freed ep->data. Do NOT enable dispatch │
* │ until that ordering is in place. │
* └──────────────────────────────────────────────────────────────────────────┘
*/
static int peer_ctrl_dispatch_to_pool(endpoint_t *ep, io_packet_t *pkt)
{
vote_ep_ctx_t *vctx = (vote_ep_ctx_t *)ep->data;
ha_state_t *ha = vctx->ha;
vtaskmgr_t *pool = (vtaskmgr_t *)ha->pool;
if (pool == NULL) {
WARN("peer_ctrl_dispatch_to_pool: vtask pool not wired — running "
"inline (stay alive, never drop a peer-control message)");
return peer_ctrl_msg_handler_inline(ep, pkt);
}
vtask_data_t *data = (vtask_data_t *)calloc(1, sizeof(*data));
vtask_t *task = (vtask_t *)calloc(1, sizeof(*task));
if (data == NULL || task == NULL) {
WARN("peer_ctrl_dispatch_to_pool: OOM allocating vtask — running "
"inline (stay alive)");
free(data);
free(task);
return peer_ctrl_msg_handler_inline(ep, pkt);
}
data->vtype = VTASK_IO;
data->u.iodata.ep = ep; /* drives per-endpoint serialization in bind_vtask */
data->u.iodata.pkt = pkt; /* ownership transferred into the task */
vtask_init(task, (vtask_func_t)peer_ctrl_vtask_entry, data);
/* raft-drive-async task 3.1: the task is the SOLE owner of pkt (the shim
* already cleared ep->fsm_in.pkt). Mark it so taskq_item_killer frees pkt
* if this task is destroyed while still queued (never executed) at
* vtaskmgr_fini — without this the pkt would leak on shutdown. The
* executed path frees pkt inside peer_ctrl_vtask_entry (and NULLs
* iod->pkt), so there is no double-free. */
task->owns_io_pkt = 1;
vtask_class_t cls = peer_ctrl_classify(ha, pkt);
int rc = vtask_enq_classified(pool, task, cls);
if (rc < 0) {
/* Task NOT enqueued — we still own task/data/pkt. LIVENESS is never
* gated so this should not happen; if it does, fail loud and run inline
* (never drop). Free the wrapper structs only — pkt is freed by the
* inline body below. */
WARN("peer_ctrl_dispatch_to_pool: vtask_enq_classified failed (rc=%d) "
"— running inline (stay alive)", rc);
free(data);
free(task);
return peer_ctrl_msg_handler_inline(ep, pkt);
}
return 0; /* returns promptly; the body runs on a pool worker */
}
/*
* peer_ctrl_msg_handler: the registered iomgr msg_handler for peer-control
* endpoints (set in on_vote_connected / on_pre_vote_connected / on_standby_
* connected). Runs on the io_worker thread.
*
* raft-drive-async task 3.1: takes ownership of the inbound packet off
* ep->fsm_in.pkt and clears that slot IMMEDIATELY (so a subsequent io_worker
* read on this endpoint allocates a fresh packet and cannot collide with the
* one we are about to process), then either runs the routing body inline
* (dispatch flag OFF — production default, byte-for-byte the prior behavior) or
* hands it to the vtask pool and returns promptly (flag ON).
*/
static int peer_ctrl_msg_handler(void *ep_raw)
{
endpoint_t *ep = (endpoint_t *)ep_raw;
io_packet_t *pkt = ep->fsm_in.pkt;
ep->fsm_in.pkt = NULL; /* take ownership before any routing/dispatch */
if (!ha_peerctrl_dispatch_enabled())
return peer_ctrl_msg_handler_inline(ep, pkt);
return peer_ctrl_dispatch_to_pool(ep, pkt);
}
#ifdef XCE_TESTING

32
src/controller/ha.h
View file

@ -173,6 +173,12 @@ typedef struct {
_Atomic uint64_t committed_config_tag; /* Req 8.1: set on quorum ACK */
/* ── New fields added by election-iomgr-refactor spec (Task 2.1) ── */
void *client_ssl_ctx; /* SSL_CTX* (TLS_client_method); created in myapp_run */
/* ── New field added by raft-drive-async spec (Task 3.1) ──────── */
void *pool; /* vtaskmgr_t* (cast; avoids pulling vtaskmgr.h
* into every ha.h consumer). Set once at init
* (cluster_init_phase2 → &c->vtaskmgr). Used ONLY
* by peer_ctrl_msg_handler when peer-control
* dispatch is enabled; NULL otherwise (gate OFF). */
int phase2_done; /* 1 after cluster_init_phase2 runs */
int n_peers; /* total cluster size (including self), matches raft convention */
/* ── New fields added by raft-library spec (Task 6.1) ──────── */
@ -1221,6 +1227,32 @@ void ha_start_primary_heartbeat(cluster_t *c);
void ha_set_reconnect_offload_for_test(int on);
int ha_reconnect_offload_enabled(void);
/*
* ha_set_peerctrl_dispatch_for_test / ha_peerctrl_dispatch_enabled:
* Feature-flag accessors (raft-drive-async Task 3.1). The dispatch flag gates
* whether the registered peer-control handler (peer_ctrl_msg_handler) runs its
* routing body INLINE on the iomgr io_worker thread (OFF — the production
* default for this commit; behavior is byte-for-byte identical to before the
* task) or hands the decoded packet to the vtask pool via vtask_enq_classified
* and returns promptly (ON — the target behavior, validated by the task 3.4
* deadlock gate before it becomes the default). These accessors are compiled
* in ALL builds; the setter is used by tests to flip RED<->GREEN.
*
* DEFAULT: OFF (0). Production behavior is UNCHANGED by this commit.
*/
void ha_set_peerctrl_dispatch_for_test(int on);
int ha_peerctrl_dispatch_enabled(void);
/*
* ha_peerctrl_body_probe: test-only observation seam (raft-drive-async Task
* 3.1). When non-NULL it is invoked at the TOP of the peer-control routing
* body (peer_ctrl_msg_handler_inline) with the endpoint, BEFORE any role
* routing. Tests use it to (a) count body invocations, (b) capture the thread
* the body ran on (proving inline-vs-pool execution), and (c) optionally park
* the body to prove per-endpoint arrival-order serialization. NULL in
* production (zero cost) — a plain global like ha_nop_req_hook. */
extern void (*ha_peerctrl_body_probe)(endpoint_t *ep);
/* ── Task 2.4: Non-blocking async election initiator ────────────────────── */
/*

269
src/controller/testsuite/test_ha_liveness.c
View file

@ -1480,6 +1480,271 @@ static void test_reconnect_preserves_other_peer_channel(void)
printf("test_reconnect_preserves_other_peer_channel: PASS\n");
}
/* ═══════════════════════════════════════════════════════════════════════════
* raft-drive-async Task 3.1 — PeerCtrlDispatch
*
* The registered peer-control handler (peer_ctrl_msg_handler) must, when the
* dispatch flag is ON, hand the decoded packet to the vtask pool and return
* PROMPTLY (so peer-control no longer runs on the iomgr io_worker thread), while
* preserving per-connection arrival order; with the flag OFF it must run the
* routing body INLINE exactly as before (no behavior change — the production
* default for this commit). The flag toggle is the RED↔GREEN driver.
*
* Test 18 (Req 1.4, 4.5, 8.1, 8.2): flag ON → handler returns promptly and
* the routing body runs on a POOL worker (a different thread).
* Test 19 (Req 6.x no-regression): flag OFF → body runs INLINE, synchronously,
* on the calling thread (identical to the pre-task behavior).
* Test 20 (Req 4.1, 4.2): two messages on ONE endpoint are processed in
* arrival order — the second cannot run until the first completes
* (per-endpoint endpoint_block_tasks serialization).
* ═══════════════════════════════════════════════════════════════════════════ */
extern int peer_ctrl_msg_handler_for_test(endpoint_t *ep, io_packet_t *pkt);
extern io_packet_t *vote_reply_build_pkt_for_test(uint64_t term, int vote_granted,
uint64_t peer_tag,
uint64_t peer_lsn);
/* Probe observation state (set via ha_peerctrl_body_probe). */
static _Atomic int g_pc_runs = 0; /* body invocation count */
/* _Atomic: written on the pool worker, read on main — avoid a formal C11 data
* race (raft-drive-async review finding-2). relaxed ordering is sufficient:
* g_pc_runs (acquire/release via fetch_add/load) already orders visibility. */
static _Atomic pthread_t g_pc_last_thread; /* thread the body ran on */
static _Atomic int g_pc_gate_on = 0; /* 1 → body parks at the gate */
static pthread_mutex_t g_pc_mu = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t g_pc_entered_cv = PTHREAD_COND_INITIALIZER;
static pthread_cond_t g_pc_release_cv = PTHREAD_COND_INITIALIZER;
static int g_pc_parked = 0; /* a body is parked at the gate */
static int g_pc_release = 0; /* gate open → bodies may proceed */
static void pc_probe(endpoint_t *ep)
{
(void)ep;
atomic_store_explicit(&g_pc_last_thread, pthread_self(), memory_order_relaxed);
atomic_fetch_add(&g_pc_runs, 1);
if (atomic_load(&g_pc_gate_on)) {
pthread_mutex_lock(&g_pc_mu);
g_pc_parked = 1;
pthread_cond_broadcast(&g_pc_entered_cv);
while (!g_pc_release)
pthread_cond_wait(&g_pc_release_cv, &g_pc_mu);
g_pc_parked = 0;
pthread_mutex_unlock(&g_pc_mu);
}
}
/* Wait (bounded) until a body has parked at the gate. */
static int pc_wait_parked(int timeout_ms)
{
struct timespec d;
clock_gettime(CLOCK_REALTIME, &d);
d.tv_sec += timeout_ms / 1000;
d.tv_nsec += (long)(timeout_ms % 1000) * 1000000L;
if (d.tv_nsec >= 1000000000L) { d.tv_sec++; d.tv_nsec -= 1000000000L; }
pthread_mutex_lock(&g_pc_mu);
int rc = 0;
while (!g_pc_parked) {
rc = pthread_cond_timedwait(&g_pc_entered_cv, &g_pc_mu, &d);
if (rc != 0) break;
}
int parked = g_pc_parked;
pthread_mutex_unlock(&g_pc_mu);
return parked ? 0 : -1;
}
static void pc_release_gate(void)
{
pthread_mutex_lock(&g_pc_mu);
g_pc_release = 1;
pthread_cond_broadcast(&g_pc_release_cv);
pthread_mutex_unlock(&g_pc_mu);
}
/* Bounded poll until g_pc_runs >= want; returns the observed count. */
static int pc_wait_runs(int want, int timeout_ms)
{
for (int i = 0; i < timeout_ms; i++) {
if (atomic_load(&g_pc_runs) >= want) break;
struct timespec ts = { .tv_sec = 0, .tv_nsec = 1000000L }; /* 1 ms */
nanosleep(&ts, NULL);
}
return atomic_load(&g_pc_runs);
}
static void pc_reset_probe_state(void)
{
atomic_store(&g_pc_runs, 0);
atomic_store(&g_pc_gate_on, 0);
pthread_mutex_lock(&g_pc_mu);
g_pc_parked = 0;
g_pc_release = 0;
pthread_mutex_unlock(&g_pc_mu);
}
/*
* Test 18 (Req 1.4, 4.5, 8.1, 8.2): with the dispatch flag ON, the registered
* handler enqueues the decoded packet onto the vtask pool and RETURNS PROMPTLY;
* the routing body then runs on a POOL worker — a DIFFERENT thread from the
* caller (proving peer-control no longer runs on the iomgr io_worker thread).
* The body (role=CTRL_INIT → bounded default-drop path) frees the packet.
*/
static void test_peerctrl_dispatch_runs_on_pool(void)
{
vtaskmgr_t mgr;
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&mgr, 4));
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&mgr));
cluster_t c;
memset(&c, 0, sizeof(c));
ha_init(&c.ha, &g_cfg, 0);
c.ha.pool = &mgr;
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_INIT, memory_order_release);
endpoint_t ep;
memset(&ep, 0, sizeof(ep));
standby_ep_ctx_build_for_test(&ep, &c.ha, &c, 0);
pc_reset_probe_state();
ha_peerctrl_body_probe = pc_probe;
ha_set_peerctrl_dispatch_for_test(1);
pthread_t caller = pthread_self();
io_packet_t *pkt = vote_reply_build_pkt_for_test(1, 1, 0, 0);
TEST_ASSERT_NOT_NULL(pkt);
int rc = peer_ctrl_msg_handler_for_test(&ep, pkt);
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc,
"Req 8.1/8.2: dispatch handler must return promptly (0) after enqueue");
int runs = pc_wait_runs(1, 2000);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, runs,
"Req 1.4/4.5: the routing body must run on the pool exactly once");
TEST_ASSERT_FALSE_MESSAGE(
pthread_equal(atomic_load_explicit(&g_pc_last_thread, memory_order_relaxed),
caller),
"Req 1.4: the body must run on a POOL worker, NOT the caller's "
"(io_worker) thread");
ha_set_peerctrl_dispatch_for_test(0); /* restore production default (OFF) */
ha_peerctrl_body_probe = NULL;
vtaskmgr_fini(&mgr);
free(ep.data);
ha_fini(&c.ha);
printf("test_peerctrl_dispatch_runs_on_pool: PASS\n");
}
/*
* Test 19 (Req 6.x no-regression): with the dispatch flag OFF (the production
* default), the handler runs the routing body INLINE on the calling thread,
* synchronously — byte-for-byte the pre-task behavior. This is the RED side of
* the flag toggle: the body did NOT move to a pool thread.
*/
static void test_peerctrl_inline_when_flag_off(void)
{
cluster_t c;
memset(&c, 0, sizeof(c));
ha_init(&c.ha, &g_cfg, 0);
c.ha.pool = NULL; /* OFF path must not need the pool */
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_INIT, memory_order_release);
endpoint_t ep;
memset(&ep, 0, sizeof(ep));
standby_ep_ctx_build_for_test(&ep, &c.ha, &c, 0);
pc_reset_probe_state();
ha_peerctrl_body_probe = pc_probe;
ha_set_peerctrl_dispatch_for_test(0); /* explicit OFF */
pthread_t caller = pthread_self();
io_packet_t *pkt = vote_reply_build_pkt_for_test(1, 1, 0, 0);
TEST_ASSERT_NOT_NULL(pkt);
int rc = peer_ctrl_msg_handler_for_test(&ep, pkt);
TEST_ASSERT_EQUAL_INT(0, rc);
/* Synchronous: the body already ran on THIS thread before the handler
* returned — no pool, no wait. */
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_pc_runs),
"Req 6.x: with the flag OFF the body must run inline (synchronously)");
TEST_ASSERT_TRUE_MESSAGE(
pthread_equal(atomic_load_explicit(&g_pc_last_thread, memory_order_relaxed),
caller),
"Req 6.x: with the flag OFF the body must run on the caller's thread "
"(inline), unchanged from before the task");
ha_peerctrl_body_probe = NULL;
free(ep.data);
ha_fini(&c.ha);
printf("test_peerctrl_inline_when_flag_off: PASS\n");
}
/*
* Test 20 (Req 4.1, 4.2): two peer-control messages arriving on the SAME
* endpoint are processed in arrival order — the second body cannot start until
* the first completes. Proof: with the flag ON, the first body parks at the
* gate (holding its pool worker; the endpoint is blocked by bind_vtask's
* endpoint_block_tasks). While it is parked, dispatching a second message on
* the same endpoint does NOT run its body (run count stays 1). Releasing the
* first lets it complete (endpoint_unblock_tasks) and only THEN does the second
* run (count → 2).
*/
static void test_peerctrl_preserves_per_endpoint_order(void)
{
vtaskmgr_t mgr;
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_init(&mgr, 4));
TEST_ASSERT_EQUAL_INT(0, vtaskmgr_start(&mgr));
cluster_t c;
memset(&c, 0, sizeof(c));
ha_init(&c.ha, &g_cfg, 0);
c.ha.pool = &mgr;
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_INIT, memory_order_release);
endpoint_t ep;
memset(&ep, 0, sizeof(ep));
standby_ep_ctx_build_for_test(&ep, &c.ha, &c, 0);
pc_reset_probe_state();
atomic_store(&g_pc_gate_on, 1); /* bodies park at the gate */
ha_peerctrl_body_probe = pc_probe;
ha_set_peerctrl_dispatch_for_test(1);
/* Message 1 — dispatched; its body parks at the gate. */
io_packet_t *pkt1 = vote_reply_build_pkt_for_test(1, 1, 0, 0);
TEST_ASSERT_NOT_NULL(pkt1);
TEST_ASSERT_EQUAL_INT(0, peer_ctrl_msg_handler_for_test(&ep, pkt1));
TEST_ASSERT_EQUAL_INT_MESSAGE(0, pc_wait_parked(2000),
"the first body must reach the gate (running on the pool)");
/* Message 2 — dispatched on the SAME endpoint while body 1 is parked. */
io_packet_t *pkt2 = vote_reply_build_pkt_for_test(1, 1, 0, 0);
TEST_ASSERT_NOT_NULL(pkt2);
TEST_ASSERT_EQUAL_INT(0, peer_ctrl_msg_handler_for_test(&ep, pkt2));
/* Give a generous window: body 2 must NOT run while body 1 holds the
* endpoint (per-connection serialization). */
struct timespec ts = { .tv_sec = 0, .tv_nsec = 200 * 1000000L }; /* 200 ms */
nanosleep(&ts, NULL);
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_pc_runs),
"Req 4.1/4.2: the second message must NOT be processed while the first "
"is in flight on the same connection");
/* Release body 1; body 2 then becomes eligible and runs. */
pc_release_gate();
int runs = pc_wait_runs(2, 2000);
TEST_ASSERT_EQUAL_INT_MESSAGE(2, runs,
"Req 4.1/4.2: the second message runs only AFTER the first completes "
"(arrival order preserved)");
ha_set_peerctrl_dispatch_for_test(0); /* restore production default (OFF) */
ha_peerctrl_body_probe = NULL;
vtaskmgr_fini(&mgr);
free(ep.data);
ha_fini(&c.ha);
printf("test_peerctrl_preserves_per_endpoint_order: PASS\n");
}
/* ── Suite entry point ──────────────────────────────────────────────────── */
void test_ha_liveness(void)
@ -1513,5 +1778,9 @@ void test_ha_liveness(void)
/* raft-drive-async Task 2.3 — reconnect-offload INTEGRATION (Req 2.1, 2.2, 2.5) */
RUN_TEST(test_reconnect_offload_sustains_survivor_heartbeats);
RUN_TEST(test_reconnect_inline_stalls_survivor_heartbeats_RED);
/* raft-drive-async Task 3.1 — PeerCtrlDispatch (Req 1.4, 4.1, 4.2, 4.5, 6.x, 8.1, 8.2) */
RUN_TEST(test_peerctrl_dispatch_runs_on_pool);
RUN_TEST(test_peerctrl_inline_when_flag_off);
RUN_TEST(test_peerctrl_preserves_per_endpoint_order);
UNITY_END();
}

10
src/include/vtaskmgr.h
View file

@ -48,6 +48,16 @@ typedef struct {
int vcpu;
vtask_func_t entry;
vtask_class_t cls; /* LivenessAdmission class (task 1.6); default LIVENESS */
/* raft-drive-async task 3.1: when non-zero AND data->vtype == VTASK_IO, the
* carried io_packet_t (data->u.iodata.pkt) is OWNED by the task and must be
* freed by taskq_item_killer if the task is destroyed while still queued
* (never executed) at vtaskmgr_fini. Default 0 preserves the legacy
* endpoint-owned-packet model used by the libxce client path (xcectx.c's
* vtask_handle_msg does NOT clear ep->fsm_in.pkt, so the packet lifetime is
* tied to the endpoint, not the task — freeing it here would double-free).
* The controller's peer-control dispatch sets this to 1 because it clears
* ep->fsm_in.pkt and the task becomes the sole owner. */
int owns_io_pkt;
} vtask_t;
#define ITPTR_TO_VTASK(p) ((vtask_t*)(((char*)p) - offsetof(vtask_t, link)))

21
src/lib/vtaskmgr.c
View file

@ -27,6 +27,7 @@ int vtask_init(vtask_t* task, vtask_func_t entry, void* data)
task->data = data;
task->owner = NULL;
task->cls = VTASK_LIVENESS; /* never-gated by default; legacy enq paths */
task->owns_io_pkt = 0; /* legacy: endpoint owns the packet (task 3.1) */
return 0;
}
@ -166,6 +167,26 @@ static void taskq_item_killer(data_item_t* it)
{
if (it != NULL) {
vtask_t* task = ITPTR_TO_VTASK(it);
/* raft-drive-async task 3.1: a task destroyed while still queued (never
* executed) at vtaskmgr_fini would otherwise leak its carried packet —
* the ownership transfer (iod->pkt = NULL) only happens inside the
* executed entry (peer_ctrl_vtask_entry). Free the packet here ONLY
* when the producer marked the task as the packet's sole owner
* (owns_io_pkt); the legacy endpoint-owned model (owns_io_pkt == 0,
* e.g. the libxce client path) is left untouched so we never double-free
* a packet whose lifetime is the endpoint's. This runs only on
* un-executed tasks, so it cannot race the entry's free. vtask_fini
* frees only task->data (the wrapper struct), never the packet. */
if (task->owns_io_pkt && task->data != NULL &&
task->data->vtype == VTASK_IO &&
task->data->u.iodata.pkt != NULL) {
io_packet_t* pkt = task->data->u.iodata.pkt;
free_packet(pkt);
free(pkt);
task->data->u.iodata.pkt = NULL;
}
vtask_fini(task);
free(task);
}