From b9ad01878ca7bf87a46b34e3ba0c2fa732af61ec Mon Sep 17 00:00:00 2001 From: Rongsong Shen Date: Thu, 2 Jul 2026 03:40:08 +0000 Subject: [PATCH] fix(lib): normalize delaymgr to one CLOCK_MONOTONIC ms timescale, fixing seconds-task starvation 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). --- src/include/delay.h | 12 +++- src/lib/delay.c | 140 +++++++++++++++----------------------------- 2 files changed, 55 insertions(+), 97 deletions(-) diff --git a/src/include/delay.h b/src/include/delay.h index ed02046..275da7b 100644 --- a/src/include/delay.h +++ b/src/include/delay.h @@ -12,10 +12,16 @@ typedef int (*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 { - time_t nextrun; - struct timespec nextrun_ts; /* non-zero tv_nsec flags an ms-task */ - int run_period; + uint64_t nextrun_ms; /* CLOCK_MONOTONIC deadline, in ms */ + int run_period_ms; int runcount; delay_task_entry_fn_t entry; void* args; diff --git a/src/lib/delay.c b/src/lib/delay.c index f9fa27f..260c7bb 100644 --- a/src/lib/delay.c +++ b/src/lib/delay.c @@ -1,4 +1,5 @@ #include +#include #include #include #include @@ -7,20 +8,22 @@ #include "delay.h" #include "utils.h" -/* Return 1 if a ms-task's deadline has passed, 0 otherwise. */ -static int ms_task_due(const delay_task_t* t) +/* Flat CLOCK_MONOTONIC "now", in milliseconds. The single time source every + * task deadline (delay_task_t.nextrun_ms) is compared against. */ +static uint64_t monotonic_ms(void) { struct timespec now; clock_gettime(CLOCK_MONOTONIC, &now); - if (now.tv_sec > t->nextrun_ts.tv_sec) { - return 1; - } - if (now.tv_sec == t->nextrun_ts.tv_sec - && now.tv_nsec >= t->nextrun_ts.tv_nsec) { - return 1; - } - return 0; + return (uint64_t)now.tv_sec * 1000ULL + (uint64_t)now.tv_nsec / 1000000ULL; +} + +/* Return 1 if a task's deadline has passed, 0 otherwise. Every task in the + * queue uses this single millisecond time domain (see delay.h) -- there is + * no separate "seconds task" due-check anymore. */ +static int task_due(const delay_task_t* t) +{ + return monotonic_ms() >= t->nextrun_ms; } static void* delay_worker(void* args) @@ -30,8 +33,6 @@ static void* delay_worker(void* args) while (1) { ordq_item_t* it; delay_task_t* t; - time_t now; - int is_ms_task; pthread_testcancel(); { @@ -48,18 +49,9 @@ static void* delay_worker(void* args) continue; } - t = (delay_task_t*)it->data; - is_ms_task = (t->nextrun_ts.tv_nsec != 0 || t->nextrun_ts.tv_sec != 0); - - if (is_ms_task) { - if (!ms_task_due(t)) { - continue; - } - } else { - now = time(NULL); - if (t->nextrun > now) { - continue; - } + t = (delay_task_t*)it->data; + if (!task_due(t)) { + continue; } it = ordq_get(&(mgr->tasks), NULL); @@ -73,17 +65,9 @@ static void* delay_worker(void* args) if (t->runcount == 0) { FREEUP(t); } else { - if (is_ms_task) { - /* advance deadline by run_period ms */ - long add_ns = (long)t->run_period * 1000000L; - - 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; - } + /* advance deadline by run_period_ms -- flat integer add, no + * sec/nsec carry arithmetic needed. */ + t->nextrun_ms += (uint64_t)t->run_period_ms; ordq_put(&(mgr->tasks), t); } @@ -93,31 +77,18 @@ static void* delay_worker(void* args) 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) { (void)ctx; delay_task_t* t1 = (delay_task_t*)arg1; delay_task_t* t2 = (delay_task_t*)arg2; - /* ms-tasks: compare by nextrun_ts (nanosecond-precision deadline). - * Both delay_task_ms and ordq_put use nextrun==0 as the ms-task marker. */ - int t1_is_ms = (t1->nextrun_ts.tv_nsec != 0 || t1->nextrun_ts.tv_sec != 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; + if (t1->nextrun_ms > t2->nextrun_ms) return 1; + if (t1->nextrun_ms < t2->nextrun_ms) return -1; + return 0; } static void delay_task_killer(delay_task_t* t) @@ -220,39 +191,33 @@ int delaymgr_start(delaymgr_t* mgr) 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, delay_task_entry_fn_t entry, void* args) { - delay_task_t* t; - time_t now = time(NULL); - time_t ts_nextrun; - - if ((mgr == NULL) || (delay <= 0) || (entry == NULL)) { + if (delay <= 0) { return -EINVAL; } - - t = (delay_task_t*)calloc(1, sizeof(*t)); - if (t == NULL) { - return -ENOMEM; + long delay_ms = (long)delay * 1000L; + if (delay_ms > INT_MAX) { + delay_ms = INT_MAX; } - - 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); + return delay_task_ms(mgr, (int)delay_ms, runcount, entry, args); } int delay_task_ms(delaymgr_t* mgr, int delay_ms, int runcount, delay_task_entry_fn_t entry, void* args) { - delay_task_t* t; - struct timespec now; - long add_ns; + delay_task_t* t; if ((mgr == NULL) || (delay_ms <= 0) || (entry == NULL)) { return -EINVAL; @@ -263,24 +228,11 @@ int delay_task_ms(delaymgr_t* mgr, int delay_ms, int runcount, return -ENOMEM; } - clock_gettime(CLOCK_MONOTONIC, &now); - - add_ns = (long)delay_ms * 1000000L; - t->nextrun_ts.tv_sec = now.tv_sec + (now.tv_nsec + add_ns) / 1000000000L; - t->nextrun_ts.tv_nsec = (now.tv_nsec + add_ns) % 1000000000L; - - /* 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; + t->nextrun_ms = monotonic_ms() + (uint64_t)delay_ms; + t->runcount = runcount; + t->run_period_ms = delay_ms; + t->entry = entry; + t->args = args; return ordq_put(&(mgr->tasks), t); }