Add: report NEXT_LEVEL reservation stalls

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.

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<std::mutex> 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<int32_t> 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<RunId> &run_snapshot) {
     }
 }
 
-std::unordered_set<int32_t> Scheduler::dispatch_next_level_group(const std::optional<RunId> &run_snapshot) {
+Scheduler::NextLevelGroupDispatchResult Scheduler::dispatch_next_level_group(const std::optional<RunId> &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,24 +576,45 @@ std::unordered_set<int32_t> 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<WorkerThread *> workers;
         workers.reserve(static_cast<size_t>(group_size));
-        std::unordered_set<int32_t> target_worker_ids;
-        target_worker_ids.reserve(static_cast<size_t>(group_size));
+        result.reserved_worker_ids.reserve(static_cast<size_t>(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);
             WorkerThread *worker = cfg_.manager->get_worker_by_id(WorkerType::NEXT_LEVEL, worker_id);
             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<int32_t>(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<int32_t> 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<std::chrono::steady_clock::time_point> 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<int32_t> &reserved_worker_ids, const std::optional<RunId> &run_snapshot
 ) {

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<void(TaskSlot, const std::string &)> 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<bool> stop_requested_{false};
     std::atomic<bool> running_{false};
+    struct NextLevelGroupDispatchResult {
+        std::unordered_set<int32_t> reserved_worker_ids;
+        TaskSlot blocked_group_slot{INVALID_SLOT};
+        std::vector<int32_t> busy_target_worker_ids;
+        std::vector<int32_t> idle_queued_target_worker_ids;
+        std::vector<TaskSlot> 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<uint64_t> 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<ReservationStallEpisode> 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<int32_t> dispatch_next_level_group(const std::optional<RunId> &run_snapshot);
+    NextLevelGroupDispatchResult dispatch_next_level_group(const std::optional<RunId> &run_snapshot);
     void dispatch_next_level_singles(
         const std::unordered_set<int32_t> &reserved_worker_ids, const std::optional<RunId> &run_snapshot
     );
     void dispatch_sub_ready(const std::optional<RunId> &run_snapshot);
+    void update_reservation_stall(const NextLevelGroupDispatchResult &dispatch_result);
+    std::optional<std::chrono::steady_clock::time_point> reservation_stall_deadline() const;
 };

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);
 }

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);

src/common/hierarchical/worker.cpp

@@ -11,6 +11,10 @@
 
 #include "worker.h"
 
+#include <unistd.h>
+
+#include <cstddef>
+#include <cstdio>
 #include <cstdlib>
 #include <mutex>
 #include <stdexcept>
@@ -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 <typename... Args>
+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<size_t>(wanted) >= cap - len ? cap - 1 : len + static_cast<size_t>(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.