Refactor: move fanout wiring to orchestrator, drop wiring queue

simpler_setup/tools/swimlane_converter.py

@@ -1401,7 +1401,6 @@ 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
             "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
@@ -1540,7 +1539,7 @@ def _find_containing_complete(thread_idx: int, finish_us: float):
                         }
                     )
                     continue
-                if phase not in ("complete", "dispatch", "release", "resolve", "early_dispatch", "wire", "dummy"):
+                if phase not in ("complete", "dispatch", "release", "resolve", "early_dispatch", "dummy"):
                     continue
                 start_us = record["start_time_us"]
                 end_us = record["end_time_us"]

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

@@ -504,8 +504,6 @@ 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.
     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/platform/shared/host/l2_swimlane_collector.cpp

@@ -835,8 +835,6 @@ int L2SwimlaneCollector::export_swimlane_json() {
                 return "dispatch";
             case L2SwimlaneSchedPhaseKind::Release:
                 return "release";
-            case L2SwimlaneSchedPhaseKind::Wire:
-                return "wire";
             case L2SwimlaneSchedPhaseKind::Dummy:
                 return "dummy";
             case L2SwimlaneSchedPhaseKind::EarlyDispatch:

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+: orchestrator-only (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 is exclusively managed by the orchestrator (allocation during wiring, reclamation inside `ensure_space`):
 
 ```text
-// Called by scheduler thread 0 during wiring_queue drain:
+// Called by the orchestrator inside dep_pool.ensure_space when the pool is full:
 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 dep_pool + ready queue access)
 - `gm_heap_base`, `gm_heap_size`: GM heap for output buffers
 
 ### 7.2 Task Submission Flow (`PTO2OrchestratorState::submit_task`)
@@ -431,25 +431,35 @@ Key members:
 | 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) |
-
-> **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.
-
-### 7.3 Deferred Fanout Wiring (Scheduler Wiring Queue)
-
-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:
-
-1. Sets `fanin_count = N + 1` (+1 redundance to prevent premature readiness)
-2. For each producer in `payload->fanin_slot_states[]`:
-   - **Acquires** the producer's `fanout_lock`
+| 6 | **Wire fanout inline (orchestrator-side)**: the orchestrator acquires each producer's `fanout_lock`, allocates `dep_pool` entries for live producers (blocking in `dep_pool.ensure_space` if the pool is full), and routes the task ready once its fanin refcount is satisfied |
+
+> **Note**: Fanout wiring runs on the orchestrator at submit time. `dep_pool` is
+> touched only by the orchestrator (single writer, no lock) — the scheduler never
+> allocates from or reclaims it. This mirrors the `fanin_pool.ensure_space` already
+> used earlier on the submit path; the only cross-thread serialization is the
+> per-producer `fanout_lock`, shared with the scheduler's completion handler.
+
+### 7.3 Fanout Wiring (Orchestrator-Side, Inline)
+
+The orchestrator wires each task's fanout edges directly at the end of
+`submit_task_common`; there is no deferred wiring queue. Because `dep_pool` has a
+single writer (the orchestrator), it needs no lock. For each task:
+
+1. If the task has live producers, blocks in `dep_pool.ensure_space`
+   (reclaim + spin + overflow detect) until `wfanin` entries are free.
+2. Sets `fanin_count = N + 1` (+1 redundance to prevent premature readiness)
+3. For each producer in `payload->fanin_inline_slot_states[]` (+ spill pool):
+   - **Acquires** the producer's `fanout_lock` — this is the one cross-thread
+     edge, serializing against the scheduler's completion handler walking the
+     same producer's `fanout_head`
    - Checks `task_state >= COMPLETED` (early-finished optimization)
-   - If not completed: prepends consumer to producer's `fanout_head` via `dep_pool.prepend`
+   - 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. Atomically releases the +1 redundance + early_finished count via
+   `fanin_refcount.fetch_add`
+5. If all deps satisfied: pushes task to the ready queue (exactly-once, via the
+   `fanin_refcount == fanin_count` transition)
 
 The scheduler's completion handler mirrors this:
 
@@ -700,11 +710,12 @@ 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. Because the orchestrator wires every task itself at submit time,
+serial mode performs all fanout wiring during graph build; no AICore work runs
+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 (dep_pool entries accumulate until the orchestrator
+finishes and the scheduler begins dispatch).
 
 ---
 

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 inline on the orchestrator (dep_pool + per-producer `fanout_lock`); route the task ready if its fanin is already satisfied.
 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

@@ -427,8 +427,8 @@ static bool prepare_task(
     // Reset the fanout/fanin bookkeeping for this reuse. The allocator only
     // 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
+    // and the slot is not published to any scheduler thread until
+    // submit_task_common's wire_task at the end — 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
@@ -445,14 +445,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 wire_task call at the end of submit_task_common —
+    // wire_task is the first read of slot_state->task / slot_state->payload by
+    // another thread (producer fanout_lock + ready-queue push).
     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 wire_task.
 
     // Fields already reset by advance_ring_pointers (eager reset after CONSUMED):
     //   fanout_lock=0, fanout_count=PTO2_FANOUT_SCOPE_BIT, fanout_head=nullptr,
@@ -686,8 +686,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
+// in-line fanout wiring.
 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
@@ -865,28 +865,23 @@ 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 fanout edges inline (orchestrator-side) ===
+    // The orchestrator wires each task's fanout edges directly; there is no
+    // deferred wiring queue. dep_pool is touched only here, so it is
+    // orchestrator-exclusive and needs no lock (previously it was thread-0 of
+    // the scheduler for the same single-writer reason). For tasks with live
+    // producers we block in ensure_space (reclaim + spin + overflow detect) —
+    // the same pattern as fanin_pool.ensure_space in append_fanin_or_fail.
+    // wire_task links fanout_head, seeds fanin_refcount, and routes the task
+    // ready itself once its fanin is satisfied.
+    {
+        auto &rss = sched->ring_sched_states[ring_id];
+        int32_t wfanin = payload.fanin_actual_count;
+        if (wfanin > 0 && !rss.dep_pool.ensure_space(*rss.ring, wfanin)) {
+            orch_mark_fatal(orch, PTO2_ERROR_DEP_POOL_OVERFLOW);
+            return result;
         }
-        sched->wiring.producer_blocked.store(0, std::memory_order_release);
+        sched->wire_task(rss, &cur_slot_state, wfanin);
     }
 
     CYCLE_COUNT_LAP(g_orch_fanin_cycle);

src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_runtime2.cpp

@@ -108,7 +108,6 @@ static bool wait_for_tensor_ready(PTO2Runtime *rt, const Tensor &tensor, bool wa
     constexpr int kSegmentCap = 64;
     const PTO2TaskSlotState *seg[kSegmentCap];
     int seg_count = 0;
-    bool signaled = false;
     bool failed = false;
 
     auto wait_one_producer = [&](const PTO2TaskSlotState &slot) {
@@ -186,10 +185,6 @@ static bool wait_for_tensor_ready(PTO2Runtime *rt, const Tensor &tensor, bool wa
             if (failed) return;
         }
         seg[seg_count++] = &s;
-        if (!signaled) {
-            orch.scheduler->wiring.orch_needs_drain.store(true, std::memory_order_release);
-            signaled = true;
-        }
     };
 
     auto do_wait = [&]() {
@@ -212,9 +207,6 @@ static bool wait_for_tensor_ready(PTO2Runtime *rt, const Tensor &tensor, bool wa
     };
 
     do_wait();
-    if (signaled) {
-        orch.scheduler->wiring.orch_needs_drain.store(false, std::memory_order_release);
-    }
     return !failed;
 }
 MAYBE_UNINITIALIZED_END

src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_runtime2.h

@@ -223,8 +223,8 @@ PTO2Runtime *runtime_init_data_from_layout(
 /**
  * Phase 3 — wire every arena-internal pointer field (rt->sm_handle,
  * rt->aicore_mailbox, orchestrator.{scope_tasks, scope_begins, scheduler,
- * tensor_map.*, rings[].fanin_pool.base}, scheduler.{ready_queues, dep_pool,
- * wiring.queue}) so each holds arena.base() + offset. Idempotent — runs on
+ * tensor_map.*, rings[].fanin_pool.base}, scheduler.{ready_queues, dep_pool})
+ * so each holds arena.base() + offset. Idempotent — runs on
  * both host (writing host-mirror addresses) and AICPU (writing device
  * addresses) sides.
  */

src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_runtime2_types.h

@@ -90,9 +90,6 @@
 // Cross-thread early-dispatch work queue (power of two)
 #define PTO2_EARLY_DISPATCH_QUEUE_SIZE 64
 
-// Wiring queue
-#define PTO2_WRIRING_QUEUE_SIZE 1024  // Per-shape queue size
-
 // Fanin storage
 #define PTO2_FANIN_INLINE_CAP 64