optimize: prewire task dependencies on orchestrator side

simpler_setup/tools/swimlane_converter.py

@@ -1401,7 +1401,7 @@ def generate_chrome_trace_json(  # noqa: PLR0912, PLR0913, PLR0915
             "complete": "good",  # green
             "dispatch": "terrible",  # red
             "release": "olive",  # deferred-release drain (on_task_release work)
-            "wire": "thread_state_running",  # drain_wiring_queue pass
+            "wire": "thread_state_running",  # legacy scheduler wire phase
             "dummy": "grey",  # dummy_drain pass (Resolve nests inside)
             "early_dispatch": "rail_animation",  # speculative early-dispatch staging
             # Inner phase — nests inside Complete or Dummy via time containment

src/a2a3/platform/include/common/l2_swimlane_profiling.h

@@ -504,8 +504,7 @@ enum class L2SwimlaneSchedPhaseKind : uint32_t {
                         // tasks_processed = subtasks published this iter.
     Release = 2,        // Deferred-release drain (on_task_release work).
                         // tasks_processed = slots released this iter.
-    Wire = 3,           // drain_wiring_queue: pop wired tasks into ready queues.
-                        // tasks_processed = wired count.
+    Wire = 3,           // Legacy scheduler wiring phase, kept for old traces.
     Dummy = 4,          // dummy_drain outer bar: covers handling of all dummies
                         // popped this iter. tasks_processed = dummy_got count.
     EarlyDispatch = 5,  // try_speculative_early_dispatch: speculative pre-staging

src/a2a3/runtime/tensormap_and_ringbuffer/docs/MULTI_RING.md

@@ -135,12 +135,11 @@ struct RingSchedState {
     int32_t last_task_alive;
     std::atomic<int32_t> advance_lock;  // multi-thread CAS
 
-    // Cache Line 1+: Thread 0 only (wiring dep_pool, cache-isolated)
+    // Cache Line 1+: Orch-side wiring dep_pool, cache-isolated
     alignas(64) PTO2DepListPool dep_pool;
 };
 
 RingSchedState ring_sched_states[PTO2_MAX_RING_DEPTH];
-PTO2SpscQueue wiring_queue;  // global SPSC queue: orchestrator pushes, scheduler thread 0 drains
 ```
 
 `slot_states`, `task_window_size`, and `task_window_mask` are no longer duplicated — callers access them via `ring->get_slot_state_by_*()` and other ring header accessors. The ring pointer shares cache line 0 with `last_task_alive` and `advance_lock`.
@@ -198,10 +197,10 @@ Dependency edges use `PTO2TaskSlotState*` pointers, which naturally span rings:
 
 ### 5.3 DepPool Reclamation
 
-DepPool is exclusively managed by scheduler thread 0 (allocation during wiring, reclamation during watermark advancement):
+DepPool entries are allocated by the orchestrator during Orch-side wiring and reclaimed during watermark advancement:
 
 ```text
-// Called by scheduler thread 0 during wiring_queue drain:
+// Called during ring watermark advancement:
 dep_pool_reclaim(ring_id):
     la = ring->fc.last_task_alive
     newest_consumed = la - 1

src/a2a3/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md

@@ -418,7 +418,7 @@ Key members:
 - `rings[PTO2_MAX_RING_DEPTH]`: per-ring `PTO2RingSet` (HeapRing + TaskRing + FaninPool). See [MULTI_RING.md §4.2](MULTI_RING.md).
 - `tensor_map`, `tensor_pool`: dependency tracking
 - `scope_tasks[]`, `scope_begins[]`, `scope_stack_top`: scope nesting stack (flat buffer partitioned by level)
-- `scheduler`: pointer to scheduler state (for wiring queue and ready queue access)
+- `scheduler`: pointer to scheduler state (for Orch-side wiring helpers and ready queue access)
 - `gm_heap_base`, `gm_heap_size`: GM heap for output buffers
 
 ### 7.2 Task Submission Flow (`PTO2OrchestratorState::submit_task`)
@@ -430,17 +430,15 @@ Key members:
 | 2 | Initialize task descriptor + slot state, copy parameters |
 | 3 | **Lookup**: for each INPUT/INOUT param, search TensorMap for producers; collect producer pointers in `PTO2FaninBuilder` |
 | 4 | **Insert**: register OUTPUT/INOUT args in TensorMap |
-| 5 | **Record fanin metadata**: store producer pointers in `payload->fanin_inline_slot_states[]` (+ spill pool if >64); increment each producer's `fanout_count` (no lock needed — single writer). This step runs **before** `payload.init()`. |
-| 6 | **Push to wiring queue**: push to global `PTO2SpscQueue`; scheduler thread 0 asynchronously wires fanout edges (lock + dep_pool + early_finished check + ready push) |
+| 5 | **Record fanin metadata**: store producer pointers in `payload->fanin_inline_slot_states[]` (+ spill pool if >64); claim each live producer by incrementing `fanout_count` under that producer's `fanout_lock`. This step runs **before** `payload.init()`. |
+| 6 | **Orch-side wiring / ready publish**: the orchestrator wires live fanout edges into the per-ring dep_pool; zero-fanin and already-completed fanin tasks publish directly to ready queues |
 
-> **Note**: Fanout wiring (Steps 4–7 in earlier versions) has been moved from the
-> orchestrator submit hot path to the scheduler's global `wiring_queue` (SPSC). This reduces the
-> orchestrator's shared L2 cache / memory bus pressure, as the orchestrator no longer
-> acquires `fanout_lock` or allocates from `dep_pool` during submission.
+> **Note**: Fanout wiring is now completed before publish in the orchestrator submit path.
+> Scheduler threads consume ready queues directly.
 
-### 7.3 Deferred Fanout Wiring (Scheduler Wiring Queue)
+### 7.3 Orch-Side Fanout Wiring
 
-The orchestrator pushes each submitted task to the global `scheduler->wiring_queue` (a wait-free SPSC queue). Scheduler thread 0 drains this queue in batches, deferring if the queue holds fewer than a full batch of items to reduce contention (unless a final flush is needed at end of execution). For each task:
+The orchestrator completes fanout wiring before publishing a task to the ready queues. For each task with live producers:
 
 1. Sets `fanin_count = N + 1` (+1 redundance to prevent premature readiness)
 2. For each producer in `payload->fanin_slot_states[]`:
@@ -449,7 +447,9 @@ The orchestrator pushes each submitted task to the global `scheduler->wiring_que
    - If not completed: prepends consumer to producer's `fanout_head` via `dep_pool.prepend`
    - **Releases** `fanout_lock`
 3. Atomically releases the +1 redundance + early_finished count via `fanin_refcount.fetch_add`
-4. If all deps satisfied: pushes task to ready queue
+4. If all deps satisfied: pushes task to the routed ready queue
+
+Zero-fanin tasks and tasks whose claimed producers are already completed skip dep_pool entry allocation and publish directly to the routed ready queue.
 
 The scheduler's completion handler mirrors this:
 
@@ -700,11 +700,9 @@ With `PTO2_SERIAL_ORCH_SCHED=1`, scheduler threads still wait for
 `runtime_init_ready_` first, then additionally wait for `orchestrator_done_`
 before entering `resolve_and_dispatch()`. The default is off, preserving the
 current overlapped orch/sched pipeline. Serial mode is intended for measurement
-and debugging. During the serial wait, scheduler thread 0 may drain deferred
-wiring records to prevent bounded wiring-queue backpressure, but it does not
-dispatch AICore work until `orchestrator_done_` is set. Large graphs may still
-require larger task-ring, heap, or dependency-pool capacity because no task
-execution/reclaim happens during graph build.
+and debugging. It does not dispatch AICore work until `orchestrator_done_` is
+set. Large graphs may still require larger task-ring, heap, or dependency-pool
+capacity because no task execution/reclaim happens during graph build.
 
 ---
 

src/a2a3/runtime/tensormap_and_ringbuffer/docs/SUBMIT_BY_CLUSTER.md

@@ -163,7 +163,7 @@ This project-defined flattened numbering is kept unchanged.
 1. Validate submit arguments.
 2. Allocate mixed-task ID and initialize descriptor/payload/slot_state once.
 3. Lookup producers via TensorMap; collect fanin metadata and increment producers' `fanout_count`.
-4. Push task to scheduler's wiring queue (scheduler thread 0 asynchronously wires fanout edges and determines readiness).
+4. Wire fanout edges on the orchestrator side and publish ready tasks to the scheduler queues.
 5. Dispatch all active lanes atomically when resources allow.
 6. Aggregate completion and release downstream once.
 

src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.cpp

@@ -279,9 +279,9 @@ static bool append_fanin_or_fail(
     //       producer; ++'ing it would corrupt an unrelated task.
     //   (2) Already CONSUMED in place — finished, output ready, no real edge.
     // In either case, adding it to the fanin and bumping fanout_count would leave
-    // a stale ++/release pair (wire_task drops the fanout edge but keeps the fanin
-    // slot, so on_task_release still release_producer()'s it) that desyncs the
-    // slot's refcount (rc != fc) and wedges in-order reclaim. Claiming a live
+    // a stale ++/release pair (Orch-side wiring drops the fanout edge but keeps
+    // the fanin slot, so on_task_release still release_producer()'s it) that
+    // desyncs the slot's refcount (rc != fc) and wedges in-order reclaim. Claiming a live
     // producer under the lock pins it: fanout_count now counts us, so it cannot
     // reach CONSUMED (rc == fc) until we release it in on_task_release, keeping the
     // slot's generation stable until then. check_and_handle_consumed flips
@@ -297,8 +297,9 @@ static bool append_fanin_or_fail(
     // keyed to the live producer, and doing it before the ++ still suppresses a
     // double-count for a producer named twice in one submission.
     prod_state->lock_fanout();
+    PTO2TaskState pstate = prod_state->task_state.load(std::memory_order_acquire);
     bool gone = prod_state->task == nullptr || prod_state->task->task_id.local() != producer_task_id.local() ||
-                prod_state->task_state.load(std::memory_order_acquire) == PTO2_TASK_CONSUMED;
+                pstate == PTO2_TASK_CONSUMED;
     bool claim = !gone && !fanin_builder->mark_seen(prod_ring, prod_slot);
     if (claim) {
         // Low bits hold the consumer count; bit31 is the scope ref. The consumer
@@ -345,6 +346,83 @@ static bool append_fanin_or_fail(
     return true;
 }
 
+static bool all_claimed_fanin_completed(const PTO2FaninBuilder &fanin_builder) {
+    if (fanin_builder.count == 0) return true;
+    return fanin_builder.for_each([](PTO2TaskSlotState *producer) -> bool {
+        return producer != nullptr && producer->task_state.load(std::memory_order_acquire) >= PTO2_TASK_COMPLETED;
+    });
+}
+
+static void mark_orch_wired_no_edges(PTO2SchedulerState *sched, PTO2TaskSlotState &slot_state) {
+    auto &rss = sched->ring_sched_states[slot_state.ring_id];
+    sched->mark_dep_pool_position(rss, slot_state);
+}
+
+static bool orch_wire_live_fanin_task(PTO2OrchestratorState *orch, PTO2TaskSlotState &slot_state, int32_t wfanin) {
+    PTO2SchedulerState *sched = orch->scheduler;
+    auto &rss = sched->ring_sched_states[slot_state.ring_id];
+
+    // dep_pool is orchestrator-exclusive (no lock). Block until space frees.
+    // A stall here is usually a *symptom* of the scheduler having stalled
+    // (e.g. a scheduler timeout latches sched_error_code); latching our own
+    // dep_pool overflow would mask that root cause. So bail without latching
+    // when any fatal is already set elsewhere, and only declare a genuine
+    // dep_pool deadlock after a backstop that runs STRICTLY LONGER than the
+    // scheduler timeout — so a scheduler-side fault (e.g. -100) surfaces first
+    // — with no forward progress and no other fault.
+    if (rss.dep_pool.available() >= wfanin) {
+        sched->wire_fanin_task(rss, slot_state, wfanin);
+        return true;
+    }
+
+    // Backstop = scheduler timeout + alloc-deadlock slack. Must exceed the
+    // scheduler timeout so the scheduler's own fault (-100) is latched and
+    // observed here before this dep_pool overflow (-4).
+    const uint64_t backstop_cycles =
+        static_cast<uint64_t>(PLATFORM_SCHEDULER_TIMEOUT_MS) * (PLATFORM_PROF_SYS_CNT_FREQ / 1000) +
+        PTO2_ALLOC_DEADLOCK_TIMEOUT_CYCLES;
+
+    int spin_count = 0;
+    int32_t prev_last_alive = rss.ring->fc.last_task_alive.load(std::memory_order_acquire);
+    uint64_t block_cycle0 = 0;
+    bool block_timing = false;
+    while (rss.dep_pool.available() < wfanin) {
+        rss.dep_pool.reclaim(*rss.ring, prev_last_alive);
+        if (rss.dep_pool.available() >= wfanin) break;
+
+        spin_count++;
+        int32_t cur_last_alive = rss.ring->fc.last_task_alive.load(std::memory_order_acquire);
+        if (cur_last_alive > prev_last_alive) {
+            spin_count = 0;
+            prev_last_alive = cur_last_alive;
+            block_timing = false;
+        } else if ((spin_count & 1023) == 0) {
+            // A fatal latched elsewhere (scheduler timeout -> sched_error_code,
+            // or an earlier orch_error_code) is the root cause of the stall —
+            // unwind without latching dep_pool overflow so that code surfaces.
+            if (orch->sm_header->sched_error_code.load(std::memory_order_acquire) != PTO2_ERROR_NONE ||
+                orch->sm_header->orch_error_code.load(std::memory_order_acquire) != PTO2_ERROR_NONE) {
+                orch->fatal = true;
+                return false;
+            }
+            // Absolute-time backstop, stable across contention. get_sys_cnt_aicpu
+            // is an MMIO read, so sample it only once per 1024 spins.
+            uint64_t now = get_sys_cnt_aicpu();
+            if (!block_timing) {
+                block_cycle0 = now;
+                block_timing = true;
+            } else if (now - block_cycle0 >= backstop_cycles) {
+                orch_mark_fatal(orch, PTO2_ERROR_DEP_POOL_OVERFLOW);
+                return false;
+            }
+        }
+        SPIN_WAIT_HINT();
+    }
+
+    sched->wire_fanin_task(rss, slot_state, wfanin);
+    return true;
+}
+
 static void scope_tasks_push(PTO2OrchestratorState *orch, PTO2TaskSlotState *task_slot_state);
 
 struct PTO2PreparedTask {
@@ -428,15 +506,16 @@ static bool prepare_task(
     // returns a slot whose previous occupant is CONSUMED and quiescent (alloc
     // spins until last_task_alive passes it; in-order reclaim + acquire load),
     // and the slot is not published to any scheduler thread until the
-    // wiring.queue.push at the end of submit_task_common — so this reset is
-    // race-free. Doing it here (not relying on the scheduler's eager
+    // Orch-side wiring publish at the end of submit_task_common — so this reset
+    // is race-free. Doing it here (not relying on the scheduler's eager
     // reset-after-CONSUMED, which only covers the contiguously-reclaimed tail)
     // makes every reused slot self-clean, which lets the per-boot SM init skip
     // its O(window) per-slot loop. bind_ring is slot-invariant but cheap to
     // re-assert on the already-dirtied cache line.
     out->slot_state->bind_ring(ring_id);
     out->slot_state->reset_for_reuse();
     out->slot_state->fanin_count = 0;
+    out->slot_state->dep_pool_mark = 0;
 
     out->payload->prefetch(args.tensor_count(), args.scalar_count());
 
@@ -445,14 +524,14 @@ static bool prepare_task(
     // here lets RingSchedState::init() skip the O(window_size) bind loop.
     // Both writes hit the same 64B slot_state cache line we're about to
     // dirty below, so the extra cost is two stores on an already-hot line.
-    // Must precede the scheduler wiring.queue.push at the end of
-    // submit_task_common — that push is the first read of slot_state->task /
-    // slot_state->payload by another thread.
+    // Must precede the Orch-side wiring publish at the end of
+    // submit_task_common — that publish is the first read of slot_state->task /
+    // slot_state->payload by scheduler threads.
     out->slot_state->bind_buffers(out->payload, out->task);
 
     // prepare_task does NO payload writes: all payload content (tensors/scalars +
     // early-dispatch spec fields) is initialized in PTO2TaskPayload::init, the
-    // single payload-init point, which runs before the scheduler wiring push.
+    // single payload-init point, which runs before Orch-side wiring publish.
 
     // Fields already reset by advance_ring_pointers (eager reset after CONSUMED):
     //   fanout_lock=0, fanout_count=PTO2_FANOUT_SCOPE_BIT, fanout_head=nullptr,
@@ -467,7 +546,7 @@ static bool prepare_task(
         static_cast<int16_t>(block_num * __builtin_popcount(active_mask.core_mask()));
     out->slot_state->logical_block_num = block_num;
     out->slot_state->active_mask = active_mask;
-    // fanin_count is set by scheduler during wiring
+    // fanin_count is set during Orch-side wiring
     scope_tasks_push(orch, out->slot_state);
 
     return true;
@@ -685,8 +764,8 @@ static bool ensure_tensormap_capacity(PTO2OrchestratorState *orch, int32_t neede
 // Shared body for submit_task / submit_dummy_task. Caller has already validated
 // args.has_error, decided active_mask (empty for dummy), and resolved the per-slot
 // kernel_ids (all INVALID_KERNEL_ID for dummy). Performs tensormap sync, fanin
-// computation (explicit_deps + auto), output registration, slot init, and pushes
-// to the scheduler wiring queue.
+// computation (explicit_deps + auto), output registration, slot init, and
+// Orch-side wiring/ready publication.
 static TaskOutputTensors submit_task_common(
     PTO2OrchestratorState *orch, const L0TaskArgs &args, ActiveMask active_mask, int32_t aic_kernel_id,
     int32_t aiv0_kernel_id, int32_t aiv1_kernel_id
@@ -864,28 +943,26 @@ static TaskOutputTensors submit_task_common(
 
     CYCLE_COUNT_LAP(g_orch_args_cycle);
 
-    // === STEP 6: push to wiring queue ===
-    // Deferred wiring: orchestrator only stores dependency metadata and increments
-    // fanout_count. The actual fanout_head wiring (lock + dep_pool + early_finished)
-    // is handled asynchronously by scheduler thread 0 via the wiring queue.
-    // Push to global wiring queue — scheduler sets fanin_count, wires fanout, checks readiness
-    if (!sched->wiring.queue.push(&cur_slot_state)) {
-        // producer_blocked is the wiring deadlock detector's "orchestrator is
-        // stuck in push" observable: set ONLY while we actually spin (queue
-        // full), cleared on exit, so the just-filled-then-scope_end case (push
-        // succeeded, no spin) never trips a false deadlock. Also poll the shared
-        // orch_error_code so a fatal latched by any party (e.g. that detector)
-        // breaks this otherwise-unbounded spin and unwinds orchestration.
-        sched->wiring.producer_blocked.store(1, std::memory_order_release);
-        while (!sched->wiring.queue.push(&cur_slot_state)) {
-            if (orch->sm_header->orch_error_code.load(std::memory_order_acquire) != PTO2_ERROR_NONE) {
-                orch->fatal = true;
-                sched->wiring.producer_blocked.store(0, std::memory_order_release);
-                return result;
-            }
-            SPIN_WAIT_HINT();
+    // === STEP 6: wire on the orchestrator side and publish readiness ===
+    // Zero-fanin tasks and tasks whose claimed producers are already completed
+    // do not need fanout links or dep_pool entries. Tasks with live producers
+    // allocate fanout links here before any scheduler thread can dispatch them.
+    if (fanin_builder.count == 0) {
+        cur_slot_state.fanin_count = 1;
+        cur_slot_state.fanin_refcount.store(1, std::memory_order_release);
+        mark_orch_wired_no_edges(sched, cur_slot_state);
+        sched->push_ready_routed(&cur_slot_state);
+    } else if (all_claimed_fanin_completed(fanin_builder)) {
+        int32_t ready_seed = fanin_builder.count + 1;
+        cur_slot_state.fanin_count = ready_seed;
+        payload.dispatch_fanin.store(fanin_builder.count, std::memory_order_release);
+        cur_slot_state.fanin_refcount.store(ready_seed, std::memory_order_release);
+        mark_orch_wired_no_edges(sched, cur_slot_state);
+        sched->push_ready_routed(&cur_slot_state);
+    } else {
+        if (!orch_wire_live_fanin_task(orch, cur_slot_state, fanin_builder.count)) {
+            return result;
         }
-        sched->wiring.producer_blocked.store(0, std::memory_order_release);
     }
 
     CYCLE_COUNT_LAP(g_orch_fanin_cycle);
@@ -1071,7 +1148,7 @@ TaskOutputTensors PTO2OrchestratorState::alloc_tensors(const L0TaskArgs &args) {
         // required so scope_end can release the producer-side reference and
         // drive the slot to CONSUMED, but worker dispatch fields are never
         // observed for hidden alloc tasks.
-        prepared.slot_state->task_state.store(PTO2_TASK_COMPLETED, std::memory_order_release);
+        prepared.slot_state->mark_completed();
     }
     orch->inline_completed_tasks++;