feat(ha-heartbeat-liveness): replace hb_timeout_thread with liveness_poll_fn (task 2.1)
Remove hb_vtask_t, hb_timeout_thread, s_hb_vtask from ha.c. Replace with liveness_poll_ctx_t + delaymgr liveness_poll_fn: checks cancelled flag and role==CTRL_STANDBY before acting; re-arms at heartbeat_interval_ms/2. ha_on_primary_ctrl_fd_rdhup calls ha_disarm_heartbeat_timeout_vtask first to prevent duplicate failover (Req 5.4). ha_on_heartbeat_received writes last_activity via clock_gettime. Add test_ha_liveness.c with 7 unit tests.
This commit is contained in:
parent
9f9a6b3c0f
commit
738859fb4e
6 changed files with 465 additions and 90 deletions
|
|
@ -2674,6 +2674,10 @@ void ha_on_primary_ctrl_fd_rdhup(ha_state_t *ha, cluster_t *c)
|
|||
{
|
||||
if (!ha) return;
|
||||
|
||||
/* Cancel the liveness poll task first (Req 5.4): prevents a concurrent
|
||||
* poll fn from triggering a second failover call after EPOLLRDHUP fires. */
|
||||
ha_disarm_heartbeat_timeout_vtask(ha);
|
||||
|
||||
/* Close and null the primary control fd */
|
||||
if (ha->primary_ctrl_fd >= 0) {
|
||||
close(ha->primary_ctrl_fd);
|
||||
|
|
@ -2734,117 +2738,160 @@ void ha_on_primary_ctrl_fd_rdhup(ha_state_t *ha, cluster_t *c)
|
|||
raft_on_leader_lost(&ha->raft);
|
||||
}
|
||||
|
||||
/* ── Heartbeat timeout watchdog ─────────────────────────────────────────── */
|
||||
/* ── Standby liveness poll (replaces hb_timeout_thread) ─────────────────── */
|
||||
|
||||
/*
|
||||
* liveness_poll_ctx_t: context for the delaymgr-based standby liveness task.
|
||||
*
|
||||
* Requirements: 1.2, 1.4, 1.5, 1.6, 5.4
|
||||
*
|
||||
* Replaces hb_vtask_t + hb_timeout_thread. No pthread, no semaphore.
|
||||
* The poll fn reads ha->last_activity (written by ha_on_heartbeat_received)
|
||||
* and fires ha_on_primary_ctrl_fd_rdhup if elapsed >= heartbeat_timeout_ms.
|
||||
*
|
||||
* cancelled is _Atomic so that writes from ha_disarm_heartbeat_timeout_vtask
|
||||
* (main/vtask context) are visible to liveness_poll_fn (delaymgr worker thread)
|
||||
* without a data race.
|
||||
*/
|
||||
typedef struct {
|
||||
ha_state_t *ha;
|
||||
cluster_t *c;
|
||||
sem_t reset_sem; /* ha_on_heartbeat_received posts here */
|
||||
atomic_int disarmed; /* set to 1 by ha_disarm_heartbeat_timeout_vtask */
|
||||
pthread_t thread;
|
||||
int started;
|
||||
} hb_vtask_t;
|
||||
ha_state_t *ha;
|
||||
cluster_t *c;
|
||||
_Atomic int cancelled;
|
||||
} liveness_poll_ctx_t;
|
||||
|
||||
/* File-scope pointer to the active heartbeat watchdog.
|
||||
* Safety: arm/disarm/heartbeat are called only from the single-threaded
|
||||
* vtaskmgr dispatch context; no locking is needed in the current architecture.
|
||||
* If concurrent access is added later, protect with ha->lock. */
|
||||
static hb_vtask_t *s_hb_vtask = NULL;
|
||||
/* Static pointer to the active context. _Atomic so concurrent arm/disarm
|
||||
* vs poll-fn execution is well-defined. Access pattern:
|
||||
* arm — allocates and stores (single writer, vtask context)
|
||||
* disarm — sets cancelled, frees (single writer, vtask context)
|
||||
* poll_fn — reads via passed arg only; never dereferences s_liveness_ctx
|
||||
*/
|
||||
static liveness_poll_ctx_t * _Atomic s_liveness_ctx = NULL;
|
||||
|
||||
static void *hb_timeout_thread(void *arg)
|
||||
/*
|
||||
* liveness_poll_fn: delaymgr task replacing hb_timeout_thread.
|
||||
*
|
||||
* Checks whether the primary has been silent for >= heartbeat_timeout_ms.
|
||||
* If so, calls ha_on_primary_ctrl_fd_rdhup and does not re-arm.
|
||||
* If not, re-arms itself for the next poll interval.
|
||||
* If cancelled, returns immediately without any action.
|
||||
*
|
||||
* Also skips if role != CTRL_STANDBY (Req 5.4): prevents duplicate failover
|
||||
* if EPOLLRDHUP already transitioned the role to CTRL_ELECTING.
|
||||
*/
|
||||
static int liveness_poll_fn(void *arg)
|
||||
{
|
||||
hb_vtask_t *t = (hb_vtask_t *)arg;
|
||||
ha_state_t *ha = t->ha;
|
||||
liveness_poll_ctx_t *ctx = (liveness_poll_ctx_t *)arg;
|
||||
|
||||
while (!atomic_load(&t->disarmed)) {
|
||||
/* Compute deadline: heartbeat_timeout_ms + rand_jitter.
|
||||
* Must use heartbeat_timeout_ms (not election_timeout_ms) so standbys
|
||||
* don't falsely trigger failover while the primary is in COMMITTING. */
|
||||
uint32_t jitter_ms = 0;
|
||||
if (ha->config.election_jitter_max_ms > 0) {
|
||||
struct timespec now;
|
||||
clock_gettime(CLOCK_REALTIME, &now);
|
||||
unsigned int seed = (unsigned int)now.tv_nsec;
|
||||
jitter_ms = (uint32_t)(rand_r(&seed) % ha->config.election_jitter_max_ms);
|
||||
}
|
||||
uint32_t total_ms = ha->config.heartbeat_timeout_ms + jitter_ms;
|
||||
/* Check cancelled flag first — set by ha_disarm_heartbeat_timeout_vtask */
|
||||
if (atomic_load_explicit(&ctx->cancelled, memory_order_acquire))
|
||||
return 0;
|
||||
|
||||
struct timespec deadline;
|
||||
clock_gettime(CLOCK_REALTIME, &deadline);
|
||||
deadline.tv_sec += (long)(total_ms / 1000);
|
||||
deadline.tv_nsec += (long)(total_ms % 1000) * 1000000L;
|
||||
if (deadline.tv_nsec >= 1000000000L) {
|
||||
deadline.tv_sec++;
|
||||
deadline.tv_nsec -= 1000000000L;
|
||||
}
|
||||
ha_state_t *ha = ctx->ha;
|
||||
cluster_t *c = ctx->c;
|
||||
|
||||
/* Wait for heartbeat signal or deadline */
|
||||
int rc = sem_timedwait(&t->reset_sem, &deadline);
|
||||
if (rc == -1 && errno == ETIMEDOUT) {
|
||||
/* Timeout with no heartbeat → failover (Req 6.3, 6.4) */
|
||||
if (!atomic_load(&t->disarmed)) {
|
||||
WARN("HA: heartbeat timeout (%ums) → EPOLLRDHUP failover path",
|
||||
total_ms);
|
||||
ha_on_primary_ctrl_fd_rdhup(ha, t->c);
|
||||
break; /* fired; thread exits */
|
||||
}
|
||||
}
|
||||
/* heartbeat received — loop and re-arm deadline */
|
||||
/* Skip if not standby — EPOLLRDHUP may have already changed the role */
|
||||
if (atomic_load_explicit(&ha->role, memory_order_acquire) != CTRL_STANDBY)
|
||||
return 0;
|
||||
|
||||
/* Compute elapsed time since last primary message */
|
||||
struct timespec now_ts;
|
||||
clock_gettime(CLOCK_MONOTONIC, &now_ts);
|
||||
time_t now = now_ts.tv_sec;
|
||||
time_t last = ha->last_activity; /* time_t write is atomic on LP64 */
|
||||
time_t elapsed = (last > 0) ? (now - last) : 0;
|
||||
uint32_t timeout_s = ha->config.heartbeat_timeout_ms / 1000;
|
||||
|
||||
if (last > 0 && (uint32_t)elapsed >= timeout_s) {
|
||||
/* Silence exceeded timeout — trigger failover */
|
||||
WARN("HA: liveness poll: no primary activity for %lus (timeout=%us) → failover",
|
||||
(unsigned long)elapsed, timeout_s);
|
||||
ha_on_primary_ctrl_fd_rdhup(ha, c);
|
||||
/* Do NOT re-arm — failover handles re-election */
|
||||
return 0;
|
||||
}
|
||||
return NULL;
|
||||
|
||||
/* Still fresh — re-arm for next poll interval */
|
||||
if (c && c->delaymgr.running) {
|
||||
uint32_t poll_ms = ha->config.heartbeat_interval_ms / 2;
|
||||
if (poll_ms < 100) poll_ms = 100;
|
||||
delay_task_ms(&c->delaymgr, (int)poll_ms, 1,
|
||||
(delay_task_entry_fn_t)liveness_poll_fn, ctx);
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
#ifdef XCE_TESTING
|
||||
/*
|
||||
* liveness_poll_fn_test_entry: test-only entry point that calls liveness_poll_fn
|
||||
* directly with a pre-constructed ctx, bypassing delaymgr.
|
||||
* Only compiled when XCE_TESTING is defined.
|
||||
*/
|
||||
void liveness_poll_fn_test_entry(liveness_poll_ctx_t *ctx)
|
||||
{
|
||||
liveness_poll_fn(ctx);
|
||||
}
|
||||
#endif
|
||||
|
||||
void ha_arm_heartbeat_timeout_vtask(ha_state_t *ha, cluster_t *c)
|
||||
{
|
||||
if (!ha) return;
|
||||
|
||||
/* Disarm any existing watchdog */
|
||||
if (s_hb_vtask && s_hb_vtask->started) {
|
||||
atomic_store(&s_hb_vtask->disarmed, 1);
|
||||
sem_post(&s_hb_vtask->reset_sem); /* wake sleeping thread */
|
||||
pthread_join(s_hb_vtask->thread, NULL);
|
||||
sem_destroy(&s_hb_vtask->reset_sem);
|
||||
free(s_hb_vtask);
|
||||
s_hb_vtask = NULL;
|
||||
/* Disarm any existing poll ctx */
|
||||
liveness_poll_ctx_t *old = atomic_load_explicit(&s_liveness_ctx,
|
||||
memory_order_acquire);
|
||||
if (old) {
|
||||
atomic_store_explicit(&old->cancelled, 1, memory_order_release);
|
||||
free(old);
|
||||
atomic_store_explicit(&s_liveness_ctx, NULL, memory_order_release);
|
||||
}
|
||||
|
||||
hb_vtask_t *t = calloc(1, sizeof(*t));
|
||||
if (!t) return;
|
||||
liveness_poll_ctx_t *ctx = calloc(1, sizeof(*ctx));
|
||||
if (!ctx) return;
|
||||
|
||||
t->ha = ha;
|
||||
t->c = c;
|
||||
t->started = 0;
|
||||
atomic_init(&t->disarmed, 0);
|
||||
sem_init(&t->reset_sem, 0, 0);
|
||||
ctx->ha = ha;
|
||||
ctx->c = c;
|
||||
atomic_init(&ctx->cancelled, 0);
|
||||
|
||||
if (pthread_create(&t->thread, NULL, hb_timeout_thread, t) != 0) {
|
||||
sem_destroy(&t->reset_sem);
|
||||
free(t);
|
||||
return;
|
||||
/* Initialise last_activity to now so the first poll computes elapsed from
|
||||
* the arm time rather than from zero. */
|
||||
struct timespec ts;
|
||||
clock_gettime(CLOCK_MONOTONIC, &ts);
|
||||
ha->last_activity = ts.tv_sec;
|
||||
|
||||
atomic_store_explicit(&s_liveness_ctx, ctx, memory_order_release);
|
||||
|
||||
uint32_t poll_ms = ha->config.heartbeat_interval_ms / 2;
|
||||
if (poll_ms < 100) poll_ms = 100;
|
||||
|
||||
INFO("HA: standby liveness poll armed (timeout=%ums, poll=%ums)",
|
||||
ha->config.heartbeat_timeout_ms, poll_ms);
|
||||
|
||||
if (c && c->delaymgr.running) {
|
||||
int rc = delay_task_ms(&c->delaymgr, (int)poll_ms, 1,
|
||||
(delay_task_entry_fn_t)liveness_poll_fn, ctx);
|
||||
if (rc < 0)
|
||||
WARN("HA: ha_arm_heartbeat_timeout_vtask: delay_task_ms failed: %d", rc);
|
||||
}
|
||||
t->started = 1;
|
||||
s_hb_vtask = t;
|
||||
}
|
||||
|
||||
void ha_on_heartbeat_received(ha_state_t *ha)
|
||||
{
|
||||
(void)ha;
|
||||
if (s_hb_vtask && s_hb_vtask->started)
|
||||
sem_post(&s_hb_vtask->reset_sem);
|
||||
if (!ha) return;
|
||||
struct timespec ts;
|
||||
clock_gettime(CLOCK_MONOTONIC, &ts);
|
||||
ha->last_activity = ts.tv_sec;
|
||||
tlog(TLOG_DEBUG3, "HA: heartbeat received — last_activity updated to %ld", (long)ts.tv_sec);
|
||||
}
|
||||
|
||||
void ha_disarm_heartbeat_timeout_vtask(ha_state_t *ha)
|
||||
{
|
||||
(void)ha;
|
||||
if (!s_hb_vtask) return;
|
||||
if (s_hb_vtask->started) {
|
||||
atomic_store(&s_hb_vtask->disarmed, 1);
|
||||
sem_post(&s_hb_vtask->reset_sem);
|
||||
pthread_join(s_hb_vtask->thread, NULL);
|
||||
sem_destroy(&s_hb_vtask->reset_sem);
|
||||
}
|
||||
free(s_hb_vtask);
|
||||
s_hb_vtask = NULL;
|
||||
liveness_poll_ctx_t *ctx = atomic_load_explicit(&s_liveness_ctx,
|
||||
memory_order_acquire);
|
||||
if (!ctx) return;
|
||||
atomic_store_explicit(&ctx->cancelled, 1, memory_order_release);
|
||||
free(ctx);
|
||||
atomic_store_explicit(&s_liveness_ctx, NULL, memory_order_release);
|
||||
}
|
||||
|
||||
/* ── Primary heartbeat sender + late-joiner outbound connect ─────────────── */
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue