Refactor: split ready queue per WorkerType (Strict-4)

docs/roadmap.md

@@ -88,8 +88,11 @@ get if I pip install `main` today", this page.
   the L2 ABI `ChipStorageTaskArgs` POD from the view right before
   `pto2_run_runtime` — the slot itself stores only the tagged
   `TaskArgs` (single) or `task_args_list` (group).
-- `Scheduler` dispatches slot ids via a single ready queue into
-  `WorkerManager` pools (next-level + sub); for group slots it pushes
+- `Scheduler` dispatches slot ids via **per-worker-type ready queues**
+  (Strict-4; one `DistReadyQueue` for `NEXT_LEVEL`, one for `SUB`) into
+  `WorkerManager` pools (next-level + sub). `dispatch_ready` drains each
+  queue with its own head-of-line break, so a saturated pool of one
+  type cannot stall dispatch for the other. For group slots it pushes
   a `WorkerDispatch { slot, group_index }` per member onto N idle
   threads.
 - `DistChipProcess` / `DistSubWorker` are separate classes today;
@@ -110,11 +113,11 @@ get if I pip install `main` today", this page.
 
 ## In flight / not yet landed
 
-### PR-D: WorkerThread unification + per-shape ready queues
+### PR-D-2: WorkerThread unification (PROCESS mode)
 
 - Fold `DistChipProcess` / `DistSubWorker` into `WorkerThread` with
-  `Mode = THREAD | PROCESS`.
-- Strict-4: 3 ready queues (AIC / AIV / MIX) instead of a single queue.
+  `Mode = THREAD | PROCESS` (no separate fork-proxy classes). Strict-4
+  per-worker-type ready queues already landed in PR-D-1.
 
 ### PR-E: uniform `Worker.run` + callable registry unification
 

docs/scheduler.md

@@ -2,8 +2,11 @@
 
 > **Status**: target design. `IWorker::run(uint64_t callable, TaskArgsView,
 > ChipCallConfig)` is the live dispatch signature; per-worker-type ready
-> queue split (Strict-4) is not yet implemented (lands in PR-D). See
-> [roadmap.md](roadmap.md) for the full landed-vs-planned breakdown.
+> queue split (Strict-4) is landed — each `WorkerType` has its own
+> `DistReadyQueue` and `dispatch_ready` walks them independently, so a
+> saturated pool of one type cannot head-of-line-block dispatch for the
+> other. See [roadmap.md](roadmap.md) for the full landed-vs-planned
+> breakdown.
 
 The Scheduler is the **DAG executor**. A dedicated C++ thread that consumes
 submitted slots, wires fanout edges, dispatches ready tasks to worker threads,
@@ -37,16 +40,21 @@ consults scheduling metadata (`fanin_count`, `fanout_consumers`, `state`).
 
 ---
 
-## 2. The three queues
+## 2. The queues
 
 ```cpp
 class Scheduler {
     // Producer: Orchestrator.submit_*. Consumer: Scheduler's own loop, Phase 0.
     LockFreeQueue<WiringEntry> wiring_queue_;       // {slot, producers}
 
-    // Producer: Scheduler Phase 0 (newly-ready) + Phase 2 (fanout-released).
-    // Consumer: Scheduler's own loop, Phase 1.
-    LockFreeQueue<TaskSlot> ready_queue_;
+    // Strict-4 — per-worker-type ready queues.
+    // Producers: Orchestrator.submit_* (routes by slot.worker_type) +
+    //            Scheduler Phase 0 / Phase 2 (fanout-released; routes by
+    //            consumer worker_type).
+    // Consumer: Scheduler's own loop, Phase 1 (one drain loop per queue,
+    //           each with its own head-of-line break).
+    DistReadyQueue *ready_next_level_queue_;
+    DistReadyQueue *ready_sub_queue_;
 
     // Producer: WorkerThread (on worker->run() return).
     // Consumer: Scheduler's own loop, Phase 2.
@@ -72,7 +80,27 @@ struct WiringEntry {
 ### Ready queue
 
 Slots whose `fanin_count == fanin_released` are ready to dispatch. The queue
-holds just the slot id; dispatch reads task data from `slots_[sid]`.
+holds just the slot id; dispatch reads task data from the
+`ring.slot_state(sid)` pool.
+
+**Strict-4 — per-worker-type split.** In practice the ready queue is two
+`DistReadyQueue` instances, one per `WorkerType`:
+
+```cpp
+DistReadyQueue ready_next_level_queue_;   // WorkerType::NEXT_LEVEL tasks
+DistReadyQueue ready_sub_queue_;          // WorkerType::SUB tasks
+```
+
+Matching L2's per-shape ready buffer (`PTO2_LocalReadyBuffer` fan-out to
+AIC / AIV / MIX queues), with the L3+ exception that we use `std::queue`
+(Allowed Exception 3: dynamic data structures on host) and only two
+worker types (Allowed Exception 2: `NEXT_LEVEL` + `SUB` at L3+, not
+AIC / AIV / MIX). `Orchestrator::submit_*` routes each slot to the queue
+matching `slot.worker_type`; `Scheduler::on_task_complete` routes a
+newly-ready consumer the same way, based on the *consumer's* worker
+type. `Scheduler::dispatch_ready` drains each queue with its own
+head-of-line break so a saturated pool of one type cannot stall dispatch
+for the other.
 
 ### Completion queue
 
@@ -93,11 +121,8 @@ void Scheduler::run() {
             wire_fanout(w);   // see §4
         }
 
-        // Phase 1: dispatch
-        TaskSlot sid;
-        while (ready_queue_.try_pop(sid)) {
-            dispatch_ready(sid);   // see §5
-        }
+        // Phase 1: dispatch (drains BOTH per-type queues; see §5)
+        dispatch_ready();
 
         // Phase 2: completion
         while (completion_queue_.try_pop(sid)) {
@@ -144,7 +169,12 @@ void Scheduler::wire_fanout(const WiringEntry &w) {
     // Update consumer's fanin to the actual live count (producers already
     // finished don't count).
     c.fanin_count = actual_live;
-    if (actual_live == 0) ready_queue_.push(csid);
+    if (actual_live == 0) {
+        // Strict-4: wiring promotes directly to the per-type queue.
+        auto *q = (c.worker_type == WorkerType::NEXT_LEVEL) ? ready_next_level_queue_
+                                                            : ready_sub_queue_;
+        q->push(csid);
+    }
 }
 ```
 
@@ -160,35 +190,44 @@ The `lock_guard(p.fanout_mu)` + `p.state.load()` check ensures we either:
 
 ## 5. Phase 1 — dispatch
 
+`dispatch_ready` drains each per-type ready queue with its own
+head-of-line break so one saturated pool cannot stall the other:
+
 ```cpp
-void Scheduler::dispatch_ready(TaskSlot sid) {
-    TaskSlotState &s = slots_[sid];
-    s.state.store(TaskState::READY);
-
-    if (s.group_size == 0) {
-        // Single-worker task
-        WorkerThread *wt = manager_->pick_idle(s.worker_type);
-        wt->dispatch(sid);
-    } else {
-        // Group task — reserve N idle workers
-        auto wts = manager_->pick_n_idle(s.worker_type, s.group_size);
-        s.state.store(TaskState::RUNNING);
-        for (size_t i = 0; i < wts.size(); i++) {
-            wts[i]->dispatch(sid, /*group_index=*/static_cast<int32_t>(i));
+void Scheduler::dispatch_ready() {
+    auto drain_one = [&](DistReadyQueue *q) {
+        DistTaskSlot slot;
+        while (q->try_pop(slot)) {
+            TaskSlotState &s = slots_[slot];
+            int N = s.group_size();  // 1 for single-task slots
+
+            auto workers = manager_->pick_n_idle(s.worker_type, N);
+            if (static_cast<int>(workers.size()) < N) {
+                q->push(slot);   // put back; try again after a completion
+                break;
+            }
+            s.state.store(TaskState::RUNNING);
+            for (int i = 0; i < N; i++) {
+                workers[i]->dispatch({slot, i});
+            }
         }
-    }
+    };
+    drain_one(ready_next_level_queue_);
+    drain_one(ready_sub_queue_);
 }
 ```
 
-Dispatch hands off the slot id to a `WorkerThread`. The WorkerThread reads
-`slots_[sid].{callable, task_args, config}` on its own thread and executes —
-see [worker-manager.md](worker-manager.md) §3 for THREAD mode and §4 for
-PROCESS mode.
+Dispatch hands off a `WorkerDispatch {slot, group_index}` to a
+`WorkerThread`. The WorkerThread reads
+`ring.slot_state(slot).{callable, task_args, config}` on its own thread
+and executes — see [worker-manager.md](worker-manager.md) §3 for THREAD
+mode and §4 for PROCESS mode.
 
-**Pick-idle back-pressure**: if no idle worker exists in the pool,
-`pick_idle` blocks. The Scheduler thread is then stalled, which is fine —
-ready tasks pile up in the queue until a worker frees up. The ring's
-back-pressure at the Orch side already caps the number of in-flight tasks.
+**Pick-idle back-pressure**: when `pick_n_idle` returns fewer workers
+than the task needs, the slot is pushed back onto *its* queue and that
+queue's drain halts; the other-type queue's drain continues. The ring's
+back-pressure at the Orch side already caps the total number of
+in-flight tasks across both types.
 
 ---
 
@@ -217,7 +256,12 @@ void Scheduler::on_task_complete(TaskSlot sid) {
     for (TaskSlot csid : consumers) {
         TaskSlotState &c = slots_[csid];
         if (++c.fanin_released == c.fanin_count) {
-            ready_queue_.push(csid);       // consumer now ready
+            // Strict-4: push to the queue matching the *consumer's*
+            // worker type. A consumer of a NEXT_LEVEL producer can itself
+            // be SUB, so we pick based on `c.worker_type`, not `s`.
+            auto *q = (c.worker_type == WorkerType::NEXT_LEVEL) ? ready_next_level_queue_
+                                                                : ready_sub_queue_;
+            q->push(csid);
         }
     }
 

src/common/distributed/dist_orchestrator.cpp

@@ -16,12 +16,14 @@
 #include <stdexcept>
 
 void DistOrchestrator::init(
-    DistTensorMap *tensormap, DistRing *allocator, DistScope *scope, DistReadyQueue *ready_queue
+    DistTensorMap *tensormap, DistRing *allocator, DistScope *scope, DistReadyQueue *ready_next_level_queue,
+    DistReadyQueue *ready_sub_queue
 ) {
     tensormap_ = tensormap;
     allocator_ = allocator;
     scope_ = scope;
-    ready_queue_ = ready_queue;
+    ready_next_level_queue_ = ready_next_level_queue;
+    ready_sub_queue_ = ready_sub_queue;
     active_tasks_.store(0, std::memory_order_relaxed);
 }
 
@@ -218,9 +220,11 @@ DistSubmitResult DistOrchestrator::submit_impl(
     if (scope_ref > 0) scope_->register_task(slot);
 
     // --- Step 6: If no live fanins → READY ---
+    // Strict-4: push to the queue dedicated to this task's worker type so a
+    // saturated sub pool cannot stall next-level dispatch (and vice versa).
     if (live_fanins == 0) {
         s.state.store(TaskState::READY, std::memory_order_release);
-        ready_queue_->push(slot);
+        ready_queue_for(worker_type)->push(slot);
     } else {
         s.state.store(TaskState::PENDING, std::memory_order_release);
     }

src/common/distributed/dist_orchestrator.h

@@ -64,7 +64,14 @@ struct DistSubmitResult {
 
 class DistOrchestrator {
 public:
-    void init(DistTensorMap *tensormap, DistRing *allocator, DistScope *scope, DistReadyQueue *ready_queue);
+    // Strict-4: the engine keeps one DistReadyQueue per WorkerType so a
+    // saturated sub pool cannot head-of-line-block chip dispatch (and vice
+    // versa). Submit routes to the queue matching the task's worker_type;
+    // the Scheduler's dispatch_ready walks each queue independently.
+    void init(
+        DistTensorMap *tensormap, DistRing *allocator, DistScope *scope, DistReadyQueue *ready_next_level_queue,
+        DistReadyQueue *ready_sub_queue
+    );
 
     // Allocate an intermediate buffer from the Worker's HeapRing (MAP_SHARED,
     // visible to forked child workers). Returns a ContinuousTensor whose
@@ -122,7 +129,18 @@ class DistOrchestrator {
     DistTensorMap *tensormap_ = nullptr;
     DistRing *allocator_ = nullptr;
     DistScope *scope_ = nullptr;
-    DistReadyQueue *ready_queue_ = nullptr;
+    // Strict-4 per-worker-type ready queues. Each queue handles tasks of
+    // exactly one WorkerType so the Scheduler can dispatch from an idle pool
+    // without being blocked by another pool's saturation.
+    DistReadyQueue *ready_next_level_queue_ = nullptr;
+    DistReadyQueue *ready_sub_queue_ = nullptr;
+
+    // Returns the ready queue that owns tasks of the given worker type.
+    // The method itself does not mutate the Orchestrator (hence `const`);
+    // the returned pointer is non-const because callers push into the queue.
+    DistReadyQueue *ready_queue_for(WorkerType t) const {
+        return t == WorkerType::NEXT_LEVEL ? ready_next_level_queue_ : ready_sub_queue_;
+    }
 
     // --- Drain support (owned here, not on Worker) ---
     std::atomic<int32_t> active_tasks_{0};

src/common/distributed/dist_scheduler.cpp

@@ -26,7 +26,8 @@
 // =============================================================================
 
 void DistScheduler::start(const Config &cfg) {
-    if (cfg.ring == nullptr || cfg.ready_queue == nullptr || cfg.manager == nullptr)
+    if (cfg.ring == nullptr || cfg.ready_next_level_queue == nullptr || cfg.ready_sub_queue == nullptr ||
+        cfg.manager == nullptr)
         throw std::invalid_argument("DistScheduler::start: null config fields");
     cfg_ = cfg;
 
@@ -38,7 +39,9 @@ void DistScheduler::start(const Config &cfg) {
 void DistScheduler::stop() {
     stop_requested_.store(true, std::memory_order_release);
     completion_cv_.notify_all();
-    cfg_.ready_queue->shutdown();
+    // Shut down both per-type ready queues so any wait_pop waiters unblock.
+    cfg_.ready_next_level_queue->shutdown();
+    cfg_.ready_sub_queue->shutdown();
 
     if (sched_thread_.joinable()) sched_thread_.join();
 
@@ -135,7 +138,11 @@ void DistScheduler::on_task_complete(DistTaskSlot slot) {
         if (released >= cs.fanin_count) {
             TaskState expected = TaskState::PENDING;
             if (cs.state.compare_exchange_strong(expected, TaskState::READY, std::memory_order_acq_rel)) {
-                cfg_.ready_queue->push(consumer);
+                // Strict-4: route the freshly-ready consumer to the queue
+                // matching its own worker type.
+                auto *q =
+                    (cs.worker_type == WorkerType::NEXT_LEVEL) ? cfg_.ready_next_level_queue : cfg_.ready_sub_queue;
+                q->push(consumer);
                 completion_cv_.notify_one();
             }
         }
@@ -174,23 +181,31 @@ void DistScheduler::try_consume(DistTaskSlot slot) {
 // =============================================================================
 
 void DistScheduler::dispatch_ready() {
-    DistTaskSlot slot;
-    while (cfg_.ready_queue->try_pop(slot)) {
-        DistTaskSlotState &s = *cfg_.ring->slot_state(slot);
-        int N = s.group_size();  // 1 for normal tasks
-
-        auto workers = cfg_.manager->pick_n_idle(s.worker_type, N);
-        if (static_cast<int>(workers.size()) < N) {
-            cfg_.ready_queue->push(slot);
-            break;
-        }
+    // Strict-4: drain each per-type queue with its OWN head-of-line break.
+    // A saturated pool of one type only stalls its own queue; the other
+    // type continues to dispatch from its pool of idle workers.
+    auto drain_one = [this](DistReadyQueue *q) {
+        DistTaskSlot slot;
+        while (q->try_pop(slot)) {
+            DistTaskSlotState &s = *cfg_.ring->slot_state(slot);
+            int N = s.group_size();  // 1 for normal tasks
+
+            auto workers = cfg_.manager->pick_n_idle(s.worker_type, N);
+            if (static_cast<int>(workers.size()) < N) {
+                q->push(slot);
+                break;
+            }
 
-        s.state.store(TaskState::RUNNING, std::memory_order_release);
-        for (int i = 0; i < N; i++) {
-            WorkerDispatch d;
-            d.task_slot = slot;
-            d.group_index = i;
-            workers[i]->dispatch(d);
+            s.state.store(TaskState::RUNNING, std::memory_order_release);
+            for (int i = 0; i < N; i++) {
+                WorkerDispatch d;
+                d.task_slot = slot;
+                d.group_index = i;
+                workers[i]->dispatch(d);
+            }
         }
-    }
+    };
+
+    drain_one(cfg_.ready_next_level_queue);
+    drain_one(cfg_.ready_sub_queue);
 }

src/common/distributed/dist_scheduler.h

@@ -53,7 +53,11 @@ class DistScheduler {
 public:
     struct Config {
         DistRing *ring;  // owns slot state storage; Scheduler reads via ring->slot_state(id)
-        DistReadyQueue *ready_queue;
+        // Strict-4 per-worker-type ready queues. `dispatch_ready` walks each
+        // queue independently so a saturated pool of one worker type cannot
+        // head-of-line-block dispatch for the other.
+        DistReadyQueue *ready_next_level_queue;
+        DistReadyQueue *ready_sub_queue;
         DistWorkerManager *manager;  // not owned — Scheduler calls manager for dispatch
         // Called when a task reaches CONSUMED (TensorMap cleanup + ring release).
         std::function<void(DistTaskSlot)> on_consumed_cb;

src/common/distributed/dist_worker.cpp

@@ -128,7 +128,7 @@ void DistWorker::add_worker(WorkerType type, IWorker *worker) {
 void DistWorker::init() {
     if (initialized_) throw std::runtime_error("DistWorker: already initialized");
 
-    orchestrator_.init(&tensormap_, &allocator_, &scope_, &ready_queue_);
+    orchestrator_.init(&tensormap_, &allocator_, &scope_, &ready_next_level_queue_, &ready_sub_queue_);
 
     // Start WorkerManager first — creates WorkerThreads.
     // The on_complete callback routes through the Scheduler's worker_done().
@@ -138,7 +138,8 @@ void DistWorker::init() {
 
     DistScheduler::Config cfg;
     cfg.ring = &allocator_;
-    cfg.ready_queue = &ready_queue_;
+    cfg.ready_next_level_queue = &ready_next_level_queue_;
+    cfg.ready_sub_queue = &ready_sub_queue_;
     cfg.manager = &manager_;
     cfg.on_consumed_cb = [this](DistTaskSlot slot) {
         orchestrator_.on_consumed(slot);