diff --git a/docs/scheduler.md b/docs/scheduler.md index cace96495..ab2b31485 100644 --- a/docs/scheduler.md +++ b/docs/scheduler.md @@ -145,6 +145,14 @@ continue normally, and later groups do not reserve workers because the Scheduler does not scan past the FIFO head. The reservation is released when the group launches. +A blocked head is structurally stalled when one of its reserved targets is +idle and that target's single FIFO is non-empty. If that state persists for +five seconds, the Scheduler emits one native warning for the episode with the +group slot, busy target IDs, idle-but-queued target IDs, and their single FIFO +head slots. A head change or disappearance of the structural condition starts +a new episode. The warning does not classify the state as a deadlock and does +not release the reservation. + ## 4. SUB dispatch SUB has no public worker-ID selection. All READY SUB tasks share one FIFO. diff --git a/src/common/hierarchical/scheduler.cpp b/src/common/hierarchical/scheduler.cpp index a0b3bfa21..78d6c6284 100644 --- a/src/common/hierarchical/scheduler.cpp +++ b/src/common/hierarchical/scheduler.cpp @@ -101,6 +101,9 @@ void Scheduler::start(const Config &cfg) { if (cfg.ring == nullptr || cfg.ready_sub_queue == nullptr || cfg.ready_next_level_queues == nullptr || cfg.manager == nullptr || !cfg.enqueue_ready_cb) throw std::invalid_argument("Scheduler::start: null config fields"); + if (cfg.reservation_stall_warn_after < std::chrono::milliseconds::zero()) { + throw std::invalid_argument("Scheduler::start: negative reservation stall warning interval"); + } cfg_ = cfg; { @@ -110,6 +113,7 @@ void Scheduler::start(const Config &cfg) { ++wake_generation_; } dispatch_round_count_.store(0, std::memory_order_relaxed); + reservation_stall_episode_.reset(); stop_requested_.store(false, std::memory_order_relaxed); running_.store(true, std::memory_order_release); sched_thread_ = std::thread(&Scheduler::run, this); @@ -251,9 +255,15 @@ void Scheduler::run() { while (true) { { std::unique_lock lk(completion_mu_); - completion_cv_.wait(lk, [this, &observed_wake_generation] { + auto ready = [this, &observed_wake_generation] { return !completion_queue_.empty() || wake_generation_ != observed_wake_generation; - }); + }; + const auto stall_deadline = reservation_stall_deadline(); + if (stall_deadline.has_value()) { + completion_cv_.wait_until(lk, *stall_deadline, ready); + } else { + completion_cv_.wait(lk, ready); + } observed_wake_generation = wake_generation_; } @@ -424,8 +434,9 @@ void Scheduler::dispatch_ready() { // Group reservations and every queue pop in one pass belong to the same // whole-run FIFO head, even if a completion advances the head mid-pass. - const std::unordered_set reserved_worker_ids = dispatch_next_level_group(run_snapshot); - dispatch_next_level_singles(reserved_worker_ids, run_snapshot); + const NextLevelGroupDispatchResult group_result = dispatch_next_level_group(run_snapshot); + update_reservation_stall(group_result); + dispatch_next_level_singles(group_result.reserved_worker_ids, run_snapshot); dispatch_sub_ready(run_snapshot); } @@ -540,7 +551,7 @@ void Scheduler::dispatch_sub_ready(const std::optional &run_snapshot) { } } -std::unordered_set Scheduler::dispatch_next_level_group(const std::optional &run_snapshot) { +Scheduler::NextLevelGroupDispatchResult Scheduler::dispatch_next_level_group(const std::optional &run_snapshot) { TaskSlot slot; while (run_snapshot ? cfg_.ready_next_level_queues->try_front_group(*run_snapshot, slot) : cfg_.ready_next_level_queues->try_front_group(slot)) { @@ -565,10 +576,11 @@ std::unordered_set Scheduler::dispatch_next_level_group(const std::opti } const int32_t group_size = s.group_size(); + NextLevelGroupDispatchResult result; + result.blocked_group_slot = slot; std::vector workers; workers.reserve(static_cast(group_size)); - std::unordered_set target_worker_ids; - target_worker_ids.reserve(static_cast(group_size)); + result.reserved_worker_ids.reserve(static_cast(group_size)); bool all_workers_idle = true; for (int32_t i = 0; i < group_size; ++i) { const int32_t worker_id = s.target_worker_id(i); @@ -576,13 +588,33 @@ std::unordered_set Scheduler::dispatch_next_level_group(const std::opti if (worker == nullptr) { throw std::runtime_error("Scheduler::dispatch_next_level_group: invalid target worker"); } - if (!target_worker_ids.insert(worker_id).second) { + if (!result.reserved_worker_ids.insert(worker_id).second) { throw std::runtime_error("Scheduler::dispatch_next_level_group: duplicate target worker"); } - if (!worker->idle()) all_workers_idle = false; + const bool worker_idle = worker->idle(); + if (!worker_idle) { + all_workers_idle = false; + result.busy_target_worker_ids.push_back(worker_id); + } workers.push_back(worker); } - if (!all_workers_idle) return target_worker_ids; + if (!all_workers_idle) { + for (size_t i = 0; i < workers.size(); ++i) { + TaskSlot single_head = INVALID_SLOT; + const int32_t worker_id = s.target_worker_id(static_cast(i)); + if (std::find(result.busy_target_worker_ids.begin(), result.busy_target_worker_ids.end(), worker_id) != + result.busy_target_worker_ids.end()) + continue; + const bool has_queued_single = + run_snapshot ? + cfg_.ready_next_level_queues->try_front_single(worker_id, *run_snapshot, single_head) : + cfg_.ready_next_level_queues->try_front_single(worker_id, single_head); + if (!has_queued_single) continue; + result.idle_queued_target_worker_ids.push_back(worker_id); + result.idle_queued_single_head_slots.push_back(single_head); + } + return result; + } // The head was observed before the worker checks above, so a run // cancelling in that window can consume the slot and erase its whole @@ -615,6 +647,40 @@ std::unordered_set Scheduler::dispatch_next_level_group(const std::opti return {}; } +void Scheduler::update_reservation_stall(const NextLevelGroupDispatchResult &dispatch_result) { + if (dispatch_result.blocked_group_slot == INVALID_SLOT || dispatch_result.idle_queued_target_worker_ids.empty()) { + reservation_stall_episode_.reset(); + return; + } + + const auto now = std::chrono::steady_clock::now(); + if (!reservation_stall_episode_.has_value() || + reservation_stall_episode_->group_slot != dispatch_result.blocked_group_slot) { + reservation_stall_episode_ = ReservationStallEpisode{dispatch_result.blocked_group_slot, now, false}; + } + + ReservationStallEpisode &episode = *reservation_stall_episode_; + if (episode.reported || now < episode.started_at + cfg_.reservation_stall_warn_after) return; + + if (cfg_.reservation_stall_sink != nullptr) { + const ReservationStallDiagnostic diagnostic{ + dispatch_result.blocked_group_slot, + dispatch_result.busy_target_worker_ids.data(), + dispatch_result.busy_target_worker_ids.size(), + dispatch_result.idle_queued_target_worker_ids.data(), + dispatch_result.idle_queued_single_head_slots.data(), + dispatch_result.idle_queued_target_worker_ids.size(), + }; + cfg_.reservation_stall_sink(cfg_.reservation_stall_sink_context, diagnostic); + } + episode.reported = true; +} + +std::optional Scheduler::reservation_stall_deadline() const { + if (!reservation_stall_episode_.has_value() || reservation_stall_episode_->reported) return std::nullopt; + return reservation_stall_episode_->started_at + cfg_.reservation_stall_warn_after; +} + void Scheduler::dispatch_next_level_singles( const std::unordered_set &reserved_worker_ids, const std::optional &run_snapshot ) { diff --git a/src/common/hierarchical/scheduler.h b/src/common/hierarchical/scheduler.h index 762b64447..c55174121 100644 --- a/src/common/hierarchical/scheduler.h +++ b/src/common/hierarchical/scheduler.h @@ -80,6 +80,17 @@ bool claim_for_dispatch(TaskSlotState &s); class Scheduler { public: + struct ReservationStallDiagnostic { + TaskSlot group_slot{INVALID_SLOT}; + const int32_t *busy_target_worker_ids{nullptr}; + size_t busy_target_count{0}; + const int32_t *idle_queued_target_worker_ids{nullptr}; + const TaskSlot *idle_queued_single_head_slots{nullptr}; + size_t idle_queued_target_count{0}; + }; + + using ReservationStallSink = void (*)(void *, const ReservationStallDiagnostic &) noexcept; + struct Config { Ring *ring; // owns slot state storage; Scheduler reads via ring->slot_state(id) ReadyQueue *ready_sub_queue; @@ -96,6 +107,11 @@ class Scheduler { // Called as soon as an endpoint reports failure so the error is // attached to the task's run even when a group has other members live. std::function on_task_failed_cb; + // Diagnostic-only reservation stall reporting. The sink must not + // block: it runs on the scheduler dispatch path. + std::chrono::milliseconds reservation_stall_warn_after{std::chrono::seconds(5)}; + ReservationStallSink reservation_stall_sink{nullptr}; + void *reservation_stall_sink_context{nullptr}; // Test seam. Invoked immediately before the dispatch claim, which is // the one instant a cancelling run can still take a slot away. The // window is unreachable from outside — every other observable point is @@ -141,18 +157,41 @@ class Scheduler { std::thread sched_thread_; std::atomic stop_requested_{false}; std::atomic running_{false}; + struct NextLevelGroupDispatchResult { + std::unordered_set reserved_worker_ids; + TaskSlot blocked_group_slot{INVALID_SLOT}; + std::vector busy_target_worker_ids; + std::vector idle_queued_target_worker_ids; + std::vector idle_queued_single_head_slots; + }; + + struct ReservationStallEpisode { + TaskSlot group_slot{INVALID_SLOT}; + std::chrono::steady_clock::time_point started_at; + bool reported{false}; + }; + std::atomic dispatch_round_count_{0}; + // sched_thread_ owns this: update_reservation_stall() writes it under + // loop_mu_ and reservation_stall_deadline() reads it under completion_mu_, + // which is only race-free because both run on that one thread. start() + // resets it before the thread exists. Any reader added off sched_thread_ + // needs a lock the two paths do not currently share. + std::optional reservation_stall_episode_; void run(); void on_task_complete(const WorkerCompletion &completion); void poison_task(TaskSlot slot, const std::string &root_message); + void try_consume(TaskSlot slot); void dispatch_ready(); void dispatch_claimed(WorkerThread *worker, WorkerDispatch dispatch, bool prepared); void dispatch_preparable_next_level_singles(); - std::unordered_set dispatch_next_level_group(const std::optional &run_snapshot); + NextLevelGroupDispatchResult dispatch_next_level_group(const std::optional &run_snapshot); void dispatch_next_level_singles( const std::unordered_set &reserved_worker_ids, const std::optional &run_snapshot ); void dispatch_sub_ready(const std::optional &run_snapshot); + void update_reservation_stall(const NextLevelGroupDispatchResult &dispatch_result); + std::optional reservation_stall_deadline() const; }; diff --git a/src/common/hierarchical/types.cpp b/src/common/hierarchical/types.cpp index cefc719c9..13d038cbe 100644 --- a/src/common/hierarchical/types.cpp +++ b/src/common/hierarchical/types.cpp @@ -260,6 +260,14 @@ bool NextLevelReadyQueues::try_pop_single(int32_t worker_id, TaskSlot &out) { return queues_[index_for(worker_id)]->try_pop(out); } +bool NextLevelReadyQueues::try_front_single(int32_t worker_id, TaskSlot &out) { + return queues_[index_for(worker_id)]->try_front(out); +} + +bool NextLevelReadyQueues::try_front_single(int32_t worker_id, RunId run_id, TaskSlot &out) { + return queues_[index_for(worker_id)]->try_front(run_id, out); +} + bool NextLevelReadyQueues::try_pop_single(int32_t worker_id, RunId run_id, TaskSlot &out) { return queues_[index_for(worker_id)]->try_pop(run_id, out); } diff --git a/src/common/hierarchical/types.h b/src/common/hierarchical/types.h index 439c81efb..aed5cefd2 100644 --- a/src/common/hierarchical/types.h +++ b/src/common/hierarchical/types.h @@ -537,6 +537,8 @@ class NextLevelReadyQueues { void push_single(int32_t worker_id, RunId run_id, TaskSlot slot); bool try_pop_single(int32_t worker_id, TaskSlot &out); bool try_pop_single(int32_t worker_id, RunId run_id, TaskSlot &out); + bool try_front_single(int32_t worker_id, TaskSlot &out); + bool try_front_single(int32_t worker_id, RunId run_id, TaskSlot &out); void push_group(TaskSlot slot); void push_group(RunId run_id, TaskSlot slot); bool try_front_group(TaskSlot &out); diff --git a/src/common/hierarchical/worker.cpp b/src/common/hierarchical/worker.cpp index 7d0c529f6..7dec778de 100644 --- a/src/common/hierarchical/worker.cpp +++ b/src/common/hierarchical/worker.cpp @@ -11,6 +11,10 @@ #include "worker.h" +#include + +#include +#include #include #include #include @@ -33,6 +37,49 @@ namespace { std::once_flag g_fork_hygiene_once; +// Appends into a NUL-terminated buffer, truncating rather than overflowing. +// snprintf reports the length it wanted, not the length it wrote, so a +// would-be-longer result saturates `len` at the last writable index. +template +size_t append_truncating(char *buf, size_t cap, size_t len, const char *fmt, Args... args) { + if (len + 1 >= cap) return cap - 1; + const int wanted = std::snprintf(buf + len, cap - len, fmt, args...); + if (wanted < 0) return len; + return static_cast(wanted) >= cap - len ? cap - 1 : len + static_cast(wanted); +} + +void report_reservation_stall(void *, const Scheduler::ReservationStallDiagnostic &diagnostic) noexcept { + // Formatted into automatic storage and emitted with one write(2): the sink + // is noexcept and runs on the scheduler dispatch path, so it allocates + // nothing (a throwing allocation here would terminate the process), takes + // no stdio lock a forked Worker child could inherit held, and leaves + // nothing running for process exit to race. A message longer than the + // buffer loses its tail, which for a diagnostic beats any of those. + char message[512]; + size_t len = append_truncating( + message, sizeof(message), 0, "[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) { + len = append_truncating( + message, sizeof(message), len, "%s%d", i == 0 ? "" : ",", diagnostic.busy_target_worker_ids[i] + ); + } + len = append_truncating(message, sizeof(message), len, "] idle_targets_with_queued_singles=["); + for (size_t i = 0; i < diagnostic.idle_queued_target_count; ++i) { + len = append_truncating( + message, sizeof(message), len, "%s%d:head_slot=%d", i == 0 ? "" : ",", + diagnostic.idle_queued_target_worker_ids[i], diagnostic.idle_queued_single_head_slots[i] + ); + } + len = append_truncating(message, sizeof(message), len, "]\n"); + // A truncated tail still has to end the line, or this diagnostic runs into + // whatever writes to stderr next. + if (len > 0 && message[len - 1] != '\n') message[len - 1] = '\n'; + ssize_t written = ::write(STDERR_FILENO, message, len); + (void)written; +} + void apply_env_defaults_once() { // setenv with overwrite=0 leaves user-supplied values intact. setenv("OMP_NUM_THREADS", "1", 0); @@ -138,6 +185,7 @@ void Worker::init() { cfg.on_task_failed_cb = [this](TaskSlot slot, const std::string &message) { orchestrator_.report_task_error(slot, message); }; + cfg.reservation_stall_sink = report_reservation_stall; scheduler_.start(cfg); // Allocator compaction and scheduler slot access share this mutex. diff --git a/tests/ut/cpp/hierarchical/test_scheduler.cpp b/tests/ut/cpp/hierarchical/test_scheduler.cpp index 38ba13866..044641ed1 100644 --- a/tests/ut/cpp/hierarchical/test_scheduler.cpp +++ b/tests/ut/cpp/hierarchical/test_scheduler.cpp @@ -2166,6 +2166,9 @@ struct GroupSchedulerFixture : public ::testing::Test { Scheduler sched; CallConfig cfg; RunId run_id{INVALID_RUN_ID}; + std::chrono::milliseconds reservation_stall_warn_after{std::chrono::seconds(5)}; + Scheduler::ReservationStallSink reservation_stall_sink{nullptr}; + void *reservation_stall_sink_context{nullptr}; std::vector consumed_slots; std::mutex consumed_mu; @@ -2225,6 +2228,9 @@ struct GroupSchedulerFixture : public ::testing::Test { c.on_task_failed_cb = [this](TaskSlot s, const std::string &message) { orch.report_task_error(s, message); }; + c.reservation_stall_warn_after = reservation_stall_warn_after; + c.reservation_stall_sink = reservation_stall_sink; + c.reservation_stall_sink_context = reservation_stall_sink_context; sched.start(c); } @@ -2251,6 +2257,116 @@ struct GroupSchedulerFixture : public ::testing::Test { } }; +struct ReservationStallCapture { + std::atomic report_count{0}; + TaskSlot group_slot{INVALID_SLOT}; + std::array busy_target_worker_ids{}; + size_t busy_target_count{0}; + std::array idle_queued_target_worker_ids{}; + std::array idle_queued_single_head_slots{}; + size_t idle_queued_target_count{0}; +}; + +void capture_reservation_stall(void *context, const Scheduler::ReservationStallDiagnostic &diagnostic) noexcept { + auto *capture = static_cast(context); + capture->group_slot = diagnostic.group_slot; + capture->busy_target_count = std::min(diagnostic.busy_target_count, capture->busy_target_worker_ids.size()); + if (capture->busy_target_count > 0) { + std::copy_n( + diagnostic.busy_target_worker_ids, capture->busy_target_count, capture->busy_target_worker_ids.begin() + ); + } + capture->idle_queued_target_count = + std::min(diagnostic.idle_queued_target_count, capture->idle_queued_target_worker_ids.size()); + if (capture->idle_queued_target_count > 0) { + std::copy_n( + diagnostic.idle_queued_target_worker_ids, capture->idle_queued_target_count, + capture->idle_queued_target_worker_ids.begin() + ); + std::copy_n( + diagnostic.idle_queued_single_head_slots, capture->idle_queued_target_count, + capture->idle_queued_single_head_slots.begin() + ); + } + capture->report_count.fetch_add(1, std::memory_order_release); +} + +struct ReservationStallSchedulerFixture : public GroupSchedulerFixture { + ReservationStallCapture stall_capture; + + ReservationStallSchedulerFixture() { + reservation_stall_warn_after = std::chrono::milliseconds(20); + reservation_stall_sink = capture_reservation_stall; + reservation_stall_sink_context = &stall_capture; + } +}; + +TEST_F(ReservationStallSchedulerFixture, ReportsStructuralStallOncePerEpisode) { + auto running_a = orch.submit_next_level(C(88), single_tensor_args(0x110, TensorArgType::OUTPUT), cfg, 0); + auto running_b = orch.submit_next_level(C(89), single_tensor_args(0x111, TensorArgType::OUTPUT), cfg, 1); + worker_a.wait_running(); + worker_b.wait_running(); + EXPECT_TRUE(worker_a.is_running.load(std::memory_order_acquire)); + EXPECT_TRUE(worker_b.is_running.load(std::memory_order_acquire)); + + auto group = orch.submit_next_level_group( + C(90), {single_tensor_args(0x112, TensorArgType::OUTPUT), single_tensor_args(0x113, TensorArgType::OUTPUT)}, + cfg, {0, 1} + ); + auto single_a = orch.submit_next_level(C(91), single_tensor_args(0x114, TensorArgType::OUTPUT), cfg, 0); + auto single_b = orch.submit_next_level(C(92), single_tensor_args(0x115, TensorArgType::OUTPUT), cfg, 1); + + std::this_thread::sleep_for(std::chrono::milliseconds(40)); + EXPECT_EQ(stall_capture.report_count.load(std::memory_order_acquire), 0); + + worker_a.complete(); + auto deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(200); + while (stall_capture.report_count.load(std::memory_order_acquire) == 0 && + std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + EXPECT_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); + EXPECT_EQ(stall_capture.busy_target_worker_ids[0], 1); + EXPECT_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); + EXPECT_EQ(worker_a.dispatched_count(), 1); + + for (int i = 0; i < 3; ++i) + sched.notify_ready(); + std::this_thread::sleep_for(std::chrono::milliseconds(30)); + EXPECT_EQ(stall_capture.report_count.load(std::memory_order_acquire), 1); + + worker_b.complete(); + 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); + worker_a.complete(); + worker_b.complete(); + + deadline = std::chrono::steady_clock::now() + std::chrono::milliseconds(500); + while ((worker_a.dispatched_count() < 3 || worker_b.dispatched_count() < 3) && + std::chrono::steady_clock::now() < deadline) { + std::this_thread::sleep_for(std::chrono::milliseconds(1)); + } + EXPECT_EQ(worker_a.dispatched_count(), 3); + EXPECT_EQ(worker_b.dispatched_count(), 3); + worker_a.complete(); + worker_b.complete(); + + wait_consumed(running_a.task_slot); + wait_consumed(running_b.task_slot); + wait_consumed(group.task_slot); + wait_consumed(single_a.task_slot); + wait_consumed(single_b.task_slot); +} + TEST_F(GroupSchedulerFixture, GroupDispatchesToNWorkers) { TaskArgs a0 = single_tensor_args(0xA0, TensorArgType::OUTPUT); TaskArgs a1 = single_tensor_args(0xA1, TensorArgType::OUTPUT); @@ -2853,8 +2969,9 @@ TEST_F(GroupSchedulerFixture, InvalidGroupIndexFailsAndConsumesGroup) { 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(). + // The EXPECTs above already recorded the failure; this retry exists only + // to unblock cleanup so a missing wake fails the test 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);