chore: starvation of background fibers under high task queue load#592
chore: starvation of background fibers under high task queue load#592kostasrim wants to merge 1 commit into
Conversation
Signed-off-by: Kostas Kyrimis <kostas@dragonflydb.io>
📝 WalkthroughWalkthroughThis PR adds a regression test for background fiber starvation in the Helio fiber scheduler. The test ChangesBackground Fiber Starvation Regression Test
Estimated code review effort🎯 2 (Simple) | ⏱️ ~10 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a new regression test, BackgroundFiberNotStarved, to verify that background fibers are not starved under specific execution conditions. The review feedback identifies a critical lifetime and use-after-free vulnerability in the test where the asynchronous spammer task captures local variables by reference and could execute after the test function exits. It is recommended to block the test thread until the last queued task has finished executing.
| stop = true; | ||
| bg.Join(); |
There was a problem hiding this comment.
Lifetime / Use-After-Free Issue in Spammer Task
The spammer lambda captures several local variables (such as stop, normal_done, bg_wake, and spammer itself) by reference.
When the test completes, these local variables are destroyed. However, because the spammer task is dispatched asynchronously via DispatchBrief, there is no guarantee that the last dispatched spammer task has finished executing on the proactor thread before the test function exits. This can lead to a use-after-free / undefined behavior when the proactor thread attempts to access the destroyed stop atomic or other local variables.
To resolve this, we can set stop = true and then call proactor()->AwaitBrief([] {});. Since the proactor's task queue is FIFO, this blocks the test thread until the last queued spammer task has executed, observed stop == true, and exited without re-posting itself.
stop = true;
proactor()->AwaitBrief([] {});
bg.Join();![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@util/fibers/fibers_test.cc`:
- Around line 1116-1143: After setting stop = true, ensure any reposted spammer
callbacks that captured stack locals have finished executing before exiting the
test: add a fence by invoking proactor()->Await([]{}) (or otherwise dispatch a
no-op and Await) after stop = true and before bg.Join()/return so the proactor
processes all previously DispatchBrief(spammer) callbacks and no queued
callbacks will dereference destroyed locals (references: spammer lambda,
proactor()->DispatchBrief(spammer), stop, proactor()->Await).
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro
Run ID: 51079460-6cfd-4fce-8d74-87ad6767de43
📒 Files selected for processing (1)
util/fibers/fibers_test.cc
| std::function<void()> spammer = [&] { | ||
| if (stop.load(std::memory_order_relaxed)) | ||
| return; | ||
| auto end = chrono::steady_clock::now() + 600us; | ||
| while (chrono::steady_clock::now() < end); | ||
| if (normal_done.exchange(false)) { | ||
| // Notify bg from task_queue context, AFTER the previous Run(NORMAL) incremented | ||
| // runtime_ns_[NORMAL] to 15ms and returned. On the next Run(NORMAL), fill(0) will zero | ||
| // runtime_ns_ before the warrant check, leaving runtime_ns_[NORMAL]=0 -> warrant fails. | ||
| bg_wake.Notify(); | ||
| } | ||
| proactor()->DispatchBrief(spammer); | ||
| }; | ||
| proactor()->DispatchBrief(spammer); | ||
|
|
||
| // NORMAL fiber: spin >kBaseSlice (10ms) to push runtime_ns_[NORMAL] past the fill(0) threshold. | ||
| // Does NOT notify bg — leaves that to the spammer so bg enters ready_queue[BACKGROUND] only | ||
| // after this fiber's runtime has been recorded and fill(0) is pending for the next Run(NORMAL). | ||
| proactor()->Await([&] { | ||
| auto end = chrono::steady_clock::now() + 15ms; | ||
| while (chrono::steady_clock::now() < end); | ||
| normal_done.store(true, std::memory_order_release); | ||
| }); | ||
|
|
||
| EXPECT_TRUE(bg_done.WaitFor(1s)); | ||
| stop = true; | ||
| bg.Join(); | ||
| } |
There was a problem hiding this comment.
Drain reposted spammer callbacks before leaving the test scope.
Line 1127 reposts spammer asynchronously while capturing stack locals by reference. After Line 1141 sets stop = true, the test can still return with queued callbacks that dereference destroyed locals.
Proposed fix
EXPECT_TRUE(bg_done.WaitFor(1s));
stop = true;
+ // Drain task-queue callbacks posted before/around stop=true so no callback
+ // touches stack locals after this test returns.
+ proactor()->AwaitBrief([] {});
bg.Join();📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| std::function<void()> spammer = [&] { | |
| if (stop.load(std::memory_order_relaxed)) | |
| return; | |
| auto end = chrono::steady_clock::now() + 600us; | |
| while (chrono::steady_clock::now() < end); | |
| if (normal_done.exchange(false)) { | |
| // Notify bg from task_queue context, AFTER the previous Run(NORMAL) incremented | |
| // runtime_ns_[NORMAL] to 15ms and returned. On the next Run(NORMAL), fill(0) will zero | |
| // runtime_ns_ before the warrant check, leaving runtime_ns_[NORMAL]=0 -> warrant fails. | |
| bg_wake.Notify(); | |
| } | |
| proactor()->DispatchBrief(spammer); | |
| }; | |
| proactor()->DispatchBrief(spammer); | |
| // NORMAL fiber: spin >kBaseSlice (10ms) to push runtime_ns_[NORMAL] past the fill(0) threshold. | |
| // Does NOT notify bg — leaves that to the spammer so bg enters ready_queue[BACKGROUND] only | |
| // after this fiber's runtime has been recorded and fill(0) is pending for the next Run(NORMAL). | |
| proactor()->Await([&] { | |
| auto end = chrono::steady_clock::now() + 15ms; | |
| while (chrono::steady_clock::now() < end); | |
| normal_done.store(true, std::memory_order_release); | |
| }); | |
| EXPECT_TRUE(bg_done.WaitFor(1s)); | |
| stop = true; | |
| bg.Join(); | |
| } | |
| std::function<void()> spammer = [&] { | |
| if (stop.load(std::memory_order_relaxed)) | |
| return; | |
| auto end = chrono::steady_clock::now() + 600us; | |
| while (chrono::steady_clock::now() < end); | |
| if (normal_done.exchange(false)) { | |
| // Notify bg from task_queue context, AFTER the previous Run(NORMAL) incremented | |
| // runtime_ns_[NORMAL] to 15ms and returned. On the next Run(NORMAL), fill(0) will zero | |
| // runtime_ns_ before the warrant check, leaving runtime_ns_[NORMAL]=0 -> warrant fails. | |
| bg_wake.Notify(); | |
| } | |
| proactor()->DispatchBrief(spammer); | |
| }; | |
| proactor()->DispatchBrief(spammer); | |
| // NORMAL fiber: spin >kBaseSlice (10ms) to push runtime_ns_[NORMAL] past the fill(0) threshold. | |
| // Does NOT notify bg — leaves that to the spammer so bg enters ready_queue[BACKGROUND] only | |
| // after this fiber's runtime has been recorded and fill(0) is pending for the next Run(NORMAL). | |
| proactor()->Await([&] { | |
| auto end = chrono::steady_clock::now() + 15ms; | |
| while (chrono::steady_clock::now() < end); | |
| normal_done.store(true, std::memory_order_release); | |
| }); | |
| EXPECT_TRUE(bg_done.WaitFor(1s)); | |
| stop = true; | |
| // Drain task-queue callbacks posted before/around stop=true so no callback | |
| // touches stack locals after this test returns. | |
| proactor()->AwaitBrief([] {}); | |
| bg.Join(); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@util/fibers/fibers_test.cc` around lines 1116 - 1143, After setting stop =
true, ensure any reposted spammer callbacks that captured stack locals have
finished executing before exiting the test: add a fence by invoking
proactor()->Await([]{}) (or otherwise dispatch a no-op and Await) after stop =
true and before bg.Join()/return so the proactor processes all previously
DispatchBrief(spammer) callbacks and no queued callbacks will dereference
destroyed locals (references: spammer lambda,
proactor()->DispatchBrief(spammer), stop, proactor()->Await).
|
@romange irrelevant to the heartbeat issue I am investigating I found this. iouring proactor run loop can starve background fibers when load on the task queue is high I don't see this being critical for now but it's worth fixing (: |
|
@dranikpg can comment on this as he designed this feature. |
2 participants
When a proactor's task queue is heavy, it sets
has_cpu_workto true. Under high and continuous load thebackgroundfiber queue isstarved. The root cause is that the proactor run loop skips the BACKGROUND tasks because has_cpu_work is true. This spins without a fallback for the entire load time leading to the starvation of the BACKGROUND task queue.CI being red is the proof the tests fails. I will follow up with a fix.