Add: report NEXT_LEVEL reservation stalls - #1613
Conversation
d688ef9 to
1a3a55c
Compare
📝 WalkthroughWalkthroughThe scheduler now uses wake generations and completion events for waiting. NEXT_LEVEL dispatch returns reservation details, tracks structural stalls, and emits timed diagnostics. Queue inspection and tests cover blocked groups, queued singles, duplicate completions, and wake-up behavior. ChangesScheduler wake and reservation stall handling
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Scheduler
participant ReadyQueues
participant CompletionFIFO
participant WorkerDiagnosticSink
Scheduler->>ReadyQueues: inspect target queue heads
ReadyQueues-->>Scheduler: reservation and queue state
Scheduler->>Scheduler: track wake generation and stall deadline
CompletionFIFO-->>Scheduler: terminal completion
Scheduler->>WorkerDiagnosticSink: report reservation stall
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
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 |
puddingfjz
changed the title
Consume each READY or stop notification once instead of treating a non-empty blocked queue as permanent work. Group-member completions that do not enqueue a terminal task completion also advance the wake generation. This keeps queued singles and stop() moving after a member was terminalized early. Tests cover blocked-head sleeping, partial group completion, and the terminalized-member idle transition that previously hung CI.
Warn once when a blocked group continuously reserves an idle target that has queued single work. The native no-throw sink reports the group and target queue structure without releasing the reservation or classifying the state as a deadlock.
1a3a55c to
3b3ebb1
Compare
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (4)
tests/ut/cpp/hierarchical/test_scheduler.cpp (2)
1179-1186: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueAssert the worker pointers before dereferencing them.
manager.get_worker_by_idreturnsnullptrfor an unregistered id.Scheduler::dispatch_next_level_groupinsrc/common/hierarchical/scheduler.cppchecks for that case explicitly. Worker ids 0 and 1 are registered inSetUp(), so the current test is safe. AddASSERT_NE(..., nullptr)so a future registration change fails as an assertion instead of a segmentation fault inside a lock scope.🛡️ Proposed change
WorkerThread *manager_worker_a = manager.get_worker_by_id(WorkerType::NEXT_LEVEL, 0); WorkerThread *manager_worker_b = manager.get_worker_by_id(WorkerType::NEXT_LEVEL, 1); + ASSERT_NE(manager_worker_a, nullptr); + ASSERT_NE(manager_worker_b, nullptr); while ((!manager_worker_a->idle() || !manager_worker_b->idle()) &&🤖 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 `@tests/ut/cpp/hierarchical/test_scheduler.cpp` around lines 1179 - 1186, Add ASSERT_NE checks for manager_worker_a and manager_worker_b immediately after their get_worker_by_id calls and before the idle() polling loop, so null worker registrations fail the test safely before dereferencing either pointer.
786-796: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winGate the diagnostic field checks with an assertion.
If the report does not arrive within 200 ms, line 790 fails and the test continues. Lines 791-796 then compare default-initialized capture fields and emit five more failures. Use
ASSERT_EQfor the report-count check so the test stops at the root cause.The same pattern applies to the exact
dispatched_count()checks at lines 797, 810-811, and 820-821. Those are less severe because a stale count fails only one expectation.♻️ Proposed change
- EXPECT_EQ(stall_capture.report_count.load(std::memory_order_acquire), 1); + ASSERT_EQ(stall_capture.report_count.load(std::memory_order_acquire), 1); EXPECT_EQ(stall_capture.group_slot, group.task_slot); - EXPECT_EQ(stall_capture.busy_target_count, 1u); + ASSERT_EQ(stall_capture.busy_target_count, 1u); EXPECT_EQ(stall_capture.busy_target_worker_ids[0], 1); - EXPECT_EQ(stall_capture.idle_queued_target_count, 1u); + ASSERT_EQ(stall_capture.idle_queued_target_count, 1u); EXPECT_EQ(stall_capture.idle_queued_target_worker_ids[0], 0); EXPECT_EQ(stall_capture.idle_queued_single_head_slots[0], single_a.task_slot);🤖 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 `@tests/ut/cpp/hierarchical/test_scheduler.cpp` around lines 786 - 796, Use ASSERT_EQ for the stall_capture.report_count check before validating the diagnostic fields, so the test returns immediately when no report arrives. Apply the same assertion-strengthening to the exact dispatched_count() checks at the indicated points, preserving their existing expected values and surrounding test logic.src/common/hierarchical/scheduler.h (2)
108-146: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDocument the single-writer-thread invariant on
reservation_stall_episode_.
reservation_stall_deadline()readsreservation_stall_episode_undercompletion_mu_(inrun()), andupdate_reservation_stall()writes it underloop_mu_(indispatch_ready()). This is safe only because both calls execute exclusively on the scheduler thread. Add a short comment near the member to record this invariant, so a future change that invokes either method from another thread does not introduce a data race.🤖 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 `@src/common/hierarchical/scheduler.h` around lines 108 - 146, Add a short comment immediately above reservation_stall_episode_ documenting that it is accessed only by the scheduler thread, including its reads in reservation_stall_deadline() and writes in update_reservation_stall(). Do not change the locking or surrounding dispatch logic.
59-69: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winDocument the pointer-lifetime contract on
ReservationStallDiagnostic.
busy_target_worker_ids,idle_queued_target_worker_ids, andidle_queued_single_head_slotsare raw pointers. Inscheduler.cpp,update_reservation_stallbuilds this struct fromdispatch_result.busy_target_worker_ids.data()and similar calls, wheredispatch_resultis a temporary owned by the caller ofdispatch_ready(). The sink call is synchronous, so the pointers stay valid only for the duration of that call.Add a comment on the struct that states this constraint. A future sink implementation that stores these pointers past the callback invocation reads freed memory.
📝 Proposed documentation addition
struct ReservationStallDiagnostic { + // All pointer/count fields below reference memory owned by the caller + // of the sink and are valid only for the duration of the sink call. + // Do not store these pointers past the callback invocation. TaskSlot group_slot{INVALID_SLOT}; const int32_t *busy_target_worker_ids{nullptr};🤖 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 `@src/common/hierarchical/scheduler.h` around lines 59 - 69, Add a documentation comment to the ReservationStallDiagnostic struct stating that its raw pointer fields are valid only during the synchronous ReservationStallSink callback and must not be retained afterward, since they reference caller-owned temporary storage.
🤖 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 `@src/common/hierarchical/worker.cpp`:
- Around line 37-54: Update report_reservation_stall so the synchronously
invoked diagnostic sink never performs potentially blocking stderr operations.
Replace flockfile and repeated std::fprintf calls with an existing bounded
non-blocking native sink, or add explicit EPIPE and backpressure handling that
prevents SIGPIPE and unbounded blocking before this sink is registered.
In `@tests/ut/cpp/hierarchical/test_scheduler.cpp`:
- Around line 1189-1208: Move the EXPECT_EQ assertions for worker_a and worker_b
dispatch counts until after the optional notify_ready retry completes. Keep the
initial wait and retry as best-effort cleanup protection, then assert the final
counts before conditionally calling complete() so a successful retry does not
leave a prior test failure.
---
Nitpick comments:
In `@src/common/hierarchical/scheduler.h`:
- Around line 108-146: Add a short comment immediately above
reservation_stall_episode_ documenting that it is accessed only by the scheduler
thread, including its reads in reservation_stall_deadline() and writes in
update_reservation_stall(). Do not change the locking or surrounding dispatch
logic.
- Around line 59-69: Add a documentation comment to the
ReservationStallDiagnostic struct stating that its raw pointer fields are valid
only during the synchronous ReservationStallSink callback and must not be
retained afterward, since they reference caller-owned temporary storage.
In `@tests/ut/cpp/hierarchical/test_scheduler.cpp`:
- Around line 1179-1186: Add ASSERT_NE checks for manager_worker_a and
manager_worker_b immediately after their get_worker_by_id calls and before the
idle() polling loop, so null worker registrations fail the test safely before
dereferencing either pointer.
- Around line 786-796: Use ASSERT_EQ for the stall_capture.report_count check
before validating the diagnostic fields, so the test returns immediately when no
report arrives. Apply the same assertion-strengthening to the exact
dispatched_count() checks at the indicated points, preserving their existing
expected values and surrounding test logic.
🪄 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: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: c6f392a1-58b3-4829-a64a-9d5720437d0d
📒 Files selected for processing (7)
docs/scheduler.mdsrc/common/hierarchical/scheduler.cppsrc/common/hierarchical/scheduler.hsrc/common/hierarchical/types.cppsrc/common/hierarchical/types.hsrc/common/hierarchical/worker.cpptests/ut/cpp/hierarchical/test_scheduler.cpp
| void report_reservation_stall(void *, const Scheduler::ReservationStallDiagnostic &diagnostic) noexcept { | ||
| flockfile(stderr); | ||
| std::fprintf( | ||
| stderr, "[WARN] NEXT_LEVEL group reservation stalled: group_slot=%d busy_target_ids=[", diagnostic.group_slot | ||
| ); | ||
| for (size_t i = 0; i < diagnostic.busy_target_count; ++i) { | ||
| std::fprintf(stderr, "%s%d", i == 0 ? "" : ",", diagnostic.busy_target_worker_ids[i]); | ||
| } | ||
| std::fputs("] idle_targets_with_queued_singles=[", stderr); | ||
| for (size_t i = 0; i < diagnostic.idle_queued_target_count; ++i) { | ||
| std::fprintf( | ||
| stderr, "%s%d:head_slot=%d", i == 0 ? "" : ",", diagnostic.idle_queued_target_worker_ids[i], | ||
| diagnostic.idle_queued_single_head_slots[i] | ||
| ); | ||
| } | ||
| std::fputs("]\n", stderr); | ||
| funlockfile(stderr); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Keep the registered diagnostic sink off the blocking scheduler path.
Scheduler::update_reservation_stall invokes this sink synchronously. flockfile and the repeated std::fprintf calls can block when stderr is slow or its pipe is full. A closed pipe can raise SIGPIPE and terminate the process. Route the warning through a bounded non-blocking native sink, or define explicit EPIPE and backpressure handling before registering this direct stderr writer.
Also applies to: 151-151
🤖 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 `@src/common/hierarchical/worker.cpp` around lines 37 - 54, Update
report_reservation_stall so the synchronously invoked diagnostic sink never
performs potentially blocking stderr operations. Replace flockfile and repeated
std::fprintf calls with an existing bounded non-blocking native sink, or add
explicit EPIPE and backpressure handling that prevents SIGPIPE and unbounded
blocking before this sink is registered.
| auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(500); | ||
| while ((worker_a.dispatched_count() < 2 || worker_b.dispatched_count() < 2) && | ||
| std::chrono::steady_clock::now() < deadline) { | ||
| std::this_thread::sleep_for(std::chrono::milliseconds(1)); | ||
| } | ||
| EXPECT_EQ(worker_a.dispatched_count(), 2); | ||
| EXPECT_EQ(worker_b.dispatched_count(), 2); | ||
|
|
||
| // Keep cleanup non-fatal so a missing wake reports a test failure instead | ||
| // of hanging the fixture in Scheduler::stop(). | ||
| if (worker_a.dispatched_count() < 2 || worker_b.dispatched_count() < 2) { | ||
| sched.notify_ready(); | ||
| deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(500); | ||
| while ((worker_a.dispatched_count() < 2 || worker_b.dispatched_count() < 2) && | ||
| std::chrono::steady_clock::now() < deadline) { | ||
| std::this_thread::sleep_for(std::chrono::milliseconds(1)); | ||
| } | ||
| } | ||
| if (worker_a.dispatched_count() >= 2) worker_a.complete(); | ||
| if (worker_b.dispatched_count() >= 2) worker_b.complete(); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
The retry block cannot recover the test outcome.
Lines 1194-1195 record a failure with EXPECT_EQ when a dispatch count is below 2. The retry at lines 1199-1206 then runs for exactly that case. If the retry succeeds, the test still fails because the earlier EXPECT_EQ already failed. The stated intent in the comment at lines 1197-1198 is not achieved.
Pick one behavior. Either treat the first wait as best-effort and assert only after the retry, or remove the retry and keep the direct expectation.
🐛 Proposed fix: assert after the retry
auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(500);
while ((worker_a.dispatched_count() < 2 || worker_b.dispatched_count() < 2) &&
std::chrono::steady_clock::now() < deadline) {
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
- EXPECT_EQ(worker_a.dispatched_count(), 2);
- EXPECT_EQ(worker_b.dispatched_count(), 2);
- // Keep cleanup non-fatal so a missing wake reports a test failure instead
- // of hanging the fixture in Scheduler::stop().
+ // Retry with an explicit notification so a missing completion-driven wake
+ // is reported as a failure below instead of hanging Scheduler::stop().
if (worker_a.dispatched_count() < 2 || worker_b.dispatched_count() < 2) {
sched.notify_ready();
deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(500);
while ((worker_a.dispatched_count() < 2 || worker_b.dispatched_count() < 2) &&
std::chrono::steady_clock::now() < deadline) {
std::this_thread::sleep_for(std::chrono::milliseconds(1));
}
+ ADD_FAILURE() << "queued singles required an explicit notify_ready(); "
+ "the completion callback did not wake the scheduler";
}
+ EXPECT_EQ(worker_a.dispatched_count(), 2);
+ EXPECT_EQ(worker_b.dispatched_count(), 2);
if (worker_a.dispatched_count() >= 2) worker_a.complete();
if (worker_b.dispatched_count() >= 2) worker_b.complete();📝 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.
| auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(500); | |
| while ((worker_a.dispatched_count() < 2 || worker_b.dispatched_count() < 2) && | |
| std::chrono::steady_clock::now() < deadline) { | |
| std::this_thread::sleep_for(std::chrono::milliseconds(1)); | |
| } | |
| EXPECT_EQ(worker_a.dispatched_count(), 2); | |
| EXPECT_EQ(worker_b.dispatched_count(), 2); | |
| // Keep cleanup non-fatal so a missing wake reports a test failure instead | |
| // of hanging the fixture in Scheduler::stop(). | |
| if (worker_a.dispatched_count() < 2 || worker_b.dispatched_count() < 2) { | |
| sched.notify_ready(); | |
| deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(500); | |
| while ((worker_a.dispatched_count() < 2 || worker_b.dispatched_count() < 2) && | |
| std::chrono::steady_clock::now() < deadline) { | |
| std::this_thread::sleep_for(std::chrono::milliseconds(1)); | |
| } | |
| } | |
| if (worker_a.dispatched_count() >= 2) worker_a.complete(); | |
| if (worker_b.dispatched_count() >= 2) worker_b.complete(); | |
| auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(500); | |
| while ((worker_a.dispatched_count() < 2 || worker_b.dispatched_count() < 2) && | |
| std::chrono::steady_clock::now() < deadline) { | |
| std::this_thread::sleep_for(std::chrono::milliseconds(1)); | |
| } | |
| // Retry with an explicit notification so a missing completion-driven wake | |
| // is reported as a failure below instead of hanging Scheduler::stop(). | |
| if (worker_a.dispatched_count() < 2 || worker_b.dispatched_count() < 2) { | |
| sched.notify_ready(); | |
| deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(500); | |
| while ((worker_a.dispatched_count() < 2 || worker_b.dispatched_count() < 2) && | |
| std::chrono::steady_clock::now() < deadline) { | |
| std::this_thread::sleep_for(std::chrono::milliseconds(1)); | |
| } | |
| ADD_FAILURE() << "queued singles required an explicit notify_ready(); " | |
| "the completion callback did not wake the scheduler"; | |
| } | |
| EXPECT_EQ(worker_a.dispatched_count(), 2); | |
| EXPECT_EQ(worker_b.dispatched_count(), 2); | |
| if (worker_a.dispatched_count() >= 2) worker_a.complete(); | |
| if (worker_b.dispatched_count() >= 2) worker_b.complete(); |
🤖 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 `@tests/ut/cpp/hierarchical/test_scheduler.cpp` around lines 1189 - 1208, Move
the EXPECT_EQ assertions for worker_a and worker_b dispatch counts until after
the optional notify_ready retry completes. Keep the initial wait and retry as
best-effort cleanup protection, then assert the final counts before
conditionally calling complete() so a successful retry does not leave a prior
test failure.
1 participant
Summary
NEXT_LEVELgroup reservation when an idle reserved target has queued single workBehavior
This is diagnostic only. It does not classify the condition as a deadlock, release the reservation, or otherwise change scheduling policy. Reporting is edge-triggered per group/stall episode and uses a native no-throw sink rather than a Python logger callback.
Dependency
Depends on #1612.
This PR is intentionally stacked on the wake-generation fix because the warning deadline uses the scheduler timed-wait path introduced there. Merge #1612 first; the branch will then be rebased onto the updated
mainbefore this PR merges.Review context
#1565 (comment)
Testing
test_scheduler: 28/28 passed