lei.chen/myxce
1
0
Fork 1
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(election-iomgr-refactor): remove old blocking election code (am_i_primary_quorum, vote_request_thread, bootstrap_config_if_needed) — task 5.3

Browse source
This commit is contained in:
Rongsong Shen 2026-05-30 13:04:19 +00:00
parent 31a54043d5
commit 27ea14cd9a
7 changed files with 83 additions and 512 deletions
Download patch file Download diff file Expand all files Collapse all files

377
src/controller/ha.c
View file

@ -1984,383 +1984,6 @@ void ha_start_election_async(ha_state_t *ha, cluster_t *c,
", n_peers=%d, connected=%d)", cur_term, n_peers, peers_connected);
/* Returns immediately — election result delivered via ha_elect_evaluate_and_signal */
}
/* ── Per-peer vote-request thread ──────────────────────────────────────────
*
* vote_thread_ctx_t: arguments passed to each vote_request_thread().
* One struct per peer; allocated on the stack of am_i_primary_quorum.
*/
typedef struct {
ha_state_t *ha; /* Candidate's HA state (term, ssl_ctx) */
ha_elect_ctx_t *ctx; /* Shared election context */
xce_candidate_t *peer; /* Target peer */
int peer_idx; /* Index in ctx->votes[] */
uint64_t my_term; /* Term sent in the request */
uint64_t my_tag; /* config_tag sent in the request */
uint64_t my_lsn; /* WAL LSN sent in the request */
int timeout_ms; /* Connect + I/O timeout */
const char *my_name; /* Candidate node name for voted_for deduplication */
} vote_thread_ctx_t;
/*
* vote_request_thread: per-peer pthread that sends OP_requestVote and waits
* for a RequestVoteReply.
*
* Wire format (raw SSL, no iomgr):
* Send: uint64_t size_hdr + capnp(RequestVoteRequest)
* Receive: uint64_t size_hdr + capnp(RequestVoteReply)
*
* On any error (connect, encode, I/O, decode) the peer counts as denied with
* zero peer_tag/peer_lsn — the quorum evaluator sees an absent peer as a no-vote.
*/
static void *vote_request_thread(void *arg)
{
vote_thread_ctx_t *vctx = (vote_thread_ctx_t *)arg;
ha_state_t *ha = vctx->ha;
ha_elect_ctx_t *ctx = vctx->ctx;
int idx = vctx->peer_idx;
SSL *ssl = NULL;
int sock = -1;
/* ── 1. TCP connect (non-blocking poll loop) ─────────── */
struct addrinfo hints, *res = NULL;
memset(&hints, 0, sizeof(hints));
hints.ai_family = AF_UNSPEC;
hints.ai_socktype = SOCK_STREAM;
char port_str[16];
snprintf(port_str, sizeof(port_str), "%d", vctx->peer->port);
if (getaddrinfo(vctx->peer->hostname, port_str, &hints, &res) != 0)
goto deny;
sock = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
if (sock < 0) { freeaddrinfo(res); sock = -1; goto deny; }
{
int flags = fcntl(sock, F_GETFL, 0);
fcntl(sock, F_SETFL, flags | O_NONBLOCK);
connect(sock, res->ai_addr, res->ai_addrlen);
freeaddrinfo(res); res = NULL;
struct pollfd pfd = { .fd = sock, .events = POLLOUT | POLLERR | POLLHUP };
int pr;
do { pr = poll(&pfd, 1, vctx->timeout_ms); } while (pr < 0 && errno == EINTR);
if (pr <= 0 || (pfd.revents & (POLLERR | POLLHUP))) goto deny;
int err = 0; socklen_t el = sizeof(err);
getsockopt(sock, SOL_SOCKET, SO_ERROR, &err, &el);
if (err != 0) goto deny;
/* Restore blocking for SSL handshake */
fcntl(sock, F_SETFL, flags & ~O_NONBLOCK);
}
/* ── 2. SSL handshake (client side) ─────────────────── */
if (ha->ssl_ctx) {
/* ha->ssl_ctx was created with TLS_server_method (for incoming connections).
* SSL_connect requires a client-method context, so create one on-demand. */
SSL_CTX *client_ctx = SSL_CTX_new(TLS_client_method());
if (!client_ctx) goto deny;
SSL_CTX_set_verify(client_ctx, SSL_VERIFY_NONE, NULL);
ssl = SSL_new(client_ctx);
SSL_CTX_free(client_ctx); /* ssl holds a reference; safe to free ctx */
if (!ssl) goto deny;
SSL_set_fd(ssl, sock);
int sr;
do {
sr = SSL_connect(ssl);
if (sr <= 0) {
int e = SSL_get_error(ssl, sr);
if (e != SSL_ERROR_WANT_READ && e != SSL_ERROR_WANT_WRITE) goto deny;
}
} while (sr <= 0);
if (sr <= 0) goto deny;
}
/* ── 3. Encode + send RequestVoteRequest wrapped in a Request envelope ── */
{
request_vote_req_t vote_req;
memset(&vote_req, 0, sizeof(vote_req));
vote_req.term = vctx->my_term;
vote_req.candidate_id = (char *)(vctx->my_name ? vctx->my_name : "");
vote_req.config_tag = vctx->my_tag;
vote_req.last_log_lsn = vctx->my_lsn;
xce_request_t xreq;
memset(&xreq, 0, sizeof(xreq));
xreq.opcode = OPCode_requestVote;
xreq.body_which = Request_body_requestvotemsg;
xreq.body.requestvotemsg = &vote_req;
io_packet_t pkt;
memset(&pkt, 0, sizeof(pkt));
if (xce_encode(&xreq, &pkt) < 0)
goto deny;
uint64_t hdr = (uint64_t)pkt.sz;
size_t body_len = pkt.sz;
int nsent;
if (ssl) {
nsent = SSL_write(ssl, &hdr, (int)sizeof(hdr));
if (nsent == (int)sizeof(hdr))
nsent = SSL_write(ssl, pkt.data, (int)body_len);
} else {
nsent = (int)write(sock, &hdr, sizeof(hdr));
if (nsent == (int)sizeof(hdr))
nsent = (int)write(sock, pkt.data, body_len);
}
free_packet(&pkt);
if (nsent != (int)body_len) goto deny;
}
/* ── 4. Read + decode RequestReply containing RequestVoteReply ─────── */
{
uint64_t hdr = 0;
size_t off = 0;
/* Read the 8-byte size prefix */
while (off < sizeof(hdr)) {
int n;
if (ssl)
n = SSL_read(ssl, (char *)&hdr + off, (int)(sizeof(hdr) - off));
else
n = (int)read(sock, (char *)&hdr + off, sizeof(hdr) - off);
if (n <= 0) goto deny;
off += (size_t)n;
}
if (hdr == 0 || hdr > 65536) goto deny; /* sanity cap */
uint8_t *buf = malloc((size_t)hdr);
if (!buf) goto deny;
off = 0;
while (off < (size_t)hdr) {
int n;
if (ssl)
n = SSL_read(ssl, (char *)buf + off, (int)((size_t)hdr - off));
else
n = (int)read(sock, (char *)buf + off, (size_t)hdr - off);
if (n <= 0) { free(buf); goto deny; }
off += (size_t)n;
}
/* Unwrap outer Request{opcode=reply, body.replymsg=RequestReply{...}} */
struct capn capn_ctx;
struct Request outer_rq;
Request_ptr outer_rp;
struct RequestReply rep_s;
if (capn_init_mem(&capn_ctx, buf, (size_t)hdr, 0) != 0) {
free(buf); goto deny;
}
outer_rp.p = capn_getp(capn_root(&capn_ctx), 0, 1);
read_Request(&outer_rq, outer_rp);
if (outer_rq.body_which != Request_body_replymsg ||
!outer_rq.body.replymsg.p.data) {
capn_free(&capn_ctx); free(buf); goto deny;
}
read_RequestReply(&rep_s, outer_rq.body.replymsg);
capn_free(&capn_ctx);
free(buf);
if (rep_s.data_which != RequestReply_data_requestvotereplymsg)
goto deny;
struct RequestVoteReply vr_raw;
read_RequestVoteReply(&vr_raw, rep_s.data.requestvotereplymsg);
/* Fencing: if peer's term > ours, step down immediately */
if (vr_raw.term > vctx->my_term)
atomic_store_explicit(&ha->term, vr_raw.term, memory_order_release);
/* Record the reply */
handle_request_vote_reply(ha, ctx, idx,
vr_raw.term,
vr_raw.voteGranted ? 1 : 0,
vr_raw.peerConfigTag,
vr_raw.peerLastLogLsn);
}
goto cleanup;
deny:
/* Treat unreachable / error peer as denied vote with zero freshness */
handle_request_vote_reply(ha, ctx, idx,
vctx->my_term, /* use our term so no fencing */
0, /* vote_granted = false */
0, 0);
cleanup:
if (ssl) { SSL_shutdown(ssl); SSL_free(ssl); }
if (sock >= 0) close(sock);
return NULL;
}
/* ── Top-Level Election Entrypoint (Req 2.1, 2.9, 2.10, 2.11) ──────────────
*
* am_i_primary_quorum: top-level quorum election.
*
* For single-node clusters (n_peers == 1): self-elect immediately (Req 2.9).
* For multi-node clusters: increment term, allocate ha_elect_ctx_t, spawn
* per-peer pthreads to send OP_requestVote, block on ctx->done, evaluate result.
* Split vote → one retry with incremented term and randomized jitter (Req 2.10).
* Second failure → CTRL_STANDBY (Req 2.11).
*
* Unit tests: peers==NULL and n_peers==1 exercises the single-node shortcircuit.
* Multi-peer network path (peers!=NULL) is exercised by integration tests.
*/
ctrl_role_t am_i_primary_quorum(ha_state_t *ha,
int n_peers,
xce_candidate_t **peers,
uint64_t my_config_tag,
uint64_t my_lsn,
const char *my_name)
{
/* Req 2.9: single-node self-elects immediately */
if (n_peers == 1) {
atomic_fetch_add_explicit(&ha->term, 1, memory_order_acq_rel);
uint64_t term = atomic_load_explicit(&ha->term, memory_order_acquire);
INFO("HA election: outcome=PRIMARY term=%" PRIu64
" votes_granted=1 quorum=1 (single-node)", term);
INFO("HA FSM: CTRL_INIT → CTRL_PRIMARY (single-node, term=%" PRIu64 ")", term);
return CTRL_PRIMARY;
}
/* Multi-node path: up to 2 election rounds (Req 2.10, 2.11) */
int n_other = n_peers - 1; /* number of peers excluding self */
int total_nodes = n_peers;
int quorum = (total_nodes + 1) / 2;
for (int round = 0; round < 2; round++) {
/* Increment term before each election round */
atomic_fetch_add_explicit(&ha->term, 1, memory_order_acq_rel);
uint64_t cur_term = atomic_load_explicit(&ha->term, memory_order_acquire);
/* Req 10.5: log FSM transition to CTRL_ELECTING */
INFO("HA FSM: → CTRL_ELECTING (round=%d, term=%" PRIu64 ", n_peers=%d)",
round + 1, cur_term, n_peers);
atomic_store_explicit(&ha->role, (int32_t)CTRL_ELECTING, memory_order_release);
/* Add jitter delay between rounds (Req 2.10) */
if (round > 0 && ha->config.election_jitter_max_ms > 0) {
if (ha->jitter_seed == 0)
ha_seed_jitter(&ha->jitter_seed, ha->my_priority);
long jitter_ms = (long)(xorshift64_next(&ha->jitter_seed)
% ((uint64_t)ha->config.election_jitter_max_ms + 1));
if (jitter_ms > 0) {
struct timespec ts = {
.tv_sec = jitter_ms / 1000,
.tv_nsec = (jitter_ms % 1000) * 1000000L,
};
nanosleep(&ts, NULL);
}
}
/* Allocate election context */
ha_elect_ctx_t *ctx = calloc(1, sizeof(*ctx) +
(size_t)n_other * sizeof(ctx->votes[0]));
if (!ctx)
return CTRL_STANDBY;
ctx->n_peers = n_other;
ctx->votes_granted = 1; /* self-vote */
ctx->n_replied = 0;
ctx->max_resp_tag = 0;
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);
/* Store candidate's own tag/lsn in votes[0] for evaluate function */
if (n_other > 0) {
ctx->votes[0].my_tag = my_config_tag;
ctx->votes[0].my_lsn = my_lsn;
}
/* Spawn per-peer vote-request threads when peers list is provided.
* When peers==NULL (unit test mode), skip spawning and evaluate
* immediately with only the self-vote. */
vote_thread_ctx_t *vctxs = NULL;
pthread_t *tids = NULL;
if (peers != NULL && n_other > 0) {
vctxs = calloc((size_t)n_other, sizeof(*vctxs));
tids = calloc((size_t)n_other, sizeof(*tids));
if (!vctxs || !tids) {
free(vctxs); free(tids);
pthread_mutex_destroy(&ctx->lock);
sem_destroy(&ctx->done);
free(ctx);
return CTRL_STANDBY;
}
for (int i = 0; i < n_other; i++) {
vctxs[i].ha = ha;
vctxs[i].ctx = ctx;
vctxs[i].peer = peers[i];
vctxs[i].peer_idx = i;
vctxs[i].my_term = cur_term;
vctxs[i].my_tag = my_config_tag;
vctxs[i].my_lsn = my_lsn;
vctxs[i].timeout_ms =
(int)ha->config.probe_connect_timeout_ms;
vctxs[i].my_name = my_name;
pthread_create(&tids[i], NULL, vote_request_thread, &vctxs[i]);
}
/* Block until quorum or all peers replied */
sem_wait(&ctx->done);
/* Join all threads before freeing ctx */
for (int i = 0; i < n_other; i++)
pthread_join(tids[i], NULL);
free(vctxs);
free(tids);
}
ctrl_role_t role;
if (peers == NULL) {
/* Unit-test path: no network peers; evaluate with self-vote only */
role = ha_elect_evaluate_and_signal(ha, ctx);
} else {
/* Production path: ctx->result was set by ha_elect_evaluate_and_signal
* inside handle_request_vote_reply; just read it back. */
role = (ctx->result == 1) ? CTRL_PRIMARY : CTRL_STANDBY;
}
/* Capture before freeing ctx */
int votes_granted_snap = ctx->votes_granted;
/* Req 10.1: log election outcome */
INFO("HA election: outcome=%s term=%" PRIu64
" votes_granted=%d quorum=%d (round=%d)",
(role == CTRL_PRIMARY) ? "PRIMARY" : "STANDBY",
cur_term, votes_granted_snap, quorum, round + 1);
/* Req 10.5: FSM log for multi-node CTRL_PRIMARY path.
* ha_do_promote() is not called here so its FSM log is absent; emit it now. */
if (role == CTRL_PRIMARY)
INFO("HA FSM: → CTRL_PRIMARY (term=%" PRIu64 " round=%d votes=%d quorum=%d)",
cur_term, round + 1, votes_granted_snap, quorum);
pthread_mutex_destroy(&ctx->lock);
sem_destroy(&ctx->done);
free(ctx);
if (role == CTRL_PRIMARY)
return CTRL_PRIMARY;
/* Split vote: retry once (Req 2.10); second failure → CTRL_STANDBY (Req 2.11) */
}
return CTRL_STANDBY;
}
/* ── Task 9.2: Persistent Control Channel ──────────────────────────────────
*
* ha_find_and_connect_to_primary: three-tier primary discovery + TCP connect.