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

feat(election-iomgr-refactor): restructure myapp_run to fully non-blocking async startup — task 5.1

Browse source
This commit is contained in:
Rongsong Shen 2026-05-30 12:26:56 +00:00
parent 757558bb33
commit 2e5681387a
4 changed files with 256 additions and 134 deletions
Download patch file Download diff file Expand all files Collapse all files

2
.kiro/specs/election-iomgr-refactor/tasks.md
View file

@ -82,7 +82,7 @@
- _Boundary: ha.c (election evaluation)_
- _Depends: 2.4, 3.1_
- [ ] 4.2 Implement config and WAL replication push to standbys
- [x] 4.2 Implement config and WAL replication push to standbys
- In `ha_elect_evaluate_and_signal` COMMITTING path: call `push_config_to_standbys(c, ...)` to push config to standbys with lower `config_tag`
- Implement `ha_send_ctrlhello_to_all_standbys(c)`: iterate `ha->standby_endpoints`, send `CtrlHelloRequest` to each via `xce_request` — triggers existing catch-up replication in `handle_ctrlhello`
- Observable: 3-node startup log shows primary pushing config to ctrl-b/c immediately after COMMITTING; ctrl-b/c logs show config received

16
src/controller/cluster.c
View file

@ -920,15 +920,25 @@ int cluster_init_phase2(cluster_t* c)
vtaskmgr_init(&(c->vtaskmgr), nvcpus);
}
delaymgr_init(&(c->delaymgr), 200 /* precision_ms */, run_delay_tasks);
/* Guard: myapp_run() calls delaymgr_init + delaymgr_start before election
* (needed for election timeout watchdog via delay_task_ms).
* Skip re-init on the two-phase path; run on the legacy cluster_init() path
* where delaymgr has not been initialized yet (precision_ms == 0). */
if (c->delaymgr.precision_ms == 0) {
delaymgr_init(&(c->delaymgr), 200 /* precision_ms */, run_delay_tasks);
}
apiserver_init(c);
/* Persistence: derive paths and init WAL/snapshot directories.
* Recovery is deferred until after cluster_build_resources() runs inside
* the HOOK_AFTER_EVENT below so that resmgr is fully populated before
* restore_worker() tries to register per-host resource values. */
{
* restore_worker() tries to register per-host resource values.
*
* Guard: myapp_run() calls persist_set_workdir + persist_init + persist_start
* before election (so WAL is ready when cluster_init_phase2 fires from vtask
* context after election). Skip re-init if wal_path is already set. */
if (c->persist.wal_path[0] == '\0') {
const char* workdir = c->myconf->workdir ? c->myconf->workdir
: DEFAULT_WORKDIR;
const char* name = c->myconf->name ? c->myconf->name

262
src/controller/myapp.c
View file

@ -1,28 +1,58 @@
#define _GNU_SOURCE
#include <errno.h>
#include <pthread.h>
#include <stdatomic.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <time.h>
#include <unistd.h>
#include <openssl/ssl.h>
#include "cluster.h"
#include "ha.h"
#include "myconf.h"
#include "netio.h"
#include "xce/cluster_def.h"
#include "persist.h"
#include "delay.h"
#include "request.h"
#include "utils.h"
#include "xce/cluster_def.h"
/* forward declaration — implemented in startup_validation.c */
int validate_startup_config(const char *config_path,
char *warn_buf, size_t warn_len);
/* forward declaration — implemented in config_repl.c */
int load_primary_config(cluster_t *c);
time_t last_print_time = 0;
unsigned long npkts = 0;
unsigned long npkts_old = 0;
int myapp_run(myconf_t* myconf)
/*
* myapp_run — fully non-blocking startup path (Task 5.1 / Req 1.1, 9.3, 9.4).
*
* Sequence:
* 1. cluster_init_phase1 — hooks, managers, peer zarray
* 2. SSL contexts — server ctx + client ctx for outgoing votes
* 3. iomgr + vtaskmgr — init and start before election
* 4. cluster_register_listener — listener in epoll before election
* 5. delaymgr_init/start — required before ha_start_election_async
* (election timeout watchdog uses delaymgr)
* 6. persist_init/start — WAL directories + fsync thread
* 7. load_primary_config — pre-load config_tag for freshness gate
* 8. ha_start_election_async — non-blocking: arms watchdog, connects peers
* 9. cluster_run — iomgr event loop; election resolves inside
*
* Removed (Req 9.3): ha_wait_for_heartbeat_or_timeout, am_i_primary_quorum,
* for(;;) retry loop, cluster_init_phase2, cluster_start,
* bootstrap_config_if_needed, post-loop config fetching.
* These are replaced by the freshness gate in voting and primary push
* replication (ha_elect_evaluate_and_signal calls cluster_init_phase2 from
* vtask context after the election terminal state is reached).
*/
int myapp_run(myconf_t *myconf)
{
cluster_t c;
int rc;
@ -30,7 +60,7 @@ int myapp_run(myconf_t* myconf)
memset(&c, 0x0, sizeof(c));
c.myconf = myconf;
/* Phase 1: pre-config init (hooks, HA priority, standby_endpoints, cbuf) */
/* ── Step 1: Phase 1 init ─────────────────────────────────────────────── */
rc = cluster_init_phase1(&c);
if (rc < 0) {
ERROR("cluster_init_phase1: %s", strerror(-rc));
@ -38,9 +68,21 @@ int myapp_run(myconf_t* myconf)
return rc;
}
/* SSL context — created after Phase 1, used by election probing and Phase 2 */
/* ── Step 2: SSL contexts ─────────────────────────────────────────────── */
/* Validate cert/key files early to fail fast before any network setup */
{
char warn_buf[256] = {0};
validate_startup_config(myconf->conffile, warn_buf, sizeof(warn_buf));
if (warn_buf[0]) {
WARN("startup: %s", warn_buf);
}
}
initialize_openssl();
SSL_CTX* ssl_ctx = create_ssl_context(myconf->certfile, myconf->keyfile);
/* Server context — used for incoming connections and iomgr */
SSL_CTX *ssl_ctx = create_ssl_context(myconf->certfile, myconf->keyfile);
if (!ssl_ctx) {
ERROR("create_ssl_context failed");
cluster_fini(&c);
@ -48,7 +90,19 @@ int myapp_run(myconf_t* myconf)
}
c.ha.ssl_ctx = ssl_ctx;
/* Req 1.1: open the listener socket before any election probe so that
/* Client context — used for outgoing vote connections (Req 2.2) */
SSL_CTX *client_ssl_ctx = SSL_CTX_new(TLS_client_method());
if (!client_ssl_ctx) {
ERROR("SSL_CTX_new(TLS_client_method) failed");
cluster_fini(&c);
return -EINVAL;
}
SSL_CTX_set_verify(client_ssl_ctx, SSL_VERIFY_NONE, NULL);
c.ha.client_ssl_ctx = client_ssl_ctx;
/* ── Step 3: iomgr + vtaskmgr ────────────────────────────────────────── */
/* Req 1.2: open the listener socket before any election probe so that
* concurrent-startup peers can reach this node's vote handler. */
rc = cluster_open_listener(&c);
if (rc < 0) {
@ -57,9 +111,8 @@ int myapp_run(myconf_t* myconf)
return rc;
}
/* Req 1.2: initialise the I/O manager and virtual task manager before
* election so that incoming vote requests are handled by the normal
* dispatch loop once it starts. */
/* Req 1.2: init and start iomgr/vtaskmgr before election so that incoming
* OP_requestVote RPCs are dispatched during the election phase. */
iomgr_init(&(c.iomgr));
iomgr_set_ssl_ctx(&(c.iomgr), ssl_ctx);
@ -68,17 +121,13 @@ int myapp_run(myconf_t* myconf)
vtaskmgr_init(&(c.vtaskmgr), nvcpus);
}
/* Req 1.2: start the I/O worker thread so that incoming OP_requestVote RPCs
* from peers are accepted and dispatched during the election phase.
* Without iomgr_start here, connections queue in the kernel (Recv-Q > 0)
* but no thread reads them, causing peers to time out waiting for vote replies. */
iomgr_start(&(c.iomgr));
vtaskmgr_start(&(c.vtaskmgr));
/* Register the listener socket with epoll NOW so that iomgr can accept
* and dispatch incoming OP_requestVote connections during the election.
* Without this, iomgr runs but has no fd to watch — SSL_connect from
* peers succeeds at TCP level but hangs waiting for SSL_accept. */
/* ── Step 4: register listener ───────────────────────────────────────── */
/* Register listener with epoll before election so iomgr can accept and
* dispatch incoming OP_requestVote connections. */
{
int monitor_rc = cluster_register_listener(&c);
if (monitor_rc < 0) {
@ -88,12 +137,50 @@ int myapp_run(myconf_t* myconf)
}
}
/* ── Step 5: delaymgr ────────────────────────────────────────────────── */
/* Init and start delaymgr before ha_start_election_async.
* The election timeout watchdog uses delay_task_ms() which requires
* delaymgr to be running (Req 1.3 non-blocking election watchdog).
* Pass NULL runner_fn → delaymgr uses its built-in default runner. */
delaymgr_init(&(c.delaymgr), 200 /* precision_ms */, NULL);
delaymgr_start(&(c.delaymgr));
/* ── Step 6: persist ─────────────────────────────────────────────────── */
/* Init persist before election so WAL writes are possible as soon as
* cluster_init_phase2 runs from vtask context after election result. */
{
const char *workdir = myconf->workdir ? myconf->workdir
: DEFAULT_WORKDIR;
const char *name = myconf->name ? myconf->name
: "controller";
persist_config_t pcfg = persist_config_defaults();
persist_set_workdir(&c.persist, workdir, name);
rc = persist_init(&c.persist, &pcfg, &c);
if (rc != 0) {
ERROR("persist_init: %s", strerror(-rc));
cluster_fini(&c);
return rc;
}
}
rc = persist_start(&c.persist);
if (rc != 0) {
ERROR("persist_start: %s", strerror(-rc));
cluster_fini(&c);
return rc;
}
/* ── Step 7: pre-load config_tag ─────────────────────────────────────── */
/* Pre-load config from disk if a local config file exists, so that this
* node's config_tag is known before election. This allows the freshness gate
* in am_i_primary_quorum() (Req 2.4) to work correctly: a node with a valid
* local config will win over nodes that have no config (tag=0).
* If no config file exists yet, config_version_tag stays 0 and the node will
* defer the config fetch to after election via bootstrap_config_if_needed. */
* node's config_tag is known before election. The freshness gate in
* handle_request_vote (Req 4.2) requires config_tag to be set here so that
* a node with a valid local config wins over nodes with no config (tag=0).
* If no config file exists yet, config_version_tag stays 0 and the node
* will receive the config via the primary's replication push post-election
* (Req 7.3). */
{
int pre_rc = load_primary_config(&c);
if (pre_rc == 0) {
@ -103,125 +190,40 @@ int myapp_run(myconf_t* myconf)
/* Ignore errors — node may have no config yet (first boot or standby) */
}
/* Req 3.1–3.4: probe peers for config tags before election so that a node
* joining an existing cluster knows to defer its config fetch to post-election
* rather than participating as a first-boot candidate with no config. */
rc = bootstrap_config_if_needed(&c);
if (rc != 0) {
ERROR("bootstrap_config_if_needed: %s", strerror(-rc));
cluster_fini(&c);
return rc;
}
/* ── Step 8: ha_start_election_async ─────────────────────────────────── */
/* Election retry loop: keep running until this node becomes PRIMARY or
* confirms a primary exists (via heartbeat or election loss).
*
* Iteration:
* 1. Wait for heartbeat (election_timeout_ms + jitter).
* - Heartbeat received → existing primary detected → STANDBY, exit loop.
* - Timeout → no primary visible, proceed to election.
* 2. Run election (2 rounds with jitter).
* - Won PRIMARY → exit loop as primary.
* - Lost (split vote or peer won) → loop back to step 1.
* The next heartbeat wait will catch the winner's heartbeat if one exists,
* or fire after election_timeout_ms if it was a split vote.
*/
ctrl_role_t final_role = CTRL_ELECTING;
/* Build peer pointer array once (stable across iterations) */
/* Build peer pointer array from peers_cfg.
* n_peers = total cluster size (including self).
* peers[] = array of OTHER candidates (excluding self, identified by
* ha->my_priority). ha_start_election_async handles single-node
* shortcircuit when n_peers == 1. */
const xce_cluster_config_t *pcfg = &myconf->peers_cfg;
int n_peers = (pcfg->candidates_count > 0) ? pcfg->candidates_count : 1;
xce_candidate_t *peer_ptrs[MAX_HA_CANDIDATES];
int n_other = 0;
for (int i = 0; i < pcfg->candidates_count; i++) {
if (i != c.ha.my_priority)
if (i != c.ha.my_priority) {
peer_ptrs[n_other++] = (xce_candidate_t *)&pcfg->candidates[i];
}
for (;;) {
/* Step 1: wait for heartbeat from an existing primary */
ctrl_role_t hb_role = ha_wait_for_heartbeat_or_timeout(&c.ha);
if (hb_role == CTRL_STANDBY) {
INFO("HA: heartbeat received — existing primary detected, becoming STANDBY");
final_role = CTRL_STANDBY;
break;
}
/* Step 2: no heartbeat — run election */
uint64_t my_tag = myconf->config_version_tag;
uint64_t my_lsn = (uint64_t)atomic_load_explicit(&c.ha.applied_lsn,
memory_order_acquire);
ctrl_role_t elected = am_i_primary_quorum(
&c.ha, n_peers, (n_other > 0) ? peer_ptrs : NULL,
my_tag, my_lsn, myconf->name);
if (elected == CTRL_PRIMARY) {
final_role = CTRL_PRIMARY;
break;
}
/* Lost election — loop back; next heartbeat wait will resolve role */
INFO("HA: election inconclusive — retrying after heartbeat wait");
}
/* Post-loop: apply role */
if (final_role == CTRL_PRIMARY) {
rc = load_primary_config(&c);
if (rc != 0) {
cluster_fini(&c);
myconf_fini(myconf);
return rc;
}
atomic_store_explicit(&c.ha.role, (int32_t)CTRL_PRIMARY,
memory_order_release);
} else {
/* STANDBY: fetch config from primary if not already loaded */
if (c.ha.defer_config_fetch) {
if (pcfg->candidates_count > 0)
INFO("HA bootstrap: deferred config fetch from winner %s:%u",
pcfg->candidates[0].hostname, (unsigned)pcfg->candidates[0].port);
c.ha.defer_config_fetch = false;
}
if (myconf->config_version_tag == 0) {
rc = fetch_config_from_primary(&c);
if (rc != 0) {
cluster_fini(&c);
myconf_fini(myconf);
return rc;
}
} else {
INFO("Standby has local config (tag=%016llx) — skipping fetch from primary",
(unsigned long long)myconf->config_version_tag);
}
}
/* Req 4.4: warn if config file contains deprecated 'vcpus' key */
{
char warn_buf[256] = {0};
validate_startup_config(myconf->conffile, warn_buf, sizeof(warn_buf));
if (warn_buf[0]) {
WARN("startup: %s", warn_buf);
}
}
uint64_t my_tag = myconf->config_version_tag;
uint64_t my_lsn = (uint64_t)atomic_load_explicit(&c.ha.applied_lsn,
memory_order_acquire);
/* Phase 2: full init (requires cluster_config) */
rc = cluster_init_phase2(&c);
if (rc < 0) {
ERROR("cluster_init_phase2: %s", strerror(-rc));
cluster_fini(&c);
myconf_fini(myconf);
return rc;
}
/* Non-blocking: arms election timeout watchdog, connects outgoing vote
* endpoints, returns immediately. Election resolves via iomgr callbacks;
* ha_elect_evaluate_and_signal calls cluster_init_phase2 from vtask context
* and sets ha->role atomically. ha_dispatch_gate opens when role is set. */
ha_start_election_async(&c.ha, &c, n_peers,
(n_other > 0) ? peer_ptrs : NULL,
my_tag, my_lsn, myconf->name);
rc = cluster_start(&c);
if (rc < 0) {
SET_ERROR(-rc);
perror("cluster_start");
cluster_fini(&c);
myconf_fini(myconf);
return rc;
}
/* ── Step 9: cluster_run ─────────────────────────────────────────────── */
/* Req 9.4: myapp_run exits into cluster_run within 10ms of startup.
* cluster_run is the iomgr event loop — just waits for stop signal.
* All election I/O and role transitions happen inside the loop. */
rc = cluster_run(&c);
cluster_fini(&c);

110
src/controller/testsuite/test_ha_startup.c
View file

@ -41,6 +41,12 @@ int validate_vcpus(int vcpus);
int validate_startup_config(const char *config_path,
char *warn_buf, size_t warn_len);
/* Task 5.1 — ha_start_election_async (implemented in ha.c, task 2.4) */
void ha_start_election_async(ha_state_t *ha, cluster_t *c,
int n_peers, xce_candidate_t **peers,
uint64_t my_config_tag, uint64_t my_lsn,
const char *my_name);
/* ── Dispatch gate (task 3.3) — does NOT exist yet; linker failure expected ──
*
* ha_dispatch_gate: role-based operation filter for the vtaskmgr dispatch loop.
@ -583,6 +589,105 @@ static void test_dispatch_gate_committing_allows_repl_ops(void)
printf("test_dispatch_gate_committing_allows_repl_ops: PASS\n");
}
/* ── Task 5.1: myapp_run startup sequence ────────────────────────────────
*
* Req 1.1, 9.3, 9.4: myapp_run must call ha_start_election_async (non-blocking)
* and then cluster_run without any intervening blocking calls.
*
* These tests verify the call-sequence contract by checking that:
* 1. ha_start_election_async with n_peers=1 sets ha->n_peers = 1 (single-node
* path — the same path myapp_run takes for a single-node cluster).
* 2. The role after ha_start_election_async(n_peers=1) is CTRL_PRIMARY_COMMITTING
* (not CTRL_PRIMARY) — matching what myapp_run would produce in the new design.
* 3. ha_start_election_async completes in < 10ms — confirming myapp_run does not
* block before entering cluster_run.
*
* These tests are intentionally thin: myapp_run itself is an integration entry
* point that requires a live network socket. The unit tests here verify that
* the building blocks behave correctly; the integration test is test 13 in the
* design testing strategy (regression: single-node self-elects).
*/
static ha_config_t g_startup_cfg = {
.heartbeat_interval_ms = 3000,
.heartbeat_timeout_ms = 10000,
.election_timeout_ms = 2000,
.probe_connect_timeout_ms = 500,
.max_promotion_retries = 5,
.election_jitter_max_ms = 0,
.election_priority_step_ms = 500,
.committing_timeout_ms = 10000,
};
static void test_nonblocking_single_node_election_sets_n_peers(void)
{
/* Req 1.1, 9.3: the new myapp_run path calls ha_start_election_async with
* n_peers=1 for a single-node cluster. This test verifies the call stores
* n_peers in ha->n_peers, matching the new myapp_run contract. */
ha_state_t ha;
ha_init(&ha, &g_startup_cfg, 0);
ha.n_peers = 0;
ha.phase2_done = 0;
ha_start_election_async(&ha, NULL, 1, NULL, 0ULL, 0ULL, "ctrl-a");
TEST_ASSERT_EQUAL_INT_MESSAGE(1, ha.n_peers,
"myapp_run path: ha_start_election_async must store n_peers=1 for single-node");
ha_fini(&ha);
printf("test_nonblocking_single_node_election_sets_n_peers: PASS\n");
}
static void test_nonblocking_single_node_election_enters_committing(void)
{
/* Req 9.3, 8.1: After ha_start_election_async(n_peers=1) the role must be
* CTRL_PRIMARY_COMMITTING, NOT CTRL_PRIMARY.
* This validates that the new myapp_run path (without blocking for(;;) loop)
* correctly transitions through the COMMITTING state. */
ha_state_t ha;
ha_init(&ha, &g_startup_cfg, 0);
ha.phase2_done = 0;
atomic_store_explicit(&ha.role, (int32_t)CTRL_ELECTING, memory_order_release);
ha_start_election_async(&ha, NULL, 1, NULL, 100ULL, 50ULL, "ctrl-a");
ctrl_role_t role = (ctrl_role_t)atomic_load_explicit(
&ha.role, memory_order_acquire);
TEST_ASSERT_EQUAL_INT_MESSAGE((int)CTRL_PRIMARY_COMMITTING, (int)role,
"myapp_run path: single-node must enter CTRL_PRIMARY_COMMITTING after election");
ha_fini(&ha);
printf("test_nonblocking_single_node_election_enters_committing: PASS (role=%d)\n",
(int)role);
}
static void test_nonblocking_election_returns_in_under_10ms(void)
{
/* Req 9.4, 1.1: ha_start_election_async must return in < 5ms (design spec).
* Observable: myapp_run exits into cluster_run within 5ms of startup;
* no blocking calls visible between ha_start_election_async and cluster_run. */
ha_state_t ha;
ha_init(&ha, &g_startup_cfg, 0);
ha.phase2_done = 0;
struct timespec t0;
clock_gettime(CLOCK_MONOTONIC, &t0);
ha_start_election_async(&ha, NULL, 1, NULL, 0ULL, 0ULL, "ctrl-a");
struct timespec t1;
clock_gettime(CLOCK_MONOTONIC, &t1);
long elapsed_ms_val = (t1.tv_sec - t0.tv_sec) * 1000L
+ (t1.tv_nsec - t0.tv_nsec) / 1000000L;
TEST_ASSERT_LESS_OR_EQUAL_INT_MESSAGE(5, (int)elapsed_ms_val,
"myapp_run path: ha_start_election_async must return in < 5ms (design spec)");
ha_fini(&ha);
printf("test_nonblocking_election_returns_in_under_10ms: PASS (%ldms)\n",
elapsed_ms_val);
}
/* ── Test suite entry point ───────────────────────────────────────────── */
void test_ha_startup(void)
@ -634,4 +739,9 @@ void test_ha_startup(void)
/* Task 3.1 — dispatch gate: CTRL_PRIMARY_COMMITTING */
RUN_TEST(test_dispatch_gate_committing_denies_job_ops);
RUN_TEST(test_dispatch_gate_committing_allows_repl_ops);
/* Task 5.1 — myapp_run non-blocking startup sequence (Req 1.1, 9.3, 9.4) */
RUN_TEST(test_nonblocking_single_node_election_sets_n_peers);
RUN_TEST(test_nonblocking_single_node_election_enters_committing);
RUN_TEST(test_nonblocking_election_returns_in_under_10ms);
}