forked from lei.chen/myxce
feat(raft-drive-async): offload delaymgr peer reconnect to slow lane (task 2.1)
Heartbeat timer detects a disconnected peer, CAS-acquires a per-peer single-flight flag, slowlane_submits the bounded connect, and returns — the delaymgr never blocks on connect. Lane worker (reconnect_lane_fn) does the bounded endpoint_connect + lock-guarded iomgr_monitor; flag cleared on all terminal paths (incl. -EEXIST race fix). Default-ON kill-switch retains the inline path for TDD-RED/fallback. Never self-terminates on failure. 2.2 adds the inbound/outbound race + failure hardening (documented seam). 196/0 green.
This commit is contained in:
parent
a36234a103
commit
e07667ff9b
4 changed files with 569 additions and 17 deletions
|
|
@ -3767,6 +3767,33 @@ int handle_request_vote(ha_state_t *ha,
|
|||
}
|
||||
#endif /* XCE_TESTING */
|
||||
|
||||
/* ── raft-drive-async Task 2.1: reconnect-offload feature flag (ALL builds) ──
|
||||
*
|
||||
* drive_async_reconnect_offload gates the peer reconnect path:
|
||||
* ON (production default) → primary_heartbeat_fn submits the peer reconnect
|
||||
* to the slow lane and returns immediately; the delaymgr thread NEVER
|
||||
* performs the blocking connect inline (ha-async-io.md thread-role
|
||||
* contract — a blocking connect on the heartbeat/election-timeout thread
|
||||
* stalls liveness to healthy peers and triggers spurious elections).
|
||||
* OFF → legacy inline blocking connect on the delaymgr thread; retained only
|
||||
* so the TDD RED test can demonstrate the pre-change stall.
|
||||
*
|
||||
* Compiled in ALL builds (production reads it on every heartbeat tick); the
|
||||
* setter is the test seam to flip RED↔GREEN. */
|
||||
static _Atomic int g_drive_async_reconnect_offload = 1;
|
||||
|
||||
void ha_set_reconnect_offload_for_test(int on)
|
||||
{
|
||||
atomic_store_explicit(&g_drive_async_reconnect_offload, on ? 1 : 0,
|
||||
memory_order_release);
|
||||
}
|
||||
|
||||
int ha_reconnect_offload_enabled(void)
|
||||
{
|
||||
return atomic_load_explicit(&g_drive_async_reconnect_offload,
|
||||
memory_order_acquire);
|
||||
}
|
||||
|
||||
/* ── Test-only hook shims (XCE_TESTING only) ──────────────────────────────
|
||||
* Type definitions (election_timeout_ctx_t), extern declarations
|
||||
* (cluster_init_phase2, push_config_transport_only,
|
||||
|
|
@ -3816,6 +3843,16 @@ int (*ha_heartbeat_req_hook)(endpoint_t *ep, uint64_t term,
|
|||
uint64_t committed_lsn,
|
||||
uint64_t committed_config_tag) = NULL;
|
||||
|
||||
/*
|
||||
* ha_reconnect_connect_hook: test-only seam for the BLOCKING half of the
|
||||
* offloaded peer reconnect (raft-drive-async Task 2.1). When non-NULL, the
|
||||
* slow-lane reconnect worker calls this INSTEAD OF the real bounded
|
||||
* endpoint_connect + iomgr_monitor, so a test can drive the reconnect path
|
||||
* deterministically (e.g. park the worker to simulate an unreachable peer).
|
||||
* Returns 0 to simulate success, <0 to simulate failure. Req 2.1, 2.2, 2.5.
|
||||
*/
|
||||
int (*ha_reconnect_connect_hook)(cluster_t *c, int peer_idx) = NULL;
|
||||
|
||||
/*
|
||||
* ha_repl_push_config_for_standby_hook: test-only seam for the per-standby
|
||||
* config push triggered by the heartbeat-reply decision loop (Task 3.3).
|
||||
|
|
@ -6406,24 +6443,43 @@ static int standby_ep_err_handler_primary(void *ep_raw)
|
|||
}
|
||||
|
||||
/*
|
||||
* ha_connect_to_peer: open a new outbound TLS connection to a peer candidate
|
||||
* and register it as a standby channel. Called from primary_heartbeat_fn
|
||||
* for peers that are absent from standby_endpoints.
|
||||
* reconnect_ctx_t: owns the prepared endpoint across the in-flight reconnect
|
||||
* window (raft-drive-async Task 2.1). Built on the delaymgr tick by the
|
||||
* non-blocking prep (ha_connect_prep_peer), consumed once on a slow-lane
|
||||
* worker by reconnect_lane_fn, then freed. The ctx keeps `ep`/`sctx` valid
|
||||
* for the whole window so the lane worker never touches freed memory while the
|
||||
* delaymgr thread races ahead to the next tick (Req 2.3 lifetime).
|
||||
*/
|
||||
static void ha_connect_to_peer(cluster_t *c, int peer_idx)
|
||||
typedef struct {
|
||||
cluster_t *c;
|
||||
int peer_idx; /* ABSOLUTE candidate index */
|
||||
endpoint_t *ep; /* prepared, not yet connected */
|
||||
standby_ep_ctx_t *sctx; /* == ep->data */
|
||||
int should_log;
|
||||
} reconnect_ctx_t;
|
||||
|
||||
/*
|
||||
* ha_connect_prep_peer: the NON-BLOCKING half of the outbound peer connect.
|
||||
* Allocates the endpoint context, initialises the endpoint, sets the SSL
|
||||
* client state and the on-connected / err handlers — all of which are
|
||||
* non-blocking and safe to run on the delaymgr tick. Does NOT perform the
|
||||
* blocking connect (that is ha_connect_finish_peer, run on the slow lane).
|
||||
*
|
||||
* Returns a heap reconnect_ctx_t* (caller owns) on success, or NULL on an
|
||||
* allocation failure (nothing leaked).
|
||||
*/
|
||||
static reconnect_ctx_t *ha_connect_prep_peer(cluster_t *c, int peer_idx)
|
||||
{
|
||||
ha_state_t *ha = &c->ha;
|
||||
const xce_candidate_t *peer =
|
||||
&c->myconf->peers_cfg.candidates[peer_idx];
|
||||
ha_state_t *ha = &c->ha;
|
||||
|
||||
standby_ep_ctx_t *sctx = calloc(1, sizeof(*sctx));
|
||||
if (!sctx) return;
|
||||
if (!sctx) return NULL;
|
||||
sctx->ha = ha;
|
||||
sctx->c = c;
|
||||
sctx->peer_idx = peer_idx;
|
||||
|
||||
endpoint_t *ep = endpoint_get();
|
||||
if (!ep) { free(sctx); return; }
|
||||
if (!ep) { free(sctx); return NULL; }
|
||||
|
||||
endpoint_init(ep, AF_INET, SOCK_STREAM, 0, 0);
|
||||
endpoint_set_peer(ep, c->myhost);
|
||||
|
|
@ -6443,14 +6499,74 @@ static void ha_connect_to_peer(cluster_t *c, int peer_idx)
|
|||
int n = ++ha->peer_connect_fail_count[peer_idx];
|
||||
int should_log = (n == 1) || (n % HA_CONNECT_LOG_THROTTLE == 0);
|
||||
|
||||
if (should_log)
|
||||
INFO("HA: peer[%d] (%s) not connected — initiating outbound connect (attempt %d)",
|
||||
peer_idx, peer->hostname, n);
|
||||
reconnect_ctx_t *rc_ctx = calloc(1, sizeof(*rc_ctx));
|
||||
if (!rc_ctx) {
|
||||
ep->data = NULL;
|
||||
endpoint_fini(ep);
|
||||
free(sctx);
|
||||
return NULL;
|
||||
}
|
||||
rc_ctx->c = c;
|
||||
rc_ctx->peer_idx = peer_idx;
|
||||
rc_ctx->ep = ep;
|
||||
rc_ctx->sctx = sctx;
|
||||
rc_ctx->should_log = should_log;
|
||||
return rc_ctx;
|
||||
}
|
||||
|
||||
/*
|
||||
* ha_connect_finish_peer: the BLOCKING half of the outbound peer connect —
|
||||
* the bounded connect + lock-guarded iomgr registration. MUST run on a
|
||||
* slow-lane worker, NEVER on the delaymgr tick (Req 2.1, 2.2; ha-async-io.md
|
||||
* thread-role contract).
|
||||
*
|
||||
* On success the endpoint is connecting and registered with iomgr; on_connected
|
||||
* (on_standby_connected) fires later from the iomgr loop. On failure the
|
||||
* endpoint is torn down here and the controller simply retries on the next
|
||||
* heartbeat tick (controllers never self-terminate).
|
||||
*
|
||||
* Returns 0 on a successful connect+register, <0 otherwise.
|
||||
*
|
||||
* NOTE (Task 2.2 seam): this implements the happy-path connect+register +
|
||||
* teardown-on-failure. The inbound-vs-outbound connection-ownership race
|
||||
* resolution, the no-lost/no-reordered-message guarantee on success, and the
|
||||
* richer failure/retry hardening are owned by Task 2.2 — they layer on top of
|
||||
* this happy path without changing this function's contract (return 0/<0,
|
||||
* caller clears the per-peer in-flight flag).
|
||||
*/
|
||||
static int ha_connect_finish_peer(reconnect_ctx_t *rc_ctx)
|
||||
{
|
||||
cluster_t *c = rc_ctx->c;
|
||||
endpoint_t *ep = rc_ctx->ep;
|
||||
standby_ep_ctx_t *sctx = rc_ctx->sctx;
|
||||
int peer_idx = rc_ctx->peer_idx;
|
||||
const xce_candidate_t *peer = &c->myconf->peers_cfg.candidates[peer_idx];
|
||||
|
||||
#ifdef XCE_TESTING
|
||||
if (ha_reconnect_connect_hook) {
|
||||
/* Deterministic test path: the hook stands in for the real bounded
|
||||
* connect + register so a test can park the worker (unreachable peer)
|
||||
* or return a synthetic success/failure without a live socket. On
|
||||
* the test path we do NOT touch the real endpoint fd. */
|
||||
int hrc = ha_reconnect_connect_hook(c, peer_idx);
|
||||
if (hrc < 0) {
|
||||
ep->data = NULL;
|
||||
endpoint_fini(ep);
|
||||
free(sctx);
|
||||
}
|
||||
return hrc;
|
||||
}
|
||||
#endif
|
||||
|
||||
if (rc_ctx->should_log)
|
||||
INFO("HA: peer[%d] (%s) not connected — initiating outbound connect "
|
||||
"(attempt %d)", peer_idx, peer->hostname,
|
||||
c->ha.peer_connect_fail_count[peer_idx]);
|
||||
|
||||
int rc = endpoint_connect(ep, peer->hostname, peer->port,
|
||||
DEFAULT_CONNECT_TIMEOUT_MS);
|
||||
if (rc < 0) {
|
||||
if (should_log)
|
||||
if (rc_ctx->should_log)
|
||||
WARN("HA: connect to peer[%d] (%s:%u) failed: %d",
|
||||
peer_idx, peer->hostname, peer->port, rc);
|
||||
/* endpoint_fini calls FREEUP(ep->data) which frees sctx — do not
|
||||
|
|
@ -6458,17 +6574,130 @@ static void ha_connect_to_peer(cluster_t *c, int peer_idx)
|
|||
ep->data = NULL;
|
||||
endpoint_fini(ep);
|
||||
free(sctx);
|
||||
return;
|
||||
return rc;
|
||||
}
|
||||
rc = iomgr_monitor(ep);
|
||||
rc = iomgr_monitor(ep); /* lock-guarded epoll registration (Req 4.4) */
|
||||
if (rc < 0) {
|
||||
ep->data = NULL;
|
||||
endpoint_fini(ep);
|
||||
free(sctx);
|
||||
return;
|
||||
return rc;
|
||||
}
|
||||
INFO("HA: connecting outbound to peer[%d] (%s:%u)",
|
||||
peer_idx, peer->hostname, peer->port);
|
||||
return 0;
|
||||
}
|
||||
|
||||
/*
|
||||
* reconnect_lane_fn: the slow-lane worker entry (raft-drive-async Task 2.1).
|
||||
* Runs the bounded blocking connect off the delaymgr thread, then ALWAYS
|
||||
* clears the per-peer in-flight flag (success or failure) so the dedup never
|
||||
* wedges and the next heartbeat tick can retry an unreachable peer (Req 2.5,
|
||||
* 2.4 — never self-terminates). Frees the reconnect ctx.
|
||||
*/
|
||||
static void reconnect_lane_fn(void *arg)
|
||||
{
|
||||
reconnect_ctx_t *rc_ctx = (reconnect_ctx_t *)arg;
|
||||
if (!rc_ctx) return;
|
||||
|
||||
cluster_t *c = rc_ctx->c;
|
||||
int peer_idx = rc_ctx->peer_idx;
|
||||
|
||||
(void)ha_connect_finish_peer(rc_ctx);
|
||||
|
||||
/* Clear in-flight LAST, after the endpoint is fully connected/registered
|
||||
* or torn down — so a concurrent heartbeat tick cannot enqueue a second
|
||||
* reconnect that races this one's endpoint setup (Req 2.5). */
|
||||
atomic_store_explicit(&c->ha.reconnect_inflight[peer_idx], 0,
|
||||
memory_order_release);
|
||||
free(rc_ctx);
|
||||
}
|
||||
|
||||
/*
|
||||
* ha_connect_to_peer: open a new outbound TLS connection to a peer candidate
|
||||
* and register it as a standby channel. Called from primary_heartbeat_fn for
|
||||
* peers that are absent from standby_endpoints.
|
||||
*
|
||||
* Offload flag ON (default, Task 2.1): prepares the endpoint (non-blocking)
|
||||
* and submits the bounded connect to the slow lane with a per-peer dedup key,
|
||||
* then returns immediately — the delaymgr thread NEVER blocks on connect.
|
||||
*
|
||||
* Offload flag OFF (legacy / TDD RED only): runs the blocking connect inline
|
||||
* on the calling (delaymgr) thread. Retained solely so the RED test can show
|
||||
* the pre-change stall; production runs with the flag ON.
|
||||
*/
|
||||
static void ha_connect_to_peer(cluster_t *c, int peer_idx)
|
||||
{
|
||||
ha_state_t *ha = &c->ha;
|
||||
|
||||
if (!ha_reconnect_offload_enabled()) {
|
||||
/* ── Legacy inline path (flag OFF) — blocks the caller. ── */
|
||||
reconnect_ctx_t *rc_ctx = ha_connect_prep_peer(c, peer_idx);
|
||||
if (!rc_ctx) return;
|
||||
(void)ha_connect_finish_peer(rc_ctx);
|
||||
free(rc_ctx);
|
||||
return;
|
||||
}
|
||||
|
||||
/* ── Offload path (flag ON) — non-blocking on the delaymgr thread. ──
|
||||
*
|
||||
* Per-peer single-flight: CAS the in-flight flag 0->1. If it was already
|
||||
* 1, a reconnect to this peer is already in flight on the lane — enqueue
|
||||
* nothing (Req 2.5). This guard is the authoritative dedup; the lane's
|
||||
* own per-key dedup is a redundant backstop. */
|
||||
int expected = 0;
|
||||
if (!atomic_compare_exchange_strong_explicit(
|
||||
&ha->reconnect_inflight[peer_idx], &expected, 1,
|
||||
memory_order_acq_rel, memory_order_acquire)) {
|
||||
return; /* already reconnecting this peer */
|
||||
}
|
||||
|
||||
reconnect_ctx_t *rc_ctx = ha_connect_prep_peer(c, peer_idx);
|
||||
if (!rc_ctx) {
|
||||
/* prep failed (alloc) — release the in-flight guard so the next tick
|
||||
* retries; controller stays alive (Req 2.4). */
|
||||
atomic_store_explicit(&ha->reconnect_inflight[peer_idx], 0,
|
||||
memory_order_release);
|
||||
return;
|
||||
}
|
||||
|
||||
/* Submit the blocking connect to the slow lane. Dedup key is the
|
||||
* per-peer absolute index (+1 so it is non-zero; key 0 disables dedup). */
|
||||
uint64_t key = (uint64_t)(peer_idx + 1);
|
||||
int srv = slowlane_submit(&c->slowlane, key, reconnect_lane_fn, rc_ctx);
|
||||
if (srv == -EEXIST) {
|
||||
/* Lane still holds this peer's dedup key from a PRIOR reconnect task.
|
||||
* This is the narrow window where that task already cleared our
|
||||
* reconnect_inflight flag (its last action) but the lane has not yet
|
||||
* removed its key. Our CAS therefore succeeded (flag was 0) even
|
||||
* though the lane refuses the submit. We MUST release the flag we just
|
||||
* set — otherwise no task owns its clear and dedup wedges permanently.
|
||||
* Tear down our duplicate prep; the next tick retries cleanly. */
|
||||
rc_ctx->ep->data = NULL;
|
||||
endpoint_fini(rc_ctx->ep);
|
||||
free(rc_ctx->sctx);
|
||||
free(rc_ctx);
|
||||
atomic_store_explicit(&ha->reconnect_inflight[peer_idx], 0,
|
||||
memory_order_release);
|
||||
return;
|
||||
}
|
||||
if (srv < 0 && srv != -EAGAIN) {
|
||||
/* Hard submit failure (alloc / lane stopping). -EAGAIN is NOT a
|
||||
* failure: the task is still queued and will run, so we keep the flag
|
||||
* set and let reconnect_lane_fn clear it. Only on a true failure do
|
||||
* we tear down and release the guard so the next tick retries. */
|
||||
WARN("HA: slowlane_submit reconnect peer[%d] failed: %d — retry next tick",
|
||||
peer_idx, srv);
|
||||
rc_ctx->ep->data = NULL;
|
||||
endpoint_fini(rc_ctx->ep);
|
||||
free(rc_ctx->sctx);
|
||||
free(rc_ctx);
|
||||
atomic_store_explicit(&ha->reconnect_inflight[peer_idx], 0,
|
||||
memory_order_release);
|
||||
return;
|
||||
}
|
||||
/* 0 or -EAGAIN: the task is enqueued and will run exactly once on a lane
|
||||
* worker; reconnect_lane_fn clears the in-flight flag on completion. */
|
||||
}
|
||||
|
||||
/*
|
||||
|
|
|
|||
|
|
@ -254,6 +254,19 @@ typedef struct {
|
|||
* cleared and the config is activated.
|
||||
* NEVER self-terminates (loud non-serving
|
||||
* fence only). 0 at init (ha_init memset). */
|
||||
/* ── New field added by raft-drive-async spec (Task 2.1) ──────── */
|
||||
_Atomic int reconnect_inflight[MAX_HA_CANDIDATES]; /* Per-peer
|
||||
* single-flight guard for the offloaded
|
||||
* delaymgr reconnect (Req 2.5): set (CAS
|
||||
* 0->1) when a reconnect to peer[i] is
|
||||
* submitted to the slow lane, cleared by
|
||||
* the lane worker on completion (success
|
||||
* or failure) so the next heartbeat tick
|
||||
* can retry. While set, the heartbeat
|
||||
* reconnect scan enqueues no duplicate for
|
||||
* that peer. Indexed by ABSOLUTE candidate
|
||||
* index (peers_cfg.candidates[]). 0 at
|
||||
* init (ha_init memset). */
|
||||
} ha_state_t;
|
||||
|
||||
/* ── Public API ────────────────────────────────────────────── */
|
||||
|
|
@ -1196,6 +1209,18 @@ void ha_disarm_heartbeat_timeout_vtask(ha_state_t *ha);
|
|||
*/
|
||||
void ha_start_primary_heartbeat(cluster_t *c);
|
||||
|
||||
/*
|
||||
* ha_set_reconnect_offload_for_test / ha_reconnect_offload_enabled:
|
||||
* Feature-flag accessors (raft-drive-async Task 2.1). The offload flag gates
|
||||
* whether primary_heartbeat_fn submits the peer reconnect to the slow lane (ON,
|
||||
* the production default — the delaymgr thread never blocks on connect) or runs
|
||||
* the legacy inline blocking connect on the delaymgr thread (OFF, retained only
|
||||
* so the TDD RED test can demonstrate the pre-change stall). These accessors
|
||||
* are compiled in ALL builds; the setter is used by tests to flip the flag.
|
||||
*/
|
||||
void ha_set_reconnect_offload_for_test(int on);
|
||||
int ha_reconnect_offload_enabled(void);
|
||||
|
||||
/* ── Task 2.4: Non-blocking async election initiator ────────────────────── */
|
||||
|
||||
/*
|
||||
|
|
@ -1353,6 +1378,23 @@ extern int (*ha_heartbeat_req_hook)(endpoint_t *ep, uint64_t term,
|
|||
uint64_t committed_lsn,
|
||||
uint64_t committed_config_tag);
|
||||
|
||||
/*
|
||||
* ha_reconnect_connect_hook: test-only seam for the BLOCKING half of the
|
||||
* offloaded peer reconnect (raft-drive-async Task 2.1). The reconnect splits
|
||||
* into a non-blocking prep (alloc endpoint, set callbacks) on the delaymgr
|
||||
* tick and a blocking connect+monitor that runs on a slow-lane worker. When
|
||||
* non-NULL, the lane worker calls this INSTEAD OF the real bounded
|
||||
* endpoint_connect + iomgr_monitor, so a test can drive the reconnect path
|
||||
* deterministically (e.g. park the worker to simulate an unreachable peer)
|
||||
* without a real socket.
|
||||
*
|
||||
* Signature: int hook(cluster_t *c, int peer_idx)
|
||||
* returns 0 to simulate a successful connect, <0 to simulate failure.
|
||||
* The lane worker clears reconnect_inflight[peer_idx] after this returns
|
||||
* regardless of outcome, so dedup never wedges (Req 2.5).
|
||||
*/
|
||||
extern int (*ha_reconnect_connect_hook)(cluster_t *c, int peer_idx);
|
||||
|
||||
/*
|
||||
* ha_repl_push_config_for_standby_hook: test-only seam for the per-standby
|
||||
* config push triggered by the heartbeat-reply decision loop (Task 3.3).
|
||||
|
|
|
|||
|
|
@ -22,9 +22,12 @@
|
|||
#include <time.h>
|
||||
#include <unistd.h>
|
||||
|
||||
#include <pthread.h>
|
||||
|
||||
#include <unity/unity.h>
|
||||
#include "ha.h"
|
||||
#include "netio.h"
|
||||
#include "slowlane.h"
|
||||
#include "xce/cluster_def.h"
|
||||
|
||||
/* ── XCE_TESTING seam ────────────────────────────────────────────────────── */
|
||||
|
|
@ -842,6 +845,279 @@ static void test_ha_arm_does_not_increase_thread_count(void)
|
|||
printf("test_ha_arm_does_not_increase_thread_count: PASS\n");
|
||||
}
|
||||
|
||||
/* ── Tests 15–17: raft-drive-async Task 2.1 — reconnect offload ──────────────
|
||||
*
|
||||
* Acceptance criteria (Req 2.1, 2.2, 2.5; design ReconnectOffload trigger):
|
||||
* - the delaymgr heartbeat tick detects a disconnected peer and submits the
|
||||
* reconnect to the slow lane, then returns immediately — it NEVER performs
|
||||
* the blocking connect inline (flag OFF = old inline path, which stalls);
|
||||
* - per-peer dedup: while a reconnect to peer X is in flight, a second tick
|
||||
* enqueues no duplicate;
|
||||
* - heartbeats to reachable peers keep flowing while one peer is unreachable.
|
||||
*
|
||||
* Determinism: the blocking connect half is replaced by ha_reconnect_connect_hook,
|
||||
* which parks on a test-owned gate to simulate an unreachable peer. The test
|
||||
* controls exactly when the "connect" completes, so no real socket/timing is
|
||||
* involved. primary_heartbeat_fn_for_test is driven on a worker thread so the
|
||||
* test can assert it returns promptly (offload) vs stalls (inline). */
|
||||
|
||||
extern int (*ha_reconnect_connect_hook)(cluster_t *c, int peer_idx);
|
||||
|
||||
/* Gate that the reconnect hook parks on (simulates an unreachable peer). */
|
||||
typedef struct {
|
||||
pthread_mutex_t mu;
|
||||
pthread_cond_t release_cv; /* test broadcasts to release the parked hook */
|
||||
pthread_cond_t entered_cv; /* hook broadcasts when it has entered/parked */
|
||||
int gate_open;
|
||||
_Atomic int entered; /* hook invocations that reached the gate */
|
||||
_Atomic int calls; /* total hook invocations (dedup oracle) */
|
||||
} rc_gate_t;
|
||||
|
||||
static rc_gate_t g_rc_gate;
|
||||
|
||||
static void rc_gate_init(rc_gate_t *g)
|
||||
{
|
||||
pthread_mutex_init(&g->mu, NULL);
|
||||
pthread_cond_init(&g->release_cv, NULL);
|
||||
pthread_cond_init(&g->entered_cv, NULL);
|
||||
g->gate_open = 0;
|
||||
atomic_store(&g->entered, 0);
|
||||
atomic_store(&g->calls, 0);
|
||||
}
|
||||
|
||||
static void rc_gate_destroy(rc_gate_t *g)
|
||||
{
|
||||
pthread_mutex_destroy(&g->mu);
|
||||
pthread_cond_destroy(&g->release_cv);
|
||||
pthread_cond_destroy(&g->entered_cv);
|
||||
}
|
||||
|
||||
/* The hook stand-in for the blocking connect: count, signal entry, park until
|
||||
* the test opens the gate, then return success. Runs on a slow-lane worker
|
||||
* (flag ON) or inline on the caller (flag OFF). */
|
||||
static int rc_gated_connect_hook(cluster_t *c, int peer_idx)
|
||||
{
|
||||
(void)c; (void)peer_idx;
|
||||
atomic_fetch_add(&g_rc_gate.calls, 1);
|
||||
|
||||
pthread_mutex_lock(&g_rc_gate.mu);
|
||||
atomic_fetch_add(&g_rc_gate.entered, 1);
|
||||
pthread_cond_broadcast(&g_rc_gate.entered_cv);
|
||||
while (!g_rc_gate.gate_open)
|
||||
pthread_cond_wait(&g_rc_gate.release_cv, &g_rc_gate.mu);
|
||||
pthread_mutex_unlock(&g_rc_gate.mu);
|
||||
return 0; /* simulate a successful connect */
|
||||
}
|
||||
|
||||
static void rc_gate_open(rc_gate_t *g)
|
||||
{
|
||||
pthread_mutex_lock(&g->mu);
|
||||
g->gate_open = 1;
|
||||
pthread_cond_broadcast(&g->release_cv);
|
||||
pthread_mutex_unlock(&g->mu);
|
||||
}
|
||||
|
||||
/* Wait (bounded) until at least n hook invocations have parked at the gate. */
|
||||
static int rc_gate_wait_entered(rc_gate_t *g, int n, int timeout_ms)
|
||||
{
|
||||
struct timespec deadline;
|
||||
clock_gettime(CLOCK_REALTIME, &deadline);
|
||||
deadline.tv_sec += timeout_ms / 1000;
|
||||
deadline.tv_nsec += (long)(timeout_ms % 1000) * 1000000L;
|
||||
if (deadline.tv_nsec >= 1000000000L) { deadline.tv_sec++; deadline.tv_nsec -= 1000000000L; }
|
||||
|
||||
pthread_mutex_lock(&g->mu);
|
||||
int rc = 0;
|
||||
while (atomic_load(&g->entered) < n) {
|
||||
rc = pthread_cond_timedwait(&g->entered_cv, &g->mu, &deadline);
|
||||
if (rc != 0) break;
|
||||
}
|
||||
int entered = atomic_load(&g->entered);
|
||||
pthread_mutex_unlock(&g->mu);
|
||||
return entered >= n ? 0 : -1;
|
||||
}
|
||||
|
||||
/* Build a primary cluster_t with: peers_cfg of 3 candidates (self at priority
|
||||
* 0), one reachable standby already in standby_endpoints (peer 1), and peer 2
|
||||
* absent (so the reconnect scan targets it). The slow lane is initialised. */
|
||||
static void rc_setup_primary(cluster_t *c, myconf_t *conf, endpoint_t *standby_ep)
|
||||
{
|
||||
memset(c, 0, sizeof(*c));
|
||||
memset(conf, 0, sizeof(*conf));
|
||||
|
||||
init_ha_primary_with_standby_eps(&c->ha);
|
||||
|
||||
/* 3-node cluster; self = priority 0. */
|
||||
conf->peers_cfg.candidates_count = 3;
|
||||
for (int i = 0; i < 3; i++) {
|
||||
snprintf(conf->peers_cfg.candidates[i].name,
|
||||
sizeof(conf->peers_cfg.candidates[i].name), "ctrl%d", i);
|
||||
snprintf(conf->peers_cfg.candidates[i].hostname,
|
||||
sizeof(conf->peers_cfg.candidates[i].hostname), "127.0.0.1");
|
||||
conf->peers_cfg.candidates[i].port = 19000 + i;
|
||||
}
|
||||
c->myconf = conf;
|
||||
c->ha.my_priority = 0;
|
||||
c->myhost = &c->myhost_info;
|
||||
|
||||
/* Slow lane for the offload path. */
|
||||
TEST_ASSERT_EQUAL_INT(0, slowlane_init(&c->slowlane, 2, 4));
|
||||
|
||||
/* Reachable standby for peer 1 (absolute index 1), already connected. */
|
||||
memset(standby_ep, 0, sizeof(*standby_ep));
|
||||
standby_ep_ctx_build_for_test(standby_ep, &c->ha, c, 1);
|
||||
standby_ep_ctx_t_test *sctx = (standby_ep_ctx_t_test *)standby_ep->data;
|
||||
sctx->last_push_time = 0; /* stale → heartbeat will be sent */
|
||||
zarray_store(c->ha.standby_endpoints, standby_ep);
|
||||
}
|
||||
|
||||
static void rc_teardown_primary(cluster_t *c, endpoint_t *standby_ep)
|
||||
{
|
||||
slowlane_fini(&c->slowlane);
|
||||
free(standby_ep->data);
|
||||
fini_ha_primary_with_standby_eps(&c->ha);
|
||||
}
|
||||
|
||||
/* Worker thread that drives one heartbeat tick; sets *done=1 when it returns. */
|
||||
typedef struct { cluster_t *c; _Atomic int done; } rc_tick_arg_t;
|
||||
static void *rc_tick_thread(void *p)
|
||||
{
|
||||
rc_tick_arg_t *a = (rc_tick_arg_t *)p;
|
||||
primary_heartbeat_fn_for_test(a->c);
|
||||
atomic_store(&a->done, 1);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
/*
|
||||
* Test 15 (Req 2.1, 2.2): with the offload flag ON, a heartbeat tick whose
|
||||
* reconnect target is unreachable (the connect hook parks) RETURNS PROMPTLY —
|
||||
* the blocking connect runs on the lane, not on the delaymgr thread — and the
|
||||
* heartbeat to the reachable standby was still sent.
|
||||
*
|
||||
* With the flag OFF (RED) the inline connect runs on the tick thread and the
|
||||
* tick BLOCKS in the gate, so the same assertion (tick done within the
|
||||
* deadline) FAILS. The flag toggle is what flips RED↔GREEN.
|
||||
*/
|
||||
static void rc_run_offload_tick(int offload_on, const char *label)
|
||||
{
|
||||
cluster_t c;
|
||||
myconf_t conf;
|
||||
endpoint_t standby_ep;
|
||||
rc_setup_primary(&c, &conf, &standby_ep);
|
||||
|
||||
rc_gate_init(&g_rc_gate);
|
||||
ha_reconnect_connect_hook = rc_gated_connect_hook;
|
||||
ha_set_reconnect_offload_for_test(offload_on);
|
||||
|
||||
g_nop_calls = 0;
|
||||
ha_nop_req_hook = stub_nop_req;
|
||||
|
||||
rc_tick_arg_t arg = { .c = &c };
|
||||
atomic_store(&arg.done, 0);
|
||||
|
||||
pthread_t th;
|
||||
pthread_create(&th, NULL, rc_tick_thread, &arg);
|
||||
|
||||
/* The tick should return within a generous deadline IFF the connect was
|
||||
* offloaded. Poll arg.done. */
|
||||
int done = 0;
|
||||
for (int i = 0; i < 200; i++) { /* up to ~2s */
|
||||
if (atomic_load(&arg.done)) { done = 1; break; }
|
||||
struct timespec ts = { .tv_sec = 0, .tv_nsec = 10 * 1000000L };
|
||||
nanosleep(&ts, NULL);
|
||||
}
|
||||
|
||||
if (offload_on) {
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, done,
|
||||
"Req 2.1/2.2: tick must return promptly with reconnect offloaded "
|
||||
"(connect parked on the lane, delaymgr thread free)");
|
||||
/* The reachable standby still got its heartbeat this tick. */
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, g_nop_calls,
|
||||
"Req 2.1: heartbeat to the reachable standby must be sent while a "
|
||||
"peer is unreachable");
|
||||
/* The reconnect connect was submitted to the lane (hook entered). */
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc_gate_wait_entered(&g_rc_gate, 1, 1000),
|
||||
"Req 2.1: the offloaded reconnect connect must run on the lane");
|
||||
} else {
|
||||
/* RED: inline connect blocks the tick thread inside the gate. */
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, done,
|
||||
"RED: with offload OFF the inline connect blocks the delaymgr tick");
|
||||
}
|
||||
|
||||
/* Release the parked hook so the worker/tick thread can finish, then join. */
|
||||
rc_gate_open(&g_rc_gate);
|
||||
pthread_join(th, NULL);
|
||||
|
||||
ha_nop_req_hook = NULL;
|
||||
ha_reconnect_connect_hook = NULL;
|
||||
ha_set_reconnect_offload_for_test(1); /* restore production default */
|
||||
rc_gate_destroy(&g_rc_gate);
|
||||
rc_teardown_primary(&c, &standby_ep);
|
||||
printf("%s: PASS\n", label);
|
||||
}
|
||||
|
||||
static void test_reconnect_offload_tick_returns_promptly(void)
|
||||
{
|
||||
rc_run_offload_tick(1, "test_reconnect_offload_tick_returns_promptly");
|
||||
}
|
||||
|
||||
static void test_reconnect_inline_tick_blocks_RED(void)
|
||||
{
|
||||
rc_run_offload_tick(0, "test_reconnect_inline_tick_blocks_RED");
|
||||
}
|
||||
|
||||
/*
|
||||
* Test 17 (Req 2.5): per-peer dedup. While a reconnect to the unreachable
|
||||
* peer is in flight (parked in the hook), a SECOND heartbeat tick enqueues NO
|
||||
* duplicate reconnect for that peer — the connect hook is invoked exactly once
|
||||
* across both ticks.
|
||||
*/
|
||||
static void test_reconnect_offload_dedups_per_peer(void)
|
||||
{
|
||||
cluster_t c;
|
||||
myconf_t conf;
|
||||
endpoint_t standby_ep;
|
||||
rc_setup_primary(&c, &conf, &standby_ep);
|
||||
|
||||
rc_gate_init(&g_rc_gate);
|
||||
ha_reconnect_connect_hook = rc_gated_connect_hook;
|
||||
ha_set_reconnect_offload_for_test(1);
|
||||
|
||||
ha_nop_req_hook = stub_nop_req;
|
||||
|
||||
/* Tick 1: submits the reconnect; the lane worker enters the hook and parks. */
|
||||
primary_heartbeat_fn_for_test(&c);
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(0, rc_gate_wait_entered(&g_rc_gate, 1, 1000),
|
||||
"tick 1 must submit the reconnect (hook parks on the lane)");
|
||||
|
||||
/* Tick 2: peer still unreachable (in flight) — must NOT enqueue a duplicate. */
|
||||
primary_heartbeat_fn_for_test(&c);
|
||||
|
||||
/* Give any (erroneously) duplicated task time to reach the gate. */
|
||||
struct timespec ts = { .tv_sec = 0, .tv_nsec = 150 * 1000000L };
|
||||
nanosleep(&ts, NULL);
|
||||
|
||||
TEST_ASSERT_EQUAL_INT_MESSAGE(1, atomic_load(&g_rc_gate.calls),
|
||||
"Req 2.5: a second tick must not enqueue a duplicate reconnect for an "
|
||||
"already in-flight peer");
|
||||
|
||||
/* Release and let the single in-flight reconnect complete. */
|
||||
rc_gate_open(&g_rc_gate);
|
||||
|
||||
/* Drain: wait for the in-flight flag to clear (lane worker done). */
|
||||
for (int i = 0; i < 200; i++) {
|
||||
if (atomic_load_explicit(&c.ha.reconnect_inflight[2],
|
||||
memory_order_acquire) == 0) break;
|
||||
nanosleep(&ts, NULL);
|
||||
}
|
||||
|
||||
ha_nop_req_hook = NULL;
|
||||
ha_reconnect_connect_hook = NULL;
|
||||
rc_gate_destroy(&g_rc_gate);
|
||||
rc_teardown_primary(&c, &standby_ep);
|
||||
printf("test_reconnect_offload_dedups_per_peer: PASS\n");
|
||||
}
|
||||
|
||||
/* ── Suite entry point ──────────────────────────────────────────────────── */
|
||||
|
||||
void test_ha_liveness(void)
|
||||
|
|
@ -865,5 +1141,9 @@ void test_ha_liveness(void)
|
|||
RUN_TEST(test_cluster_dispatch_nop_skips_when_primary);
|
||||
RUN_TEST(test_cluster_dispatch_ctrlhello_resets_activity_when_standby);
|
||||
RUN_TEST(test_ha_arm_does_not_increase_thread_count);
|
||||
/* raft-drive-async Task 2.1 — reconnect offload (Req 2.1, 2.2, 2.5) */
|
||||
RUN_TEST(test_reconnect_offload_tick_returns_promptly);
|
||||
RUN_TEST(test_reconnect_inline_tick_blocks_RED);
|
||||
RUN_TEST(test_reconnect_offload_dedups_per_peer);
|
||||
UNITY_END();
|
||||
}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue