From 2763e7d78463ea149fe3650e3c6c720cf9bf7882 Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Wed, 10 Jun 2026 18:41:07 +0300 Subject: [PATCH 1/6] MOD-15307: drain LibMR threads to a safe point before fork() Add MR_DrainForFork()/MR_ResumeAfterFork(). On the main thread (from the module's FORK_CHILD_PRE handler) park the event-loop thread at a between-tasks safe point via a posted task and bounded-wait the worker pool to idle, so no LibMR thread holds a libc lock at fork() (ghost-lock). Bounded + fail-open; cooperative (not the SIGUSR2 mr_thpool_pause, which can freeze a worker mid-malloc). Resume releases the parked event-loop thread. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/mr.c | 107 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/mr.h | 8 +++++ 2 files changed, 115 insertions(+) diff --git a/src/mr.c b/src/mr.c index 3f9d5a82..3aabe8fe 100644 --- a/src/mr.c +++ b/src/mr.c @@ -22,6 +22,8 @@ #include "utils/buffer.h" #include +#include +#include #include #ifndef EXECUTION_DEFAULT_MAX_IDLE_MS #define EXECUTION_DEFAULT_MAX_IDLE_MS 5000 @@ -1408,6 +1410,111 @@ void MR_ExecutionSetMaxIdle(Execution* e, size_t maxIdle) { e->timeoutMS = maxIdle; } +/* ---- MOD-15307: drain LibMR background threads to a safe point before fork() ---- + * + * A LibMR thread (a worker in the execution pool, or the event-loop thread) that holds + * a libc lock (e.g. the malloc arena lock) at the instant redis calls fork() leaves the + * child holding a locked mutex with no owner -> the child ghost-locks the first time it + * mallocs (RDB save, slot-migration snapshot, etc.). That is the root trigger behind the + * ASM-migration nightly hangs (MOD-15307) and, downstream, the multi-shard query max-idle + * timeout (MOD-14615). + * + * MR_DrainForFork() (called on the main thread from the FORK_CHILD_PRE module event) brings + * both kinds of LibMR thread to a safe, lock-free point: + * - the event-loop thread is parked *between tasks* via a posted task (so it is not + * mid-message-deserialize / mid-malloc, and is not holding the module GIL); + * - with the event-loop thread parked no new executions are dispatched, so the worker + * pool drains to idle. + * Both waits are bounded; on timeout we fork anyway (fail-open -> never worse than today). + * Note a worker still blocked acquiring the module GIL (held by us) is itself malloc-safe, + * so a timeout there is benign. MR_ResumeAfterFork() releases the parked event-loop thread + * and must be called after fork() (FORK_CHILD_BORN) or if the fork was cancelled. */ +#define MR_FORK_DRAIN_TIMEOUT_MS 2000 + +static pthread_mutex_t mr_forkDrainLock = PTHREAD_MUTEX_INITIALIZER; +static pthread_cond_t mr_forkDrainCond = PTHREAD_COND_INITIALIZER; +static int mr_forkElParked = 0; /* set by the park task once the el-thread is quiesced */ +static int mr_forkElRelease = 0; /* set by MR_ResumeAfterFork to release the el-thread */ + +/* Runs on the event-loop thread, between tasks (a safe point). Parks the thread until + * MR_ResumeAfterFork() is called. */ +static void MR_ForkParkElThread(void* ctx) { + REDISMODULE_NOT_USED(ctx); + pthread_mutex_lock(&mr_forkDrainLock); + mr_forkElParked = 1; + pthread_cond_broadcast(&mr_forkDrainCond); + while (!mr_forkElRelease) { + pthread_cond_wait(&mr_forkDrainCond, &mr_forkDrainLock); + } + mr_forkElParked = 0; + pthread_mutex_unlock(&mr_forkDrainLock); +} + +static int mr_forkDeadlinePassed(const struct timespec* deadline) { + struct timespec now; + clock_gettime(CLOCK_REALTIME, &now); + return (now.tv_sec > deadline->tv_sec) || + (now.tv_sec == deadline->tv_sec && now.tv_nsec >= deadline->tv_nsec); +} + +void MR_DrainForFork(void) { + /* Nothing to drain if the pool / event loop were never started. */ + if (!mrCtx.executionsThreadPool) return; + + RedisModule_Log(mr_staticCtx, "notice", "MOD-15307: draining LibMR threads before fork()"); + + struct timespec deadline; + clock_gettime(CLOCK_REALTIME, &deadline); + deadline.tv_sec += MR_FORK_DRAIN_TIMEOUT_MS / 1000; + deadline.tv_nsec += (MR_FORK_DRAIN_TIMEOUT_MS % 1000) * 1000000L; + if (deadline.tv_nsec >= 1000000000L) { deadline.tv_sec++; deadline.tv_nsec -= 1000000000L; } + + /* 1) Park the event-loop thread at a between-tasks safe point. */ + pthread_mutex_lock(&mr_forkDrainLock); + mr_forkElParked = 0; + mr_forkElRelease = 0; + pthread_mutex_unlock(&mr_forkDrainLock); + + MR_EventLoopAddTask(MR_ForkParkElThread, NULL); + + pthread_mutex_lock(&mr_forkDrainLock); + while (!mr_forkElParked) { + if (pthread_cond_timedwait(&mr_forkDrainCond, &mr_forkDrainLock, &deadline) == ETIMEDOUT) + break; + } + int parked = mr_forkElParked; + pthread_mutex_unlock(&mr_forkDrainLock); + + if (!parked) { + RedisModule_Log(mr_staticCtx, "warning", + "MR_DrainForFork: event-loop thread did not park within %dms; forking anyway", + MR_FORK_DRAIN_TIMEOUT_MS); + } + + /* 2) Wait (bounded) for in-flight worker jobs to finish. With the event loop parked no + * new work is dispatched, so the working count drains to 0 unless a worker is blocked + * acquiring the GIL we hold -- a malloc-safe state, so a timeout there is benign. */ + while (mr_thpool_num_threads_working(mrCtx.executionsThreadPool) > 0) { + if (mr_forkDeadlinePassed(&deadline)) { + RedisModule_Log(mr_staticCtx, "warning", + "MR_DrainForFork: %d worker(s) still busy at timeout; forking anyway", + mr_thpool_num_threads_working(mrCtx.executionsThreadPool)); + break; + } + struct timespec nap = { .tv_sec = 0, .tv_nsec = 1000000L }; /* 1ms */ + nanosleep(&nap, NULL); + } +} + +void MR_ResumeAfterFork(void) { + if (!mrCtx.executionsThreadPool) return; + RedisModule_Log(mr_staticCtx, "notice", "MOD-15307: resuming LibMR threads after fork()"); + pthread_mutex_lock(&mr_forkDrainLock); + mr_forkElRelease = 1; + pthread_cond_broadcast(&mr_forkDrainCond); + pthread_mutex_unlock(&mr_forkDrainLock); +} + void MR_Run(Execution* e) { /* take ownership on the execution */ __atomic_add_fetch(&e->refCount, 1, __ATOMIC_RELAXED); diff --git a/src/mr.h b/src/mr.h index 19b203d1..fffccb64 100644 --- a/src/mr.h +++ b/src/mr.h @@ -182,6 +182,14 @@ LIBMR_API bool MR_IsInternalCommandsExecution(const Execution* e); /* Free the given execution */ LIBMR_API void MR_FreeExecution(Execution* e); +/* MOD-15307: drain LibMR background threads (worker pool + event-loop thread) to a safe, + * lock-free point before fork(), so the forked child does not inherit a libc lock held by a + * LibMR thread (ghost-lock). Bounded (a few seconds), then proceeds anyway (fail-open). Call + * on the main thread from the FORK_CHILD_PRE module event; pair every call with exactly one + * MR_ResumeAfterFork() (on FORK_CHILD_BORN, or if the fork was cancelled). */ +LIBMR_API void MR_DrainForFork(void); +LIBMR_API void MR_ResumeAfterFork(void); + /* Initialize mr library */ LIBMR_API int MR_Init(struct RedisModuleCtx* ctx, size_t numThreads, char *password); From 0b2a3b19cdfa0d1cd6cb8a7e0f2be161af105e2e Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Thu, 11 Jun 2026 13:20:50 +0300 Subject: [PATCH 2/6] Add debug-level drain-only timing to MR_DrainForFork Logs the time spent quiescing the threads (excluding the fork that follows), plus whether the event-loop thread parked and the busy-worker count, so the pre-fork drain cost can be measured. Debug level so it is silent in production. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/mr.c | 14 ++++++++++++-- 1 file changed, 12 insertions(+), 2 deletions(-) diff --git a/src/mr.c b/src/mr.c index 3aabe8fe..b97e0ec8 100644 --- a/src/mr.c +++ b/src/mr.c @@ -1461,7 +1461,10 @@ void MR_DrainForFork(void) { /* Nothing to drain if the pool / event loop were never started. */ if (!mrCtx.executionsThreadPool) return; - RedisModule_Log(mr_staticCtx, "notice", "MOD-15307: draining LibMR threads before fork()"); + /* Precise drain-only timing: measures just the quiesce (park + worker wait), + * excluding the fork() that follows. */ + struct timespec _drain_t0; + clock_gettime(CLOCK_MONOTONIC, &_drain_t0); struct timespec deadline; clock_gettime(CLOCK_REALTIME, &deadline); @@ -1504,11 +1507,18 @@ void MR_DrainForFork(void) { struct timespec nap = { .tv_sec = 0, .tv_nsec = 1000000L }; /* 1ms */ nanosleep(&nap, NULL); } + + struct timespec _drain_t1; + clock_gettime(CLOCK_MONOTONIC, &_drain_t1); + long long _drain_us = (long long)(_drain_t1.tv_sec - _drain_t0.tv_sec) * 1000000LL + + (_drain_t1.tv_nsec - _drain_t0.tv_nsec) / 1000LL; + RedisModule_Log(mr_staticCtx, "debug", + "MR_DrainForFork drained in %lld us (el_parked=%d, busy_workers=%d)", + _drain_us, parked, mr_thpool_num_threads_working(mrCtx.executionsThreadPool)); } void MR_ResumeAfterFork(void) { if (!mrCtx.executionsThreadPool) return; - RedisModule_Log(mr_staticCtx, "notice", "MOD-15307: resuming LibMR threads after fork()"); pthread_mutex_lock(&mr_forkDrainLock); mr_forkElRelease = 1; pthread_cond_broadcast(&mr_forkDrainCond); From a774007cc50439d8c496d01e3539d50ff71c4986 Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Thu, 11 Jun 2026 14:29:44 +0300 Subject: [PATCH 3/6] MOD-14615: log which shard stalled on max-idle + dispatch timing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a multi-shard execution hits max-idle (EXECUTION_DEFAULT_MAX_IDLE_MS), the coordinator only logged a fixed "execution max idle reached" string with no clue which peer stalled. Now, at timeout, log the non-responding peer shard id(s) and endpoint(s), replies received vs expected, the elapsed wait, and time since last progress — at warning level so it is captured at the default loglevel (it coincides with the user-visible failure and is rate-bounded by nMaxIdleReached). Track responders by recording each ACK / NOTIFY_DONE sender node-id into a heap-strings set on the Execution (freed in MR_FreeExecution, the single final owner); pending = cluster peers minus that set, formatted by a new MR_ClusterFormatPendingPeers helper. Dispatch time is stamped and logged at debug. All new state is touched only on the event-loop thread, so no extra locking. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/cluster.c | 25 +++++++++++++++++++++ src/cluster.h | 7 ++++++ src/mr.c | 60 +++++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 92 insertions(+) diff --git a/src/cluster.c b/src/cluster.c index f629ffe9..1b2ddc6b 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -1731,6 +1731,31 @@ size_t MR_ClusterGetSize(){ return clusterCtx.clusterSize; } +/* MOD-14615: format the peer shards that are NOT me and NOT present in the + * `responded` set into "id(ip:port),..." in `out`, so a max-idle timeout can + * name the non-responding shard(s). Runs on the event-loop thread (same as the + * message handlers), so reading the cluster node table needs no extra locking. + * The node-id is the same NUL-terminated string used as the responded-set key. */ +void MR_ClusterFormatPendingPeers(mr_dict* responded, char* out, size_t outLen) { + if (outLen == 0) return; + out[0] = '\0'; + if (!clusterCtx.CurrCluster || !clusterCtx.CurrCluster->nodes) return; + mr_dictIterator *iter = mr_dictGetIterator(clusterCtx.CurrCluster->nodes); + mr_dictEntry *entry = NULL; + size_t off = 0; + while ((entry = mr_dictNext(iter))) { + Node* n = mr_dictGetVal(entry); + if (n->isMe) continue; + if (responded && mr_dictFind(responded, n->id)) continue; + int w = snprintf(out + off, outLen - off, "%s%s(%s:%u)", + off ? "," : "", n->id, n->ip ? n->ip : "?", n->port); + if (w < 0) break; + if ((size_t)w >= outLen - off) { out[outLen - 1] = '\0'; break; } /* truncated */ + off += (size_t)w; + } + mr_dictReleaseIterator(iter); +} + int MR_ClusterIsClusterMode(){ return MR_ClusterGetSize() > 1; } diff --git a/src/cluster.h b/src/cluster.h index e2c3995c..fece52d0 100644 --- a/src/cluster.h +++ b/src/cluster.h @@ -35,6 +35,13 @@ int MR_IsClusterInitialize(); size_t MR_ClusterGetSize(); +/* MOD-14615: format peer shards (id(ip:port)) that are NOT in `responded` into + * `out`, used to name non-responding shards on a max-idle timeout. The tag is + * forward-declared so cluster.h stays independent of utils/dict.h include order + * (full definition lives in utils/dict.h). */ +struct mr_dict; +void MR_ClusterFormatPendingPeers(struct mr_dict* responded, char* out, size_t outLen); + int MR_ClusterInit(RedisModuleCtx* rctx, char *password); size_t MR_ClusterGetSlotByKey(const char* key, size_t len); diff --git a/src/mr.c b/src/mr.c index b97e0ec8..846a4dab 100644 --- a/src/mr.c +++ b/src/mr.c @@ -288,6 +288,16 @@ struct Execution { ExecutionCallbacks callbacks; MR_LoopTaskCtx* timeoutTask; size_t timeoutMS; + + /* MOD-14615 diagnostics: so a max-idle timeout can name the shard(s) that + * did not reply. respondedAck/respondedDone hold the sender node-ids that + * sent an ACK / NOTIFY_DONE; pending = cluster peers minus that set. The + * timestamps give the elapsed wait. Touched only on the event-loop thread + * (ack/done/timeout handlers), so no extra locking is needed. */ + mr_dict* respondedAck; + mr_dict* respondedDone; + long long dispatchMonoMs; + long long lastProgressMonoMs; }; struct ExecutionCtx { @@ -533,9 +543,24 @@ static Execution* MR_ExecutionAlloc() { e->timeoutTask = NULL; e->timeoutMS = EXECUTION_DEFAULT_MAX_IDLE_MS; e->flags = 0; + /* MOD-14615 diagnostics. Node-ids are NUL-terminated strings, so use the + * heap-strings dict type (matches the cluster node table), NOT the fixed + * ID_LEN dict type used for execution ids. */ + e->respondedAck = mr_dictCreate(&mr_dictTypeHeapStrings, NULL); + e->respondedDone = mr_dictCreate(&mr_dictTypeHeapStrings, NULL); + e->dispatchMonoMs = 0; + e->lastProgressMonoMs = 0; return e; } +/* MOD-14615: monotonic milliseconds for measuring how long an execution waited + * for peer replies. CLOCK_MONOTONIC so it is unaffected by wall-clock changes. */ +static long long mr_nowMonoMs(void) { + struct timespec t; + clock_gettime(CLOCK_MONOTONIC, &t); + return (long long)t.tv_sec * 1000LL + t.tv_nsec / 1000000LL; +} + Execution* MR_CreateExecution(ExecutionBuilder* builder, MRError** err) { if (!MR_IsClusterInitialize()) { *err = &UINITIALIZED_CLUSTER_ERROR; @@ -1018,6 +1043,8 @@ static void MR_NotifyDone(RedisModuleCtx *ctx, const char *sender_id, uint8_t ty return; } + if (sender_id) mr_dictAdd(e->respondedDone, (void*)sender_id, NULL); /* MOD-14615 */ + e->lastProgressMonoMs = mr_nowMonoMs(); ++e->nCompleted; if (e->nCompleted == MR_ClusterGetSize() - 1) { /* Execution is finished on all the shards, @@ -1079,6 +1106,8 @@ static void MR_AckExecution(RedisModuleCtx *ctx, const char *sender_id, uint8_t return; } + if (sender_id) mr_dictAdd(e->respondedAck, (void*)sender_id, NULL); /* MOD-14615 */ + e->lastProgressMonoMs = mr_nowMonoMs(); ++e->nReceived; if (e->nReceived == MR_ClusterGetSize() - 1) { /* all shards have received the execution, we can invoke it. */ @@ -1295,6 +1324,12 @@ static void MR_ExecutionDistribute(Execution* e, void* pd) { MR_ClusterSendMsg(NULL, fid, buff.buff, buff.size); /* now we wait for shards to respond that they got the execution */ + /* MOD-14615: stamp dispatch so a max-idle timeout can report the elapsed + * wait. Debug-level (per-execution, would be noisy at notice). */ + e->dispatchMonoMs = e->lastProgressMonoMs = mr_nowMonoMs(); + RedisModule_Log(mr_staticCtx, "debug", + "MR execution %s dispatched to %zu peers (timeoutMS=%zu)", + e->idStr, MR_ClusterGetSize() - 1, e->timeoutMS); } static void MR_ExecutionTimedOutInternal(Execution* e, void* pd) { @@ -1312,6 +1347,29 @@ static void MR_ExecutionTimedOut(void* ctx) { /* execution timed out */ e->timeoutTask = NULL; ++mrCtx.stats.nMaxIdleReached; + + /* MOD-14615: name the shard(s) that did not reply before we discard the + * execution. We are stalled in the ACK phase if not all shards acknowledged + * receipt, otherwise in the DONE phase; pick the matching responded set so + * `pending` lists exactly the non-responders. Warning level: this coincides + * with the user-visible multi-shard failure and is rate-bounded by + * nMaxIdleReached, and warning IS captured at the default loglevel. */ + { + size_t expected = MR_ClusterGetSize() - 1; + long long now = mr_nowMonoMs(); + int ackPhase = (e->nReceived < expected); + mr_dict* got = ackPhase ? e->respondedAck : e->respondedDone; + size_t nGot = ackPhase ? e->nReceived : e->nCompleted; + char pending[1024]; + MR_ClusterFormatPendingPeers(got, pending, sizeof(pending)); + RedisModule_Log(mr_staticCtx, "warning", + "MR execution %s max-idle reached: phase=%s replies=%zu/%zu waited=%lldms " + "sinceLastProgress=%lldms pending=[%s]", + e->idStr, ackPhase ? "ack" : "done", nGot, expected, + e->dispatchMonoMs ? now - e->dispatchMonoMs : -1, + e->lastProgressMonoMs ? now - e->lastProgressMonoMs : -1, pending); + } + /* Delete the execution from the executions dictionary, * We will ignore further messages on this execution. */ mr_dictDelete(mrCtx.executionsDict, e->id); @@ -1640,6 +1698,8 @@ void MR_FreeExecution(Execution* e) { MR_RecordFree(e->errors[i]); } array_free(e->errors); + if (e->respondedAck) mr_dictRelease(e->respondedAck); /* MOD-14615 */ + if (e->respondedDone) mr_dictRelease(e->respondedDone); /* MOD-14615 */ MR_FREE(e); } From f6fe68792a7d72247716210f114ef8ba83bcf734 Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Wed, 17 Jun 2026 21:41:49 +0300 Subject: [PATCH 4/6] fork-drain: also wait for queued (not just in-flight) jobs MR_DrainForFork only waited for mr_thpool_num_threads_working() to reach 0, ignoring jobs that are queued but not yet picked up by a worker. Such a job could be dequeued and start running (and allocating) immediately after the check, right as fork() runs -- defeating the drain. Add mr_thpool_num_jobs_in_queue() and wait for BOTH the working count and the queue to reach 0 (the same invariant thpool_wait uses). Addresses the bugbot "fork drain ignores queued jobs" finding. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/mr.c | 21 +++++++++++++-------- src/utils/thpool.c | 4 ++++ src/utils/thpool.h | 8 ++++++++ 3 files changed, 25 insertions(+), 8 deletions(-) diff --git a/src/mr.c b/src/mr.c index 0a8af925..c94cbe1d 100644 --- a/src/mr.c +++ b/src/mr.c @@ -1552,14 +1552,18 @@ void MR_DrainForFork(void) { MR_FORK_DRAIN_TIMEOUT_MS); } - /* 2) Wait (bounded) for in-flight worker jobs to finish. With the event loop parked no - * new work is dispatched, so the working count drains to 0 unless a worker is blocked - * acquiring the GIL we hold -- a malloc-safe state, so a timeout there is benign. */ - while (mr_thpool_num_threads_working(mrCtx.executionsThreadPool) > 0) { + /* 2) Wait (bounded) for the worker pool to go fully idle -- both in-flight jobs AND jobs + * still queued but not yet picked up (a queued job would otherwise start running, and + * allocate, right after the fork). With the event loop parked no new work is dispatched, + * so the pool drains to empty unless a worker is blocked acquiring the GIL we hold -- a + * malloc-safe state, so a timeout there is benign. */ + while (mr_thpool_num_threads_working(mrCtx.executionsThreadPool) > 0 || + mr_thpool_num_jobs_in_queue(mrCtx.executionsThreadPool) > 0) { if (mr_forkDeadlinePassed(&deadline)) { RedisModule_Log(mr_staticCtx, "warning", - "MR_DrainForFork: %d worker(s) still busy at timeout; forking anyway", - mr_thpool_num_threads_working(mrCtx.executionsThreadPool)); + "MR_DrainForFork: pool not idle at timeout (busy=%d, queued=%d); forking anyway", + mr_thpool_num_threads_working(mrCtx.executionsThreadPool), + mr_thpool_num_jobs_in_queue(mrCtx.executionsThreadPool)); break; } struct timespec nap = { .tv_sec = 0, .tv_nsec = 1000000L }; /* 1ms */ @@ -1571,8 +1575,9 @@ void MR_DrainForFork(void) { long long _drain_us = (long long)(_drain_t1.tv_sec - _drain_t0.tv_sec) * 1000000LL + (_drain_t1.tv_nsec - _drain_t0.tv_nsec) / 1000LL; RedisModule_Log(mr_staticCtx, "debug", - "MR_DrainForFork drained in %lld us (el_parked=%d, busy_workers=%d)", - _drain_us, parked, mr_thpool_num_threads_working(mrCtx.executionsThreadPool)); + "MR_DrainForFork drained in %lld us (el_parked=%d, busy_workers=%d, queued=%d)", + _drain_us, parked, mr_thpool_num_threads_working(mrCtx.executionsThreadPool), + mr_thpool_num_jobs_in_queue(mrCtx.executionsThreadPool)); } void MR_ResumeAfterFork(void) { diff --git a/src/utils/thpool.c b/src/utils/thpool.c index 92b44b21..2a5383b8 100644 --- a/src/utils/thpool.c +++ b/src/utils/thpool.c @@ -314,6 +314,10 @@ int mr_thpool_num_threads_working(mr_thpool_* thpool_p) { return thpool_p->num_threads_working; } +int mr_thpool_num_jobs_in_queue(mr_thpool_* thpool_p) { + return thpool_p->jobqueue.len; +} + /* ============================ THREAD ============================== */ /* Initialize a thread in the thread pool diff --git a/src/utils/thpool.h b/src/utils/thpool.h index ddf471c3..800d8907 100755 --- a/src/utils/thpool.h +++ b/src/utils/thpool.h @@ -188,6 +188,14 @@ void mr_thpool_destroy(mr_threadpool); */ int mr_thpool_num_threads_working(mr_threadpool); +/** + * @brief Number of jobs queued but not yet picked up by a worker. + * + * @param threadpool the threadpool of interest + * @return integer number of queued jobs + */ +int mr_thpool_num_jobs_in_queue(mr_threadpool); + #ifdef __cplusplus } From 995f84211fb947a011fbfe433042274b33e07c1e Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Thu, 18 Jun 2026 07:22:58 +0300 Subject: [PATCH 5/6] fork-drain: read job-queue length under the queue lock mr_thpool_num_jobs_in_queue() read jobqueue.len without holding jobqueue.rwmutex, while jobqueue_push/pull mutate it under that lock -- a data race that could make MR_DrainForFork see a stale (zero) queue and fork early. Take rwmutex for the read. Addresses the bugbot "unlocked job queue length read" finding. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/utils/thpool.c | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/utils/thpool.c b/src/utils/thpool.c index 2a5383b8..5927df97 100644 --- a/src/utils/thpool.c +++ b/src/utils/thpool.c @@ -315,7 +315,12 @@ int mr_thpool_num_threads_working(mr_thpool_* thpool_p) { } int mr_thpool_num_jobs_in_queue(mr_thpool_* thpool_p) { - return thpool_p->jobqueue.len; + /* len is written under rwmutex (jobqueue_push/pull); read it under the same + * lock to avoid a data race. */ + pthread_mutex_lock(&thpool_p->jobqueue.rwmutex); + int len = thpool_p->jobqueue.len; + pthread_mutex_unlock(&thpool_p->jobqueue.rwmutex); + return len; } /* ============================ THREAD ============================== */ From df7cc0902ab672e4722f01f8ee3af1cb2d2da8d0 Mon Sep 17 00:00:00 2001 From: Tom Gabsow Date: Sun, 21 Jun 2026 18:36:56 +0300 Subject: [PATCH 6/6] fork-drain: simplify and drop the max-idle diagnostics (review feedback) - Wait for the worker pool to drain via the pool's own bounded idle wait (new mr_thpool_wait_timeout, reusing thcount_lock/threads_all_idle) instead of a hand-rolled poll loop plus a per-read jobqueue lock. Removes the hot-path lock contention that risked added command latency. - Remove the max-idle "which shard didn't reply" diagnostics entirely (responded-sets, monotonic timestamps, dispatch/timeout logging, MR_ClusterFormatPendingPeers, mr_thpool_num_jobs_in_queue) -- out of scope for the fork-coordination fix; will be tracked in a separate ticket if needed. - Trim comments (no ticket numbers / implementation rationale). Co-Authored-By: Claude Opus 4.8 (1M context) --- src/cluster.c | 25 -------- src/cluster.h | 7 --- src/mr.c | 144 +++++---------------------------------------- src/utils/thpool.c | 29 ++++++--- src/utils/thpool.h | 14 ++--- 5 files changed, 42 insertions(+), 177 deletions(-) diff --git a/src/cluster.c b/src/cluster.c index 8df59fa3..373f00e8 100644 --- a/src/cluster.c +++ b/src/cluster.c @@ -1800,31 +1800,6 @@ size_t MR_ClusterGetSize(){ return clusterCtx.clusterSize; } -/* MOD-14615: format the peer shards that are NOT me and NOT present in the - * `responded` set into "id(ip:port),..." in `out`, so a max-idle timeout can - * name the non-responding shard(s). Runs on the event-loop thread (same as the - * message handlers), so reading the cluster node table needs no extra locking. - * The node-id is the same NUL-terminated string used as the responded-set key. */ -void MR_ClusterFormatPendingPeers(mr_dict* responded, char* out, size_t outLen) { - if (outLen == 0) return; - out[0] = '\0'; - if (!clusterCtx.CurrCluster || !clusterCtx.CurrCluster->nodes) return; - mr_dictIterator *iter = mr_dictGetIterator(clusterCtx.CurrCluster->nodes); - mr_dictEntry *entry = NULL; - size_t off = 0; - while ((entry = mr_dictNext(iter))) { - Node* n = mr_dictGetVal(entry); - if (n->isMe) continue; - if (responded && mr_dictFind(responded, n->id)) continue; - int w = snprintf(out + off, outLen - off, "%s%s(%s:%u)", - off ? "," : "", n->id, n->ip ? n->ip : "?", n->port); - if (w < 0) break; - if ((size_t)w >= outLen - off) { out[outLen - 1] = '\0'; break; } /* truncated */ - off += (size_t)w; - } - mr_dictReleaseIterator(iter); -} - int MR_ClusterIsClusterMode(){ return MR_ClusterGetSize() > 1; } diff --git a/src/cluster.h b/src/cluster.h index fece52d0..e2c3995c 100644 --- a/src/cluster.h +++ b/src/cluster.h @@ -35,13 +35,6 @@ int MR_IsClusterInitialize(); size_t MR_ClusterGetSize(); -/* MOD-14615: format peer shards (id(ip:port)) that are NOT in `responded` into - * `out`, used to name non-responding shards on a max-idle timeout. The tag is - * forward-declared so cluster.h stays independent of utils/dict.h include order - * (full definition lives in utils/dict.h). */ -struct mr_dict; -void MR_ClusterFormatPendingPeers(struct mr_dict* responded, char* out, size_t outLen); - int MR_ClusterInit(RedisModuleCtx* rctx, char *password); size_t MR_ClusterGetSlotByKey(const char* key, size_t len); diff --git a/src/mr.c b/src/mr.c index c94cbe1d..9715d949 100644 --- a/src/mr.c +++ b/src/mr.c @@ -288,16 +288,6 @@ struct Execution { ExecutionCallbacks callbacks; MR_LoopTaskCtx* timeoutTask; size_t timeoutMS; - - /* MOD-14615 diagnostics: so a max-idle timeout can name the shard(s) that - * did not reply. respondedAck/respondedDone hold the sender node-ids that - * sent an ACK / NOTIFY_DONE; pending = cluster peers minus that set. The - * timestamps give the elapsed wait. Touched only on the event-loop thread - * (ack/done/timeout handlers), so no extra locking is needed. */ - mr_dict* respondedAck; - mr_dict* respondedDone; - long long dispatchMonoMs; - long long lastProgressMonoMs; }; struct ExecutionCtx { @@ -543,24 +533,9 @@ static Execution* MR_ExecutionAlloc() { e->timeoutTask = NULL; e->timeoutMS = EXECUTION_DEFAULT_MAX_IDLE_MS; e->flags = 0; - /* MOD-14615 diagnostics. Node-ids are NUL-terminated strings, so use the - * heap-strings dict type (matches the cluster node table), NOT the fixed - * ID_LEN dict type used for execution ids. */ - e->respondedAck = mr_dictCreate(&mr_dictTypeHeapStrings, NULL); - e->respondedDone = mr_dictCreate(&mr_dictTypeHeapStrings, NULL); - e->dispatchMonoMs = 0; - e->lastProgressMonoMs = 0; return e; } -/* MOD-14615: monotonic milliseconds for measuring how long an execution waited - * for peer replies. CLOCK_MONOTONIC so it is unaffected by wall-clock changes. */ -static long long mr_nowMonoMs(void) { - struct timespec t; - clock_gettime(CLOCK_MONOTONIC, &t); - return (long long)t.tv_sec * 1000LL + t.tv_nsec / 1000000LL; -} - Execution* MR_CreateExecution(ExecutionBuilder* builder, MRError** err) { if (!MR_IsClusterInitialize()) { *err = &UINITIALIZED_CLUSTER_ERROR; @@ -1043,8 +1018,6 @@ static void MR_NotifyDone(RedisModuleCtx *ctx, const char *sender_id, uint8_t ty return; } - if (sender_id) mr_dictAdd(e->respondedDone, (void*)sender_id, NULL); /* MOD-14615 */ - e->lastProgressMonoMs = mr_nowMonoMs(); ++e->nCompleted; if (e->nCompleted == MR_ClusterGetSize() - 1) { /* Execution is finished on all the shards, @@ -1106,8 +1079,6 @@ static void MR_AckExecution(RedisModuleCtx *ctx, const char *sender_id, uint8_t return; } - if (sender_id) mr_dictAdd(e->respondedAck, (void*)sender_id, NULL); /* MOD-14615 */ - e->lastProgressMonoMs = mr_nowMonoMs(); ++e->nReceived; if (e->nReceived == MR_ClusterGetSize() - 1) { /* all shards have received the execution, we can invoke it. */ @@ -1324,12 +1295,6 @@ static void MR_ExecutionDistribute(Execution* e, void* pd) { MR_ClusterSendMsg(NULL, fid, buff.buff, buff.size); /* now we wait for shards to respond that they got the execution */ - /* MOD-14615: stamp dispatch so a max-idle timeout can report the elapsed - * wait. Debug-level (per-execution, would be noisy at notice). */ - e->dispatchMonoMs = e->lastProgressMonoMs = mr_nowMonoMs(); - RedisModule_Log(mr_staticCtx, "debug", - "MR execution %s dispatched to %zu peers (timeoutMS=%zu)", - e->idStr, MR_ClusterGetSize() - 1, e->timeoutMS); } static void MR_ExecutionTimedOutInternal(Execution* e, void* pd) { @@ -1347,29 +1312,6 @@ static void MR_ExecutionTimedOut(void* ctx) { /* execution timed out */ e->timeoutTask = NULL; ++mrCtx.stats.nMaxIdleReached; - - /* MOD-14615: name the shard(s) that did not reply before we discard the - * execution. We are stalled in the ACK phase if not all shards acknowledged - * receipt, otherwise in the DONE phase; pick the matching responded set so - * `pending` lists exactly the non-responders. Warning level: this coincides - * with the user-visible multi-shard failure and is rate-bounded by - * nMaxIdleReached, and warning IS captured at the default loglevel. */ - { - size_t expected = MR_ClusterGetSize() - 1; - long long now = mr_nowMonoMs(); - int ackPhase = (e->nReceived < expected); - mr_dict* got = ackPhase ? e->respondedAck : e->respondedDone; - size_t nGot = ackPhase ? e->nReceived : e->nCompleted; - char pending[1024]; - MR_ClusterFormatPendingPeers(got, pending, sizeof(pending)); - RedisModule_Log(mr_staticCtx, "warning", - "MR execution %s max-idle reached: phase=%s replies=%zu/%zu waited=%lldms " - "sinceLastProgress=%lldms pending=[%s]", - e->idStr, ackPhase ? "ack" : "done", nGot, expected, - e->dispatchMonoMs ? now - e->dispatchMonoMs : -1, - e->lastProgressMonoMs ? now - e->lastProgressMonoMs : -1, pending); - } - /* Delete the execution from the executions dictionary, * We will ignore further messages on this execution. */ mr_dictDelete(mrCtx.executionsDict, e->id); @@ -1468,34 +1410,19 @@ void MR_ExecutionSetMaxIdle(Execution* e, size_t maxIdle) { e->timeoutMS = maxIdle; } -/* ---- MOD-15307: drain LibMR background threads to a safe point before fork() ---- - * - * A LibMR thread (a worker in the execution pool, or the event-loop thread) that holds - * a libc lock (e.g. the malloc arena lock) at the instant redis calls fork() leaves the - * child holding a locked mutex with no owner -> the child ghost-locks the first time it - * mallocs (RDB save, slot-migration snapshot, etc.). That is the root trigger behind the - * ASM-migration nightly hangs (MOD-15307) and, downstream, the multi-shard query max-idle - * timeout (MOD-14615). - * - * MR_DrainForFork() (called on the main thread from the FORK_CHILD_PRE module event) brings - * both kinds of LibMR thread to a safe, lock-free point: - * - the event-loop thread is parked *between tasks* via a posted task (so it is not - * mid-message-deserialize / mid-malloc, and is not holding the module GIL); - * - with the event-loop thread parked no new executions are dispatched, so the worker - * pool drains to idle. - * Both waits are bounded; on timeout we fork anyway (fail-open -> never worse than today). - * Note a worker still blocked acquiring the module GIL (held by us) is itself malloc-safe, - * so a timeout there is benign. MR_ResumeAfterFork() releases the parked event-loop thread - * and must be called after fork() (FORK_CHILD_BORN) or if the fork was cancelled. */ +/* Drain LibMR background threads to a safe point before fork(): a thread holding a + * libc lock (e.g. the allocator lock) at fork() would leave the child holding a locked + * mutex with no owner and deadlock it. MR_DrainForFork() parks the event-loop thread + * (between tasks, so it stops dispatching and isn't holding the GIL) and waits for the + * worker pool to go idle; MR_ResumeAfterFork() releases it after FORK_CHILD_BORN/CANCELLED. */ #define MR_FORK_DRAIN_TIMEOUT_MS 2000 static pthread_mutex_t mr_forkDrainLock = PTHREAD_MUTEX_INITIALIZER; static pthread_cond_t mr_forkDrainCond = PTHREAD_COND_INITIALIZER; -static int mr_forkElParked = 0; /* set by the park task once the el-thread is quiesced */ -static int mr_forkElRelease = 0; /* set by MR_ResumeAfterFork to release the el-thread */ +static int mr_forkElParked = 0; +static int mr_forkElRelease = 0; -/* Runs on the event-loop thread, between tasks (a safe point). Parks the thread until - * MR_ResumeAfterFork() is called. */ +/* Runs on the event-loop thread between tasks (a safe point); parks it until resumed. */ static void MR_ForkParkElThread(void* ctx) { REDISMODULE_NOT_USED(ctx); pthread_mutex_lock(&mr_forkDrainLock); @@ -1508,29 +1435,16 @@ static void MR_ForkParkElThread(void* ctx) { pthread_mutex_unlock(&mr_forkDrainLock); } -static int mr_forkDeadlinePassed(const struct timespec* deadline) { - struct timespec now; - clock_gettime(CLOCK_REALTIME, &now); - return (now.tv_sec > deadline->tv_sec) || - (now.tv_sec == deadline->tv_sec && now.tv_nsec >= deadline->tv_nsec); -} - void MR_DrainForFork(void) { - /* Nothing to drain if the pool / event loop were never started. */ if (!mrCtx.executionsThreadPool) return; - /* Precise drain-only timing: measures just the quiesce (park + worker wait), - * excluding the fork() that follows. */ - struct timespec _drain_t0; - clock_gettime(CLOCK_MONOTONIC, &_drain_t0); - struct timespec deadline; clock_gettime(CLOCK_REALTIME, &deadline); deadline.tv_sec += MR_FORK_DRAIN_TIMEOUT_MS / 1000; deadline.tv_nsec += (MR_FORK_DRAIN_TIMEOUT_MS % 1000) * 1000000L; if (deadline.tv_nsec >= 1000000000L) { deadline.tv_sec++; deadline.tv_nsec -= 1000000000L; } - /* 1) Park the event-loop thread at a between-tasks safe point. */ + /* Park the event-loop thread between tasks so it stops dispatching work. */ pthread_mutex_lock(&mr_forkDrainLock); mr_forkElParked = 0; mr_forkElRelease = 0; @@ -1546,38 +1460,14 @@ void MR_DrainForFork(void) { int parked = mr_forkElParked; pthread_mutex_unlock(&mr_forkDrainLock); - if (!parked) { - RedisModule_Log(mr_staticCtx, "warning", - "MR_DrainForFork: event-loop thread did not park within %dms; forking anyway", - MR_FORK_DRAIN_TIMEOUT_MS); - } - - /* 2) Wait (bounded) for the worker pool to go fully idle -- both in-flight jobs AND jobs - * still queued but not yet picked up (a queued job would otherwise start running, and - * allocate, right after the fork). With the event loop parked no new work is dispatched, - * so the pool drains to empty unless a worker is blocked acquiring the GIL we hold -- a - * malloc-safe state, so a timeout there is benign. */ - while (mr_thpool_num_threads_working(mrCtx.executionsThreadPool) > 0 || - mr_thpool_num_jobs_in_queue(mrCtx.executionsThreadPool) > 0) { - if (mr_forkDeadlinePassed(&deadline)) { - RedisModule_Log(mr_staticCtx, "warning", - "MR_DrainForFork: pool not idle at timeout (busy=%d, queued=%d); forking anyway", - mr_thpool_num_threads_working(mrCtx.executionsThreadPool), - mr_thpool_num_jobs_in_queue(mrCtx.executionsThreadPool)); - break; - } - struct timespec nap = { .tv_sec = 0, .tv_nsec = 1000000L }; /* 1ms */ - nanosleep(&nap, NULL); - } + /* With the event loop parked no new work is dispatched; wait for the worker pool + * to go idle via its own idle signal. */ + int idle = mr_thpool_wait_timeout(mrCtx.executionsThreadPool, MR_FORK_DRAIN_TIMEOUT_MS); - struct timespec _drain_t1; - clock_gettime(CLOCK_MONOTONIC, &_drain_t1); - long long _drain_us = (long long)(_drain_t1.tv_sec - _drain_t0.tv_sec) * 1000000LL + - (_drain_t1.tv_nsec - _drain_t0.tv_nsec) / 1000LL; - RedisModule_Log(mr_staticCtx, "debug", - "MR_DrainForFork drained in %lld us (el_parked=%d, busy_workers=%d, queued=%d)", - _drain_us, parked, mr_thpool_num_threads_working(mrCtx.executionsThreadPool), - mr_thpool_num_jobs_in_queue(mrCtx.executionsThreadPool)); + if (!parked || !idle) + RedisModule_Log(mr_staticCtx, "warning", + "MR_DrainForFork: not fully quiesced before fork (el_parked=%d, pool_idle=%d)", + parked, idle); } void MR_ResumeAfterFork(void) { @@ -1703,8 +1593,6 @@ void MR_FreeExecution(Execution* e) { MR_RecordFree(e->errors[i]); } array_free(e->errors); - if (e->respondedAck) mr_dictRelease(e->respondedAck); /* MOD-14615 */ - if (e->respondedDone) mr_dictRelease(e->respondedDone); /* MOD-14615 */ MR_FREE(e); } diff --git a/src/utils/thpool.c b/src/utils/thpool.c index 5927df97..cf8d6497 100644 --- a/src/utils/thpool.c +++ b/src/utils/thpool.c @@ -254,6 +254,26 @@ void mr_thpool_wait(mr_thpool_* thpool_p) { pthread_mutex_unlock(&thpool_p->thcount_lock); } +/* Like mr_thpool_wait but bounded; returns 1 if the pool went idle, 0 on timeout. */ +int mr_thpool_wait_timeout(mr_thpool_* thpool_p, long timeout_ms) { + struct timespec deadline; + clock_gettime(CLOCK_REALTIME, &deadline); + deadline.tv_sec += timeout_ms / 1000; + deadline.tv_nsec += (timeout_ms % 1000) * 1000000L; + if (deadline.tv_nsec >= 1000000000L) { deadline.tv_sec++; deadline.tv_nsec -= 1000000000L; } + + int idle = 1; + pthread_mutex_lock(&thpool_p->thcount_lock); + while (thpool_p->jobqueue.len || thpool_p->num_threads_working) { + if (pthread_cond_timedwait(&thpool_p->threads_all_idle, &thpool_p->thcount_lock, &deadline) == ETIMEDOUT) { + idle = 0; + break; + } + } + pthread_mutex_unlock(&thpool_p->thcount_lock); + return idle; +} + /* Destroy the threadpool */ void mr_thpool_destroy(mr_thpool_* thpool_p) { /* No need to destory if it's NULL */ @@ -314,15 +334,6 @@ int mr_thpool_num_threads_working(mr_thpool_* thpool_p) { return thpool_p->num_threads_working; } -int mr_thpool_num_jobs_in_queue(mr_thpool_* thpool_p) { - /* len is written under rwmutex (jobqueue_push/pull); read it under the same - * lock to avoid a data race. */ - pthread_mutex_lock(&thpool_p->jobqueue.rwmutex); - int len = thpool_p->jobqueue.len; - pthread_mutex_unlock(&thpool_p->jobqueue.rwmutex); - return len; -} - /* ============================ THREAD ============================== */ /* Initialize a thread in the thread pool diff --git a/src/utils/thpool.h b/src/utils/thpool.h index 800d8907..2aeb4103 100755 --- a/src/utils/thpool.h +++ b/src/utils/thpool.h @@ -105,6 +105,12 @@ int mr_thpool_add_work(mr_threadpool, void (*function_p)(void*), void* arg_p); */ void mr_thpool_wait(mr_threadpool); +/** + * @brief Like mr_thpool_wait, bounded by timeout_ms. + * @return 1 if the pool went idle, 0 on timeout. + */ +int mr_thpool_wait_timeout(mr_threadpool, long timeout_ms); + /** * @brief Pauses all threads immediately @@ -188,14 +194,6 @@ void mr_thpool_destroy(mr_threadpool); */ int mr_thpool_num_threads_working(mr_threadpool); -/** - * @brief Number of jobs queued but not yet picked up by a worker. - * - * @param threadpool the threadpool of interest - * @return integer number of queued jobs - */ -int mr_thpool_num_jobs_in_queue(mr_threadpool); - #ifdef __cplusplus }