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

fix(lib): normalize delaymgr to one CLOCK_MONOTONIC ms timescale, fixing seconds-task starvation

Browse source 
Root-caused via raft-ha-core-safety 5.10a investigation: schedule_checkpoint()
(persist.c) fires correctly once wal_record_count crosses its threshold, but
its delay_task()-scheduled checkpoint_task_fn never executed in a live
multi-node deployment -- confirmed live via temporary diagnostic logging
(wal_record_count=100 max=100 logged, but the task itself never ran).

Root cause: delay_task_t carried TWO incomparable time representations --
`nextrun` (wall-clock time_t seconds, used by delay_task()) and `nextrun_ts`
(CLOCK_MONOTONIC timespec, used by delay_task_ms()). task_cmp()'s mixed-type
fallback compared them directly: an ms-task's nextrun is always 0, a
seconds-task's nextrun is a real epoch value (e.g. 1785600123) -- so 0
always sorted "less than" any seconds-task, regardless of actual due time.
ordq (src/lib/ordq.c) keeps its queue as an ascending sorted linked list by
this comparator, and delay_worker() only ever inspects the list HEAD
(ordq_peek with no matchfn returns q->head immediately, no scanning). Since
cluster_t.delaymgr is shared with ha.c's ~10 recurring delay_task_ms() call
sites (heartbeats, election timers, RDHUP polling), the head was
structurally almost-always occupied by an ms-task -- any delay_task()
(seconds) caller sharing that delaymgr, including schedule_checkpoint's
checkpoint_task_fn, could be starved indefinitely.

Fix: eliminate the dual representation entirely. Every task -- whether
scheduled via delay_task() (seconds) or delay_task_ms() (milliseconds) --
now stores one flat uint64_t CLOCK_MONOTONIC millisecond deadline
(nextrun_ms). delay_task() becomes a thin seconds->ms wrapper around
delay_task_ms(). task_cmp() and task_due() collapse to a single integer
comparison; delay_worker()'s reschedule is a flat integer add (no more
tv_sec/tv_nsec carry arithmetic, no more is_ms_task branching). Net -42
lines. Live-verified: after this fix, a forced checkpoint round produces a
real installed snapshot file (previously: schedule_checkpoint fired but the
task silently never ran).

controller_unit_test: full 58-group sweep, 0 Failures, 0 ASAN reports
(before and after this change) -- no regression, including every
delaymgr-touching group (test_demotion_convergence, test_ha_start_election_async,
test_ha_core_safety, test_ha_startup, etc).
This commit is contained in:
Rongsong Shen 2026-07-02 03:40:08 +00:00
parent 4696f43a74
commit b9ad01878c
2 changed files with 55 additions and 97 deletions
Download patch file Download diff file Expand all files Collapse all files

12
src/include/delay.h
View file

@ -12,10 +12,16 @@
typedef int (*delay_task_entry_fn_t)(void*); typedef int (*delay_task_entry_fn_t)(void*);
typedef int (*task_run_fn_t)(delay_task_entry_fn_t, void*); typedef int (*task_run_fn_t)(delay_task_entry_fn_t, void*);
/* Single normalized time domain: every task (whether scheduled via
* delay_task() seconds or delay_task_ms() milliseconds) is stored as one
* flat CLOCK_MONOTONIC millisecond deadline. There is no separate
* wall-clock/ms task "flavor" to distinguish -- delay_task() is a thin
* seconds->ms wrapper around delay_task_ms() (see delay.c), so every entry
* in the queue is always comparable on the same scale with a single
* integer comparison (no tv_sec/tv_nsec pair, no carry arithmetic). */
typedef struct { typedef struct {
time_t nextrun; uint64_t nextrun_ms; /* CLOCK_MONOTONIC deadline, in ms */
struct timespec nextrun_ts; /* non-zero tv_nsec flags an ms-task */ int run_period_ms;
int run_period;
int runcount; int runcount;
delay_task_entry_fn_t entry; delay_task_entry_fn_t entry;
void* args; void* args;

140
src/lib/delay.c
View file

@ -1,4 +1,5 @@
#include <errno.h> #include <errno.h>
#include <limits.h>
#include <stdio.h> #include <stdio.h>
#include <stdlib.h> #include <stdlib.h>
#include <string.h> #include <string.h>
@ -7,20 +8,22 @@
#include "delay.h" #include "delay.h"
#include "utils.h" #include "utils.h"
/* Return 1 if a ms-task's deadline has passed, 0 otherwise. */ /* Flat CLOCK_MONOTONIC "now", in milliseconds. The single time source every
static int ms_task_due(const delay_task_t* t) * task deadline (delay_task_t.nextrun_ms) is compared against. */
static uint64_t monotonic_ms(void)
{ {
struct timespec now; struct timespec now;
clock_gettime(CLOCK_MONOTONIC, &now); clock_gettime(CLOCK_MONOTONIC, &now);
if (now.tv_sec > t->nextrun_ts.tv_sec) { return (uint64_t)now.tv_sec * 1000ULL + (uint64_t)now.tv_nsec / 1000000ULL;
return 1; }
}
if (now.tv_sec == t->nextrun_ts.tv_sec /* Return 1 if a task's deadline has passed, 0 otherwise. Every task in the
&& now.tv_nsec >= t->nextrun_ts.tv_nsec) { * queue uses this single millisecond time domain (see delay.h) -- there is
return 1; * no separate "seconds task" due-check anymore. */
} static int task_due(const delay_task_t* t)
return 0; {
return monotonic_ms() >= t->nextrun_ms;
} }
static void* delay_worker(void* args) static void* delay_worker(void* args)
@ -30,8 +33,6 @@ static void* delay_worker(void* args)
while (1) { while (1) {
ordq_item_t* it; ordq_item_t* it;
delay_task_t* t; delay_task_t* t;
time_t now;
int is_ms_task;
pthread_testcancel(); pthread_testcancel();
{ {
@ -48,18 +49,9 @@ static void* delay_worker(void* args)
continue; continue;
} }
t = (delay_task_t*)it->data; t = (delay_task_t*)it->data;
is_ms_task = (t->nextrun_ts.tv_nsec != 0 || t->nextrun_ts.tv_sec != 0); if (!task_due(t)) {
continue;
if (is_ms_task) {
if (!ms_task_due(t)) {
continue;
}
} else {
now = time(NULL);
if (t->nextrun > now) {
continue;
}
} }
it = ordq_get(&(mgr->tasks), NULL); it = ordq_get(&(mgr->tasks), NULL);
@ -73,17 +65,9 @@ static void* delay_worker(void* args)
if (t->runcount == 0) { if (t->runcount == 0) {
FREEUP(t); FREEUP(t);
} else { } else {
if (is_ms_task) { /* advance deadline by run_period_ms -- flat integer add, no
/* advance deadline by run_period ms */ * sec/nsec carry arithmetic needed. */
long add_ns = (long)t->run_period * 1000000L; t->nextrun_ms += (uint64_t)t->run_period_ms;
t->nextrun_ts.tv_nsec += add_ns;
t->nextrun_ts.tv_sec += t->nextrun_ts.tv_nsec / 1000000000L;
t->nextrun_ts.tv_nsec %= 1000000000L;
} else {
/* put back to taskq as it need to be run again */
t->nextrun = now + t->run_period;
}
ordq_put(&(mgr->tasks), t); ordq_put(&(mgr->tasks), t);
} }
@ -93,31 +77,18 @@ static void* delay_worker(void* args)
return NULL; return NULL;
} }
/* Single normalized comparison: every task carries one flat millisecond
* deadline (nextrun_ms), so ordering is always apples-to-apples regardless
* of whether the task was scheduled via delay_task() or delay_task_ms(). */
static int task_cmp(void* ctx, void* arg1, void* arg2) static int task_cmp(void* ctx, void* arg1, void* arg2)
{ {
(void)ctx; (void)ctx;
delay_task_t* t1 = (delay_task_t*)arg1; delay_task_t* t1 = (delay_task_t*)arg1;
delay_task_t* t2 = (delay_task_t*)arg2; delay_task_t* t2 = (delay_task_t*)arg2;
/* ms-tasks: compare by nextrun_ts (nanosecond-precision deadline). if (t1->nextrun_ms > t2->nextrun_ms) return 1;
* Both delay_task_ms and ordq_put use nextrun==0 as the ms-task marker. */ if (t1->nextrun_ms < t2->nextrun_ms) return -1;
int t1_is_ms = (t1->nextrun_ts.tv_nsec != 0 || t1->nextrun_ts.tv_sec != 0); return 0;
int t2_is_ms = (t2->nextrun_ts.tv_nsec != 0 || t2->nextrun_ts.tv_sec != 0);
if (t1_is_ms && t2_is_ms) {
if (t1->nextrun_ts.tv_sec > t2->nextrun_ts.tv_sec) return 1;
if (t1->nextrun_ts.tv_sec < t2->nextrun_ts.tv_sec) return -1;
if (t1->nextrun_ts.tv_nsec > t2->nextrun_ts.tv_nsec) return 1;
if (t1->nextrun_ts.tv_nsec < t2->nextrun_ts.tv_nsec) return -1;
return 0;
}
/* Mixed or seconds-based: fall back to nextrun comparison. */
time_t k1 = t1->nextrun;
time_t k2 = t2->nextrun;
if (k1 > k2) return 1;
if (k1 == k2) return 0;
return -1;
} }
static void delay_task_killer(delay_task_t* t) static void delay_task_killer(delay_task_t* t)
@ -220,39 +191,33 @@ int delaymgr_start(delaymgr_t* mgr)
return 0; return 0;
} }
/* delay_task: seconds-granularity scheduling API. A thin wrapper over
* delay_task_ms() -- every task, regardless of which API scheduled it, is
* stored and compared on the single CLOCK_MONOTONIC timescale (delay.h), so
* a seconds-task and an ms-task in the same delaymgr are always correctly
* ordered against each other (previously a mixed wall-clock/monotonic
* comparison meant a seconds-task could be starved indefinitely behind a
* steady stream of ms-tasks -- see the bug this simplification fixes).
* INT_MAX/1000 seconds (~24 days) is far beyond any real caller's delay; a
* value that would overflow the ms conversion is clamped rather than
* wrapped. */
int delay_task(delaymgr_t* mgr, int delay, int runcount, int delay_task(delaymgr_t* mgr, int delay, int runcount,
delay_task_entry_fn_t entry, void* args) delay_task_entry_fn_t entry, void* args)
{ {
delay_task_t* t; if (delay <= 0) {
time_t now = time(NULL);
time_t ts_nextrun;
if ((mgr == NULL) || (delay <= 0) || (entry == NULL)) {
return -EINVAL; return -EINVAL;
} }
long delay_ms = (long)delay * 1000L;
t = (delay_task_t*)calloc(1, sizeof(*t)); if (delay_ms > INT_MAX) {
if (t == NULL) { delay_ms = INT_MAX;
return -ENOMEM;
} }
return delay_task_ms(mgr, (int)delay_ms, runcount, entry, args);
ts_nextrun = now + delay;
t->nextrun = ts_nextrun;
t->runcount = runcount;
t->run_period = delay;
t->entry = entry;
t->args = args;
return ordq_put(&(mgr->tasks), t);
} }
int delay_task_ms(delaymgr_t* mgr, int delay_ms, int runcount, int delay_task_ms(delaymgr_t* mgr, int delay_ms, int runcount,
delay_task_entry_fn_t entry, void* args) delay_task_entry_fn_t entry, void* args)
{ {
delay_task_t* t; delay_task_t* t;
struct timespec now;
long add_ns;
if ((mgr == NULL) || (delay_ms <= 0) || (entry == NULL)) { if ((mgr == NULL) || (delay_ms <= 0) || (entry == NULL)) {
return -EINVAL; return -EINVAL;
@ -263,24 +228,11 @@ int delay_task_ms(delaymgr_t* mgr, int delay_ms, int runcount,
return -ENOMEM; return -ENOMEM;
} }
clock_gettime(CLOCK_MONOTONIC, &now); t->nextrun_ms = monotonic_ms() + (uint64_t)delay_ms;
t->runcount = runcount;
add_ns = (long)delay_ms * 1000000L; t->run_period_ms = delay_ms;
t->nextrun_ts.tv_sec = now.tv_sec + (now.tv_nsec + add_ns) / 1000000000L; t->entry = entry;
t->nextrun_ts.tv_nsec = (now.tv_nsec + add_ns) % 1000000000L; t->args = args;
/* Ensure nextrun_ts is never all-zero so the ms-task flag is unambiguous.
* tv_nsec == 0 can only arise when add_ns is an exact multiple of 1s and
* now.tv_nsec == 0; in that edge case bump tv_nsec by 1 ns. */
if (t->nextrun_ts.tv_nsec == 0) {
t->nextrun_ts.tv_nsec = 1;
}
t->nextrun = 0; /* unused for ms-tasks */
t->runcount = runcount;
t->run_period = delay_ms;
t->entry = entry;
t->args = args;
return ordq_put(&(mgr->tasks), t); return ordq_put(&(mgr->tasks), t);
} }