From 0712132c48cf6ef6f755af727a153d0c6aafd986 Mon Sep 17 00:00:00 2001 From: Chao Wang <26245345+ChaoWao@users.noreply.github.com> Date: Fri, 3 Jul 2026 14:38:34 +0800 Subject: [PATCH] Refactor: move fanout wiring to the orchestrator, remove the wiring queue MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The orchestrator now wires each task's fanout edges inline at submit time (dep_pool.ensure_space + wire_task) instead of pushing to a deferred wiring queue drained by scheduler thread 0. The whole wiring-queue subsystem is deleted: PTO2SpscQueue, WiringState, drain_wiring_queue, report_wiring_deadlock, producer_blocked, orch_needs_drain, dep_deadlock_reported, and the Wire swimlane phase. - dep_pool becomes orchestrator-exclusive (single writer), so it needs no lock; the only cross-thread serialization is the existing per-producer fanout_lock, shared with the scheduler's completion handler. - Forward-progress safety moves from the S-side structural wiring-deadlock detector to PTO2DepListPool::ensure_space's existing no-progress overflow detector (same PTO2_ERROR_DEP_POOL_OVERFLOW fatal), mirroring the fanin_pool.ensure_space already used on the submit path. - wait_for_tensor_ready's orch_needs_drain signal is removed: with O wiring every task at submit time, the producer it waits on is already wired, and the scheduler dispatch loop is a busy poll that needs no "drain now" nudge. Docs (RUNTIME_LOGIC.md, MULTI_RING.md, SUBMIT_BY_CLUSTER.md) updated to the O-side wiring model. test_spsc_queue.cpp (a2a3) removed; the drain_wiring_queue cases dropped from test_wiring.cpp (wire_task cases retained). Testing: a2a3sim tensormap_and_ringbuffer suite — 30 passed, 1 skipped. --- simpler_setup/tools/swimlane_converter.py | 3 +- .../include/common/l2_swimlane_profiling.h | 2 - .../shared/host/l2_swimlane_collector.cpp | 2 - .../docs/MULTI_RING.md | 7 +- .../docs/RUNTIME_LOGIC.md | 57 ++-- .../docs/SUBMIT_BY_CLUSTER.md | 2 +- .../runtime/pto_orchestrator.cpp | 53 ++- .../runtime/pto_runtime2.cpp | 8 - .../runtime/pto_runtime2.h | 4 +- .../runtime/pto_runtime2_types.h | 3 - .../runtime/scheduler/pto_scheduler.h | 297 +---------------- .../runtime/scheduler/scheduler_cold_path.cpp | 27 +- .../runtime/scheduler/scheduler_context.h | 3 +- .../runtime/scheduler/scheduler_dispatch.cpp | 30 -- .../runtime/scheduler/scheduler_types.h | 2 - .../runtime/shared/pto_runtime2_init.cpp | 20 +- tests/ut/cpp/CMakeLists.txt | 1 - tests/ut/cpp/a2a3/test_spsc_queue.cpp | 311 ------------------ tests/ut/cpp/a2a3/test_wiring.cpp | 67 ---- 19 files changed, 73 insertions(+), 826 deletions(-) delete mode 100644 tests/ut/cpp/a2a3/test_spsc_queue.cpp diff --git a/simpler_setup/tools/swimlane_converter.py b/simpler_setup/tools/swimlane_converter.py index 0dd802204..9cf91e80f 100644 --- a/simpler_setup/tools/swimlane_converter.py +++ b/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"] diff --git a/src/a2a3/platform/include/common/l2_swimlane_profiling.h b/src/a2a3/platform/include/common/l2_swimlane_profiling.h index b7a4fc354..63429ba09 100644 --- a/src/a2a3/platform/include/common/l2_swimlane_profiling.h +++ b/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 diff --git a/src/a2a3/platform/shared/host/l2_swimlane_collector.cpp b/src/a2a3/platform/shared/host/l2_swimlane_collector.cpp index 6572a498b..d24b333f8 100644 --- a/src/a2a3/platform/shared/host/l2_swimlane_collector.cpp +++ b/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: diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/docs/MULTI_RING.md b/src/a2a3/runtime/tensormap_and_ringbuffer/docs/MULTI_RING.md index dbbfb5cd0..d9c02472b 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/docs/MULTI_RING.md +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/docs/MULTI_RING.md @@ -135,12 +135,11 @@ struct RingSchedState { int32_t last_task_alive; std::atomic 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 diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md b/src/a2a3/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md index bcc50f25d..98c5ebe3c 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md +++ b/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). --- diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/docs/SUBMIT_BY_CLUSTER.md b/src/a2a3/runtime/tensormap_and_ringbuffer/docs/SUBMIT_BY_CLUSTER.md index 50f734fee..c02447c5a 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/docs/SUBMIT_BY_CLUSTER.md +++ b/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. diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.cpp b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.cpp index 89f0bcbb3..838b26c52 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.cpp +++ b/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); diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_runtime2.cpp b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_runtime2.cpp index 83a44c957..75390879c 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_runtime2.cpp +++ b/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 diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_runtime2.h b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_runtime2.h index 64f4c6319..6426ee386 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_runtime2.h +++ b/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. */ diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_runtime2_types.h b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_runtime2_types.h index 01194134a..7bc53e5d4 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_runtime2_types.h +++ b/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 diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/pto_scheduler.h b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/pto_scheduler.h index 3ed6058ce..f85574dbb 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/pto_scheduler.h +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/pto_scheduler.h @@ -386,135 +386,6 @@ bool ready_queue_init_data_from_layout(PTO2ReadyQueue *queue, DeviceArena &arena void ready_queue_wire_arena_pointers(PTO2ReadyQueue *queue, DeviceArena &arena, size_t slots_off); void ready_queue_destroy(PTO2ReadyQueue *queue); -// ============================================================================= -// SPSC Queue (Single-Producer Single-Consumer, wait-free) -// ============================================================================= -// -// Bounded ring buffer optimized for the wiring queue use case: -// - Producer: orchestrator thread (push) -// - Consumer: scheduler thread 0 (pop_batch) -// -// Design based on Rigtorp's cached-index technique: each side caches -// the other's index locally, avoiding cross-core cache line bouncing -// on the hot path. Only when the local cache says "full" or "empty" -// does the thread issue an acquire load on the remote index. -// -// Memory layout: 5 cache-line-aligned fields ensure zero false sharing. - -struct alignas(64) PTO2SpscQueue { - // --- Producer cache lines (orchestrator thread) --- - alignas(64) std::atomic head_{0}; - alignas(64) uint64_t tail_cached_{0}; - - // --- Consumer cache lines (scheduler thread 0) --- - alignas(64) std::atomic tail_{0}; - alignas(64) uint64_t head_cached_{0}; - - // --- Shared Cacheline (read only) with mask and data ptr (immutable after init) --- - alignas(64) PTO2TaskSlotState **buffer_{nullptr}; - uint64_t mask_{0}; - - // Padding to exactly 5 cache lines - char padding[64 - sizeof(PTO2TaskSlotState **) - sizeof(uint64_t)]; - - // Reserve the backing buffer region on the supplied arena. Returns the - // region offset, to be passed to init_from_layout() after the arena is - // committed. Cache-line aligned: the buffer is shared between the - // orchestrator (push) and scheduler thread 0 (pop_batch), so its base - // must not false-share with neighboring regions. - static size_t reserve_layout(DeviceArena &arena, uint64_t capacity) { - return arena.reserve(capacity * sizeof(uintptr_t), PTO2_ALIGN_SIZE); - } - - // Writes everything except the arena-internal `buffer_` pointer field - // (zeros the slot pointer array, mask/head/tail). The host pre-builds the - // image without storing a host address in buffer_; the AICPU wires - // buffer_ at boot via wire_arena_pointers(). - bool init_data_from_layout(DeviceArena &arena, size_t buffer_off, uint64_t capacity) { - if (capacity == 0 || (capacity & (capacity - 1)) != 0) return false; - auto *buf = static_cast(arena.region_ptr(buffer_off)); - // calloc'd-equivalent: zero the slot pointers so spurious early pops - // observe nullptr. - for (uint64_t i = 0; i < capacity; i++) - buf[i] = nullptr; - mask_ = capacity - 1; - head_.store(0, std::memory_order_relaxed); - tail_.store(0, std::memory_order_relaxed); - tail_cached_ = 0; - head_cached_ = 0; - return true; - } - - // Wire the arena-internal pointer. Called by both host (with host arena) - // and AICPU (with device arena attached to the prebuilt image). - void wire_arena_pointers(DeviceArena &arena, size_t buffer_off) { - buffer_ = static_cast(arena.region_ptr(buffer_off)); - } - - void reset_for_reuse() { - uint64_t h = head_.load(std::memory_order_relaxed); - tail_.store(h, std::memory_order_relaxed); - tail_cached_ = h; - head_cached_ = h; - } - - // Arena owns the buffer; here we only forget our pointer. - void destroy() { buffer_ = nullptr; } - - // Push one item (producer only). Returns false if queue is full. - // Full condition: next_h - tail > mask_ (i.e. > capacity-1), so the - // effective usable capacity is capacity-1 (one slot is wasted as a - // sentinel to distinguish full from empty). uint64_t wrapping is safe - // since head and tail are monotonically increasing and subtraction - // wraps correctly. - bool push(PTO2TaskSlotState *item) { - uint64_t h = head_.load(std::memory_order_relaxed); - uint64_t next_h = h + 1; - if (next_h - tail_cached_ > mask_) { - tail_cached_ = tail_.load(std::memory_order_acquire); - if (next_h - tail_cached_ > mask_) { - return false; - } - } - buffer_[h & mask_] = item; - head_.store(next_h, std::memory_order_release); - return true; - } - - // Pop up to max_count items (consumer only). Returns actual count. - int pop_batch(PTO2TaskSlotState **out, int max_count) { - uint64_t t = tail_.load(std::memory_order_relaxed); - uint64_t avail = head_cached_ - t; - if (avail < static_cast(max_count)) { - head_cached_ = head_.load(std::memory_order_acquire); - avail = head_cached_ - t; - if (avail == 0) return 0; - } - int count = (avail < static_cast(max_count)) ? static_cast(avail) : max_count; - for (int i = 0; i < count; i++) { - out[i] = buffer_[(t + i) & mask_]; - } - tail_.store(t + count, std::memory_order_release); - return count; - } - - // Approximate size (used for backoff decisions, not exact). - uint64_t size() const { - uint64_t h = head_.load(std::memory_order_acquire); - uint64_t t = tail_.load(std::memory_order_acquire); - return h - t; - } - - // Full ⟺ the producer's next push() would fail: size has reached the - // usable capacity (mask_ = capacity - 1, one slot reserved as sentinel). - // Used by the wiring-queue deadlock detector to prove the orchestrator is - // blocked in push(). - bool full() const { return size() >= mask_; } -}; - -static_assert(sizeof(PTO2SpscQueue) == 5 * 64, "PTO2SpscQueue must be exactly 5 cache lines (320B)"); -// ============================================================================= - /** * Statistics returned by mixed-task completion processing */ @@ -535,9 +406,7 @@ struct PTO2SchedulerLayout { size_t off_dummy_ready_queue_slots; size_t off_early_dispatch_queue_slots; size_t off_dep_pool_entries[PTO2_MAX_RING_DEPTH]; - size_t off_wiring_spsc_buffer; uint64_t ready_queue_capacity; - uint64_t spsc_capacity; int32_t dep_pool_capacities[PTO2_MAX_RING_DEPTH]; }; @@ -559,12 +428,8 @@ struct PTO2SchedulerState { int32_t last_task_alive; std::atomic advance_lock; // multi-thread CAS - // --- Cache Line 1+: Thread 0 only (wiring dep_pool) --- + // --- Cache Line 1+: dep_pool (orchestrator-only writer, no lock) --- alignas(64) PTO2DepListPool dep_pool; - // One-shot latch for the wiring-queue deadlock report (thread 0 only): - // the drain breaks on dep_pool exhaustion every call while wedged, so - // the tier-1 structural diagnostic is emitted once, not per call. - bool dep_deadlock_reported = false; #if PTO2_PROFILING // Published only for scope_stats; orchestrator must not read dep_pool's non-atomic counters directly. alignas(64) std::atomic dep_pool_snapshot_tail; @@ -627,41 +492,6 @@ struct PTO2SchedulerState { // the dispatch loop and completed inline -- never goes to AICore. PTO2ReadyQueue dummy_ready_queue; - // Wiring subsystem — groups all wiring-related state for cache-line isolation. - // - // Three cache-line regions by writer: - // 1. batch_* / backoff — thread 0 exclusive (local batch buffer) - // 2. queue — SPSC: orchestrator push, thread 0 pop - // 3. orch_needs_drain — orchestrator write, thread 0 read - struct alignas(64) WiringState { - static constexpr uint64_t BATCH_SIZE = 30; - static constexpr int BACKOFF_LIMIT = 32; - - // --- Thread 0 exclusive: local batch buffer + backoff --- - int batch_count = 0; - int batch_index = 0; - int backoff_counter = 0; - PTO2TaskSlotState *batch[BATCH_SIZE]; - - // --- SPSC queue: orchestrator (push) ↔ thread 0 (pop) --- - PTO2SpscQueue queue; - - // --- Orchestrator write, thread 0 read --- - alignas(64) std::atomic orch_needs_drain{false}; - // Set to 1 only while the orchestrator is actually spinning in - // queue.push() (queue full), cleared on a successful push. The wiring - // deadlock detector reads this as the producer-blocked observable: it - // proves the orchestrator is stuck BEFORE its scope_end, as opposed to - // having just filled the queue with its last in-scope push and being - // about to call scope_end (which would release the head -> no deadlock). - std::atomic producer_blocked{0}; - } wiring; - - static_assert( - offsetof(WiringState, queue) == 256, "WiringState: batch region must be exactly 4 cache lines before queue" - ); - static_assert(sizeof(WiringState) == 640, "WiringState must be exactly 10 cache lines (640B)"); - alignas(64) AsyncWaitList async_wait_list; // Statistics (cold path, isolated from hot-path fields) @@ -673,129 +503,6 @@ struct PTO2SchedulerState { // Inline hot-path methods // ========================================================================= - /** - * Drain wiring queue: pop submitted tasks and wire their fanout edges. - * Called by scheduler thread 0 each loop iteration. Sets fanin_count, - * acquires fanout_lock per producer, allocates dep_pool entries, and - * pushes ready tasks to the appropriate ready queue. - * - * @return Number of tasks wired this call. - */ - - int drain_wiring_queue(bool force_drain = false) { - int wired = 0; - - // Refill local batch buffer when exhausted. - if (wiring.batch_index >= wiring.batch_count) { - // Backoff: defer pop when queue holds fewer than a full batch, - // unless force_drain, orch_needs_drain, or backoff limit reached. - if (!force_drain && wiring.queue.size() < WiringState::BATCH_SIZE) { - if (!wiring.orch_needs_drain.load(std::memory_order_acquire) && - wiring.backoff_counter < WiringState::BACKOFF_LIMIT) { - wiring.backoff_counter++; - return 0; - } - } - wiring.backoff_counter = 0; - wiring.batch_count = wiring.queue.pop_batch(wiring.batch, WiringState::BATCH_SIZE); - wiring.batch_index = 0; - if (wiring.batch_count == 0) return 0; - } - - // Process tasks from local buffer in strict FIFO order. - while (wiring.batch_index < wiring.batch_count) { - PTO2TaskSlotState *ws = wiring.batch[wiring.batch_index]; - int ring_id = ws->ring_id; - auto &rss = ring_sched_states[ring_id]; - int32_t wfanin = ws->payload->fanin_actual_count; - - if (wfanin > 0 && rss.dep_pool.available() < wfanin) { - rss.dep_pool.reclaim(*rss.ring, rss.last_task_alive); - if (rss.dep_pool.available() < wfanin) { -#if PTO2_PROFILING - if (is_scope_stats_enabled()) { - rss.publish_dep_pool_snapshot(); - } -#endif - // dep_pool can't reclaim because the reclaim watermark is - // wedged. This runs on the scheduler thread, so unlike - // alloc()'s detector it cannot self-observe that the - // orchestrator is blocked; wiring.producer_blocked is the - // external certificate -- the orchestrator sets it ONLY while - // it is actually spinning in queue.push() (cleared on a - // successful push), so the "just filled the queue then called - // scope_end" case (push succeeded -> flag stays 0) cannot trip - // a false report. With the producer provably stuck in push - // (program-order before its scope_end) AND the head COMPLETED, - // all consumers released, scope still open (only scope_end - // frees it), scope_end can never run -> provable head-of-line - // deadlock. The producer-blocked gate also pins the head: - // scope_end has not run, so the scope-gated head cannot be - // CONSUMED/reset concurrently while we read it. - if (!rss.dep_deadlock_reported && wiring.producer_blocked.load(std::memory_order_acquire) != 0) { - int32_t last_alive = rss.last_task_alive; - PTO2TaskSlotState &h = rss.ring->get_slot_state_by_task_id(last_alive); - // Read the head under its fanout_lock: fanout_count is a - // lock-protected field, and one snapshot keeps the check - // and the report consistent. - h.lock_fanout(); - int32_t state = h.task_state.load(std::memory_order_acquire); - uint32_t fc = h.fanout_count; - uint32_t rc = h.fanout_refcount.load(std::memory_order_acquire); - h.unlock_fanout(); - bool head_scope_gated = (state == PTO2_TASK_COMPLETED) && (rc == (fc & ~PTO2_FANOUT_SCOPE_BIT)); - if (head_scope_gated) { - rss.dep_deadlock_reported = true; - report_wiring_deadlock(rss, wfanin, last_alive, state, fc, rc); - // Latch the shared fatal so both sides exit fast off - // one error code: the scheduler cold-path poll - // (handle_orchestrator_exit) emergency_shutdowns, and - // the orchestrator's push spin breaks out and unwinds. - if (rss.dep_pool.error_code_ptr != nullptr) { - int32_t expected = PTO2_ERROR_NONE; - rss.dep_pool.error_code_ptr->compare_exchange_strong( - expected, PTO2_ERROR_DEP_POOL_OVERFLOW, std::memory_order_acq_rel - ); - } - } - } - break; // not enough dep_pool space — keep remainder for next call - } - } - - wiring.batch_index++; - wire_task(rss, ws, wfanin); - wired++; - } - - return wired; - } - - // Tier-1 structural diagnostic for a provable wiring-queue deadlock (head - // COMPLETED + all consumers released + scope still open, dep_pool exhausted, - // orchestrator provably blocked in push). The head snapshot (state/fc/rc) is - // taken under fanout_lock by the caller and passed in, so the report agrees - // with the check and reads no lock-protected field unlocked. - void report_wiring_deadlock( - RingSchedState &rss, int32_t wfanin, int32_t last_alive, int32_t state, uint32_t fc, uint32_t rc - ) { - LOG_ERROR("========================================"); - LOG_ERROR("FATAL: Wiring-Queue Deadlock - Dep Pool Exhausted!"); - LOG_ERROR("========================================"); - LOG_ERROR("Head task %d COMPLETED, all consumers released, scope still open ->", last_alive); - LOG_ERROR("only scope_end can free it, but the orchestrator is blocked on a full wiring"); - LOG_ERROR("queue (in push, before its scope_end). Provable head-of-line deadlock."); - LOG_ERROR( - " Head task %d: state=%d, consumers=%u/%u, scope_released=%d", last_alive, state, - rc & ~PTO2_FANOUT_SCOPE_BIT, fc & ~PTO2_FANOUT_SCOPE_BIT, (rc & PTO2_FANOUT_SCOPE_BIT) ? 1 : 0 - ); - LOG_ERROR(" Dep pool: used=%d/%d, needed=%d entries", rss.dep_pool.used(), rss.dep_pool.capacity, wfanin); - LOG_ERROR("Solution:"); - LOG_ERROR(" The open scope's fanout exceeds the dep pool. Either split the scope, or"); - LOG_ERROR(" raise PTO2_RING_DEP_POOL (compile-time PTO2_DEP_LIST_POOL_SIZE)."); - LOG_ERROR("========================================"); - } - // Route a ready slot to the right global queue. Dummy tasks (empty // active_mask) live in dummy_ready_queue; everything else goes to the // per-shape ready_queues[]. @@ -1367,7 +1074,7 @@ struct PTO2SchedulerState { // Phase 3b: write the arena-internal pointer fields // (ready_queues[].slots, dummy_ready_queue.slots, dep_pool.base for each - // ring, wiring.queue.buffer_). Called on both host and device sides. + // ring). Called on both host and device sides. void wire_arena_pointers(const PTO2SchedulerLayout &layout, DeviceArena &arena); // Forget per-region pointers; arena owns the backing memory. diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_cold_path.cpp b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_cold_path.cpp index 8e1813367..616b21e96 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_cold_path.cpp +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_cold_path.cpp @@ -507,8 +507,8 @@ void SchedulerContext::log_l2_swimlane_summary(int32_t thread_idx, [[maybe_unuse cycles_to_us(sched_end_ts - l2_swimlane.sched_start_ts) ); - uint64_t sched_total = l2_swimlane.sched_wiring_cycle + l2_swimlane.sched_complete_cycle + - l2_swimlane.sched_dispatch_cycle + l2_swimlane.sched_idle_cycle; + uint64_t sched_total = + l2_swimlane.sched_complete_cycle + l2_swimlane.sched_dispatch_cycle + l2_swimlane.sched_idle_cycle; if (sched_total == 0) sched_total = 1; { @@ -600,18 +600,8 @@ void SchedulerContext::log_l2_swimlane_summary(int32_t thread_idx, [[maybe_unuse l2_swimlane.sched_dispatch_setup_cycle * 100.0 / d_parent ); -#if PTO2_SCHED_PROFILING - LOG_INFO_V9( - "Thread %d: wiring : %.3fus (%.1f%%) tasks=%d", thread_idx, - cycles_to_us(l2_swimlane.sched_wiring_cycle), l2_swimlane.sched_wiring_cycle * 100.0 / sched_total, - l2_swimlane.phase_wiring_count - ); -#else - LOG_INFO_V9( - "Thread %d: wiring : %.3fus (%.1f%%)", thread_idx, cycles_to_us(l2_swimlane.sched_wiring_cycle), - l2_swimlane.sched_wiring_cycle * 100.0 / sched_total - ); -#endif + // Wiring is no longer a scheduler phase (the orchestrator wires inline + // at submit time), so there is no per-iteration wiring bar to report. LOG_INFO_V9( "Thread %d: idle : %.3fus (%.1f%%)", thread_idx, cycles_to_us(l2_swimlane.sched_idle_cycle), @@ -1037,15 +1027,6 @@ void SchedulerContext::bind_runtime(PTO2Runtime *rt) { void SchedulerContext::wait_for_orchestration_done_before_dispatch(Runtime *runtime, int32_t thread_idx) { while (!orchestration_done() && !completed_.load(std::memory_order_acquire)) { - if (thread_idx == 0 && sched_ != nullptr) { - // Use the wiring subsystem's normal batch/backoff policy while - // waiting. This still honors orch_needs_drain/producer_blocked - // signals without force-draining an empty queue every spin. - int wired = sched_->drain_wiring_queue(/*force_drain=*/false); - if (wired > 0) { - continue; - } - } if (sched_ != nullptr && sched_->sm_header != nullptr && check_idle_fatal_error(thread_idx, sched_->sm_header, runtime) == LoopAction::BREAK_LOOP) { break; diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_context.h b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_context.h index 9ce98ac5f..1041698c8 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_context.h +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_context.h @@ -297,8 +297,7 @@ class SchedulerContext { // True if mix tasks remain in the global MIX ready queue. Approximate — // PTO2ReadyQueue::size() (see pto_scheduler.h) snapshots its enqueue/dequeue // positions with std::memory_order_relaxed and may interleave with concurrent - // push/pop. Don't confuse with PTO2SpscQueue::size(), which uses acquire - // loads — that one isn't on this path. A stale read here causes at most one + // push/pop. A stale read here causes at most one // extra/missed AIC/AIV skip and self-corrects on the next loop iteration. bool has_residual_mix() const { return sched_->ready_queues[static_cast(PTO2ResourceShape::MIX)].size() > 0; diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_dispatch.cpp b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_dispatch.cpp index b7e3fa0f7..c333ea63e 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_dispatch.cpp +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_dispatch.cpp @@ -931,36 +931,6 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ continue; } - // Phase 3: Drain wiring queue (thread 0 only) - int wired = 0; - if (thread_idx == 0) { - wired = sched_->drain_wiring_queue(orchestrator_done_.load(std::memory_order_relaxed)); - if (wired > 0) { - made_progress = true; -#if PTO2_SCHED_PROFILING - l2_swimlane.phase_wiring_count += wired; -#endif - } - } -#if PTO2_PROFILING - CYCLE_COUNT_LAP(l2_swimlane.sched_wiring_cycle); - // Wire outer phase: emit one bar covering this iter's drain_wiring_queue - // pass when it wired any tasks. tasks_processed = wired count. Resolve - // does NOT nest under Wire — wiring only enqueues, the consumer release - // happens later in Complete/Dummy. - if (l2_swimlane_level_ >= L2SwimlaneLevel::SCHED_PHASES && wired > 0) { - int16_t phase_end_shared[L2SWIMLANE_NUM_QUEUE_SHAPES]; - capture_phase_end_fresh(phase_end_shared); - l2_swimlane_aicpu_record_sched_phase( - thread_idx, L2SwimlaneSchedPhaseKind::Wire, _t0_phase, _t1, l2_swimlane.sched_loop_count, - static_cast(wired), /*pop_hit=*/0, /*pop_miss=*/0, phase_start_shared, phase_end_shared - ); - for (int s = 0; s < L2SWIMLANE_NUM_QUEUE_SHAPES; s++) - phase_start_shared[s] = phase_end_shared[s]; - _t0_phase = _t1; - } -#endif - // Phase 3b: Drain dummy ready queue (thread 0 only). // // Dependency-only tasks bypass AICore dispatch: they go through the diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_types.h b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_types.h index 445c46a56..5040e0257 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_types.h +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_types.h @@ -419,7 +419,6 @@ struct alignas(64) SchedL2SwimlaneCounters { uint64_t sched_start_ts{0}; uint64_t sched_complete_cycle{0}; uint64_t sched_dispatch_cycle{0}; - uint64_t sched_wiring_cycle{0}; uint64_t sched_idle_cycle{0}; uint64_t sched_loop_count{0}; uint32_t phase_complete_count{0}; @@ -437,7 +436,6 @@ struct alignas(64) SchedL2SwimlaneCounters { uint64_t pop_hit_at_last_emit{0}; uint64_t pop_miss_at_last_emit{0}; #if PTO2_SCHED_PROFILING - uint32_t phase_wiring_count{0}; uint64_t complete_probe_count{0}; uint64_t complete_hit_count{0}; uint64_t sched_complete_perf_cycle{0}; diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/shared/pto_runtime2_init.cpp b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/shared/pto_runtime2_init.cpp index 1561acc56..722deadfe 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/shared/pto_runtime2_init.cpp +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/shared/pto_runtime2_init.cpp @@ -110,7 +110,6 @@ void PTO2SchedulerState::RingSchedState::reset_for_reuse( ring = pto2_sm_layout::ring_header_addr(sm_dev_base, ring_id); last_task_alive = 0; advance_lock.store(0, std::memory_order_relaxed); - dep_deadlock_reported = false; dep_pool.reset_for_reuse(orch_err); #if PTO2_PROFILING dep_pool_snapshot_tail.store(1, std::memory_order_relaxed); @@ -132,7 +131,6 @@ PTO2SchedulerLayout PTO2SchedulerState::reserve_layout(DeviceArena &arena, const int32_t dep_pool_capacities[PTO2_MAX_RING_DEPTH]) { PTO2SchedulerLayout layout{}; layout.ready_queue_capacity = PTO2_READY_QUEUE_SIZE; - layout.spsc_capacity = PTO2_WRIRING_QUEUE_SIZE; for (int r = 0; r < PTO2_MAX_RING_DEPTH; r++) { layout.dep_pool_capacities[r] = dep_pool_capacities[r]; } @@ -143,13 +141,12 @@ PTO2SchedulerState::reserve_layout(DeviceArena &arena, const int32_t dep_pool_ca layout.off_dummy_ready_queue_slots = ready_queue_reserve_layout(arena, PTO2_READY_QUEUE_SIZE); layout.off_early_dispatch_queue_slots = ready_queue_reserve_layout(arena, PTO2_EARLY_DISPATCH_QUEUE_SIZE); for (int r = 0; r < PTO2_MAX_RING_DEPTH; r++) { - // Force a cache-line base so writes from scheduler thread 0 (sole + // Force a cache-line base so writes from the orchestrator (sole // writer of this ring's dep_pool) do not invalidate adjacent // multi-threaded regions like ready_queue.slots. layout.off_dep_pool_entries[r] = arena.reserve(static_cast(dep_pool_capacities[r]) * sizeof(PTO2DepListEntry), PTO2_ALIGN_SIZE); } - layout.off_wiring_spsc_buffer = PTO2SpscQueue::reserve_layout(arena, PTO2_WRIRING_QUEUE_SIZE); return layout; } @@ -194,13 +191,6 @@ bool PTO2SchedulerState::init_data_from_layout( sched->ring_sched_states[r].dep_pool.init(dep_entries, layout.dep_pool_capacities[r], orch_err); } - if (!sched->wiring.queue.init_data_from_layout(arena, layout.off_wiring_spsc_buffer, layout.spsc_capacity)) { - return false; - } - sched->wiring.batch_count = 0; - sched->wiring.batch_index = 0; - sched->wiring.backoff_counter = 0; - return true; } @@ -223,12 +213,6 @@ void PTO2SchedulerState::reset_for_reuse(const PTO2SchedulerLayout &layout, void sched->dummy_ready_queue.reset_for_reuse(); sched->early_dispatch_queue.reset_for_reuse(); - sched->wiring.queue.reset_for_reuse(); - sched->wiring.batch_count = 0; - sched->wiring.batch_index = 0; - sched->wiring.backoff_counter = 0; - sched->wiring.orch_needs_drain.store(false, std::memory_order_relaxed); - sched->wiring.producer_blocked.store(0, std::memory_order_relaxed); sched->async_wait_list.reset_for_reuse(); (void)layout; } @@ -244,7 +228,6 @@ void PTO2SchedulerState::wire_arena_pointers(const PTO2SchedulerLayout &layout, sched->ring_sched_states[r].dep_pool.base = static_cast(arena.region_ptr(layout.off_dep_pool_entries[r])); } - sched->wiring.queue.wire_arena_pointers(arena, layout.off_wiring_spsc_buffer); } void PTO2SchedulerState::destroy() { @@ -253,7 +236,6 @@ void PTO2SchedulerState::destroy() { sched->ring_sched_states[r].destroy(); sched->ring_sched_states[r].dep_pool.base = nullptr; } - sched->wiring.queue.destroy(); for (int i = 0; i < PTO2_NUM_RESOURCE_SHAPES; i++) { ready_queue_destroy(&sched->ready_queues[i]); } diff --git a/tests/ut/cpp/CMakeLists.txt b/tests/ut/cpp/CMakeLists.txt index 0c868a45f..00e10c809 100644 --- a/tests/ut/cpp/CMakeLists.txt +++ b/tests/ut/cpp/CMakeLists.txt @@ -564,7 +564,6 @@ add_a2a3_runtime_test(test_shared_memory a2a3/test_shared_memory.cpp) add_a2a3_runtime_test(test_a2a3_tensormap a2a3/test_tensormap.cpp) add_a2a3_runtime_test(test_fanin_pool a2a3/test_fanin_pool.cpp) add_a2a3_orchestrator_runtime_test(test_a2a3_orchestrator_fanin a2a3/test_orchestrator_fanin.cpp) -add_a2a3_runtime_test(test_spsc_queue a2a3/test_spsc_queue.cpp) add_a2a3_runtime_test(test_wiring a2a3/test_wiring.cpp) add_a2a3_runtime_test(test_aicore_completion_mailbox a2a3/test_aicore_completion_mailbox.cpp) add_a2a3_runtime_test(test_a2a3_scope_stats_collector a2a3/test_scope_stats_collector.cpp) diff --git a/tests/ut/cpp/a2a3/test_spsc_queue.cpp b/tests/ut/cpp/a2a3/test_spsc_queue.cpp deleted file mode 100644 index 664342d01..000000000 --- a/tests/ut/cpp/a2a3/test_spsc_queue.cpp +++ /dev/null @@ -1,311 +0,0 @@ -/* - * Copyright (c) PyPTO Contributors. - * This program is free software, you can redistribute it and/or modify it under the terms and conditions of - * CANN Open Software License Agreement Version 2.0 (the "License"). - * Please refer to the License for details. You may not use this file except in compliance with the License. - * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, - * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. - * See LICENSE in the root of the software repository for the full text of the License. - * ----------------------------------------------------------------------------------------------------------- - */ -/** - * Unit tests for PTO2SpscQueue from pto_scheduler.h - * - * Tests the Rigtorp cached-index SPSC queue used as the orchestrator → - * scheduler wiring channel: - * - Basic push / pop_batch correctness - * - Full / empty detection (including cached-index lazy refresh) - * - Wrap-around via modulo indexing - * - Capacity is capacity-1 (one sentinel slot) - * - pop_batch partial reads - * - size() accuracy - */ - -#include - -#include -#include -#include - -#include "utils/device_arena.h" -#include "scheduler/pto_scheduler.h" - -// ============================================================================= -// Fixture -// ============================================================================= - -class SpscQueueTest : public ::testing::Test { -protected: - static constexpr uint64_t CAPACITY = 16; // must be power of 2 - - PTO2SpscQueue queue{}; - DeviceArena arena; - // Dummy slot states used as push values - alignas(64) PTO2TaskSlotState slots[64]{}; - - void SetUp() override { - memset(&queue, 0, sizeof(queue)); - const size_t off = PTO2SpscQueue::reserve_layout(arena, CAPACITY); - ASSERT_NE(arena.commit(), nullptr); - ASSERT_TRUE(queue.init_data_from_layout(arena, off, CAPACITY)); - queue.wire_arena_pointers(arena, off); - } - - void TearDown() override { - queue.destroy(); - arena.release(); - } -}; - -// ============================================================================= -// Initialization -// ============================================================================= - -TEST_F(SpscQueueTest, InitValidState) { - EXPECT_EQ(queue.size(), 0u); - EXPECT_EQ(queue.mask_, CAPACITY - 1); - EXPECT_NE(queue.buffer_, nullptr); -} - -TEST_F(SpscQueueTest, InitRejectsNonPowerOfTwo) { - // init_from_layout rejects non-power-of-two capacities. Use a fresh arena - // each time since reserve runs before commit. - PTO2SpscQueue bad{}; - DeviceArena local; - const size_t off = PTO2SpscQueue::reserve_layout(local, 1); // dummy reservation so commit succeeds - (void)off; - ASSERT_NE(local.commit(), nullptr); - EXPECT_FALSE(bad.init_data_from_layout(local, off, 3)); - EXPECT_FALSE(bad.init_data_from_layout(local, off, 7)); - EXPECT_FALSE(bad.init_data_from_layout(local, off, 0)); -} - -TEST_F(SpscQueueTest, InitAcceptsPowerOfTwo) { - PTO2SpscQueue q{}; - DeviceArena local; - const size_t off4 = PTO2SpscQueue::reserve_layout(local, 4); - const size_t off1024 = PTO2SpscQueue::reserve_layout(local, 1024); - ASSERT_NE(local.commit(), nullptr); - EXPECT_TRUE(q.init_data_from_layout(local, off4, 4)); - q.destroy(); - EXPECT_TRUE(q.init_data_from_layout(local, off1024, 1024)); - q.destroy(); -} - -// ============================================================================= -// Basic push / pop -// ============================================================================= - -TEST_F(SpscQueueTest, PushPopSingle) { - EXPECT_TRUE(queue.push(&slots[0])); - EXPECT_EQ(queue.size(), 1u); - - PTO2TaskSlotState *out[1]; - int count = queue.pop_batch(out, 1); - ASSERT_EQ(count, 1); - EXPECT_EQ(out[0], &slots[0]); - EXPECT_EQ(queue.size(), 0u); -} - -TEST_F(SpscQueueTest, FIFOOrdering) { - for (int i = 0; i < 5; i++) { - ASSERT_TRUE(queue.push(&slots[i])); - } - - PTO2TaskSlotState *out[5]; - int count = queue.pop_batch(out, 5); - ASSERT_EQ(count, 5); - for (int i = 0; i < 5; i++) { - EXPECT_EQ(out[i], &slots[i]) << "FIFO order violated at i=" << i; - } -} - -TEST_F(SpscQueueTest, PopBatchPartial) { - for (int i = 0; i < 3; i++) { - queue.push(&slots[i]); - } - - // Request 5 but only 3 available - PTO2TaskSlotState *out[5]; - int count = queue.pop_batch(out, 5); - EXPECT_EQ(count, 3); -} - -TEST_F(SpscQueueTest, PopBatchEmpty) { - PTO2TaskSlotState *out[5]; - int count = queue.pop_batch(out, 5); - EXPECT_EQ(count, 0); -} - -// ============================================================================= -// Full detection -// ============================================================================= - -TEST_F(SpscQueueTest, FullReturnsFalse) { - // Usable capacity = CAPACITY - 1 = 15 - for (uint64_t i = 0; i < CAPACITY - 1; i++) { - ASSERT_TRUE(queue.push(&slots[i])) << "push failed at i=" << i; - } - EXPECT_EQ(queue.size(), CAPACITY - 1); - - // Queue full - EXPECT_FALSE(queue.push(&slots[CAPACITY - 1])) << "Push to full queue must return false"; -} - -TEST_F(SpscQueueTest, UsableCapacityIsCapacityMinusOne) { - int pushed = 0; - while (queue.push(&slots[pushed % 64])) { - pushed++; - if (pushed > 100) break; // safety - } - EXPECT_EQ(pushed, static_cast(CAPACITY - 1)); -} - -// ============================================================================= -// Full then recover -// ============================================================================= - -TEST_F(SpscQueueTest, FullThenPopThenPush) { - for (uint64_t i = 0; i < CAPACITY - 1; i++) { - queue.push(&slots[i]); - } - EXPECT_FALSE(queue.push(&slots[0])); - - // Pop one - PTO2TaskSlotState *out[1]; - int count = queue.pop_batch(out, 1); - ASSERT_EQ(count, 1); - - // Now push should succeed - EXPECT_TRUE(queue.push(&slots[0])); -} - -// ============================================================================= -// Wrap-around -// ============================================================================= - -TEST_F(SpscQueueTest, WrapAroundCorrectness) { - // Push-pop cycles to advance head/tail past capacity boundary - for (int cycle = 0; cycle < 100; cycle++) { - ASSERT_TRUE(queue.push(&slots[cycle % 64])) << "push failed at cycle=" << cycle; - PTO2TaskSlotState *out[1]; - int count = queue.pop_batch(out, 1); - ASSERT_EQ(count, 1) << "pop_batch failed at cycle=" << cycle; - EXPECT_EQ(out[0], &slots[cycle % 64]); - } - EXPECT_EQ(queue.size(), 0u); -} - -TEST_F(SpscQueueTest, WrapAroundBatchCorrectness) { - // Multiple cycles of batch push/pop across wrap boundary - for (int cycle = 0; cycle < 20; cycle++) { - int batch = 5; - for (int i = 0; i < batch; i++) { - ASSERT_TRUE(queue.push(&slots[(cycle * batch + i) % 64])); - } - PTO2TaskSlotState *out[5]; - int count = queue.pop_batch(out, batch); - ASSERT_EQ(count, batch); - for (int i = 0; i < batch; i++) { - EXPECT_EQ(out[i], &slots[(cycle * batch + i) % 64]); - } - } -} - -// ============================================================================= -// size() accuracy -// ============================================================================= - -TEST_F(SpscQueueTest, SizeTracksOperations) { - EXPECT_EQ(queue.size(), 0u); - - queue.push(&slots[0]); - EXPECT_EQ(queue.size(), 1u); - - queue.push(&slots[1]); - queue.push(&slots[2]); - EXPECT_EQ(queue.size(), 3u); - - PTO2TaskSlotState *out[2]; - queue.pop_batch(out, 2); - EXPECT_EQ(queue.size(), 1u); - - queue.pop_batch(out, 1); - EXPECT_EQ(queue.size(), 0u); -} - -// ============================================================================= -// Producer-consumer (two threads) -// ============================================================================= - -TEST_F(SpscQueueTest, TwoThreadProducerConsumer) { - constexpr int TOTAL = 10000; - std::vector consumed; - consumed.reserve(TOTAL); - - // Use a large pool of slot states for unique pointers - std::vector big_pool(TOTAL); - - std::thread producer([&]() { - for (int i = 0; i < TOTAL; i++) { - while (!queue.push(&big_pool[i])) { - // spin - } - } - }); - - std::thread consumer([&]() { - int total = 0; - PTO2TaskSlotState *out[32]; - while (total < TOTAL) { - int count = queue.pop_batch(out, 32); - for (int i = 0; i < count; i++) { - consumed.push_back(out[i]); - } - total += count; - } - }); - - producer.join(); - consumer.join(); - - ASSERT_EQ(consumed.size(), static_cast(TOTAL)); - // Verify FIFO order - for (int i = 0; i < TOTAL; i++) { - EXPECT_EQ(consumed[i], &big_pool[i]) << "FIFO violated at i=" << i; - } -} - -// ============================================================================= -// Cached index behavior -// ============================================================================= - -TEST_F(SpscQueueTest, CachedIndexLazyRefresh) { - // Fill queue - for (uint64_t i = 0; i < CAPACITY - 1; i++) { - queue.push(&slots[i]); - } - - // Consumer pops all - PTO2TaskSlotState *out[16]; - int count = queue.pop_batch(out, CAPACITY); - EXPECT_EQ(count, static_cast(CAPACITY - 1)); - - // Producer's tail_cached_ is stale (still thinks queue is full) - // Next push should refresh tail_cached_ and succeed - EXPECT_TRUE(queue.push(&slots[0])); -} - -TEST_F(SpscQueueTest, CachedIndexConsumerRefresh) { - // Consumer calls pop_batch on empty queue (head_cached_ is 0) - PTO2TaskSlotState *out[1]; - EXPECT_EQ(queue.pop_batch(out, 1), 0); - - // Producer pushes - queue.push(&slots[0]); - - // Consumer's head_cached_ is stale, pop_batch must refresh - int count = queue.pop_batch(out, 1); - EXPECT_EQ(count, 1); - EXPECT_EQ(out[0], &slots[0]); -} diff --git a/tests/ut/cpp/a2a3/test_wiring.cpp b/tests/ut/cpp/a2a3/test_wiring.cpp index 93e1d04e1..662c83bdc 100644 --- a/tests/ut/cpp/a2a3/test_wiring.cpp +++ b/tests/ut/cpp/a2a3/test_wiring.cpp @@ -394,70 +394,3 @@ TEST_F(WiringTest, AdvanceRingPointersResetsSlots) { EXPECT_EQ(slot.completed_subtasks.load(), 0); EXPECT_EQ(slot.fanout_head, nullptr); } - -// ============================================================================= -// drain_wiring_queue: pushes tasks through SPSC queue -// ============================================================================= - -TEST_F(WiringTest, DrainWiringQueueProcessesTasks) { - alignas(64) PTO2TaskSlotState task_slot; - alignas(64) PTO2TaskPayload payload; - memset(&payload, 0, sizeof(payload)); - PTO2TaskDescriptor desc{}; - - init_slot(task_slot, PTO2_TASK_PENDING, 0, 1); - payload.fanin_actual_count = 0; - task_slot.payload = &payload; - task_slot.task = &desc; - - // Push into wiring SPSC queue (orchestrator side) - ASSERT_TRUE(sched.wiring.queue.push(&task_slot)); - - // Drain (scheduler thread 0 side) - int wired = sched.drain_wiring_queue(true /* force_drain */); - EXPECT_EQ(wired, 1); - - // Task should be ready - PTO2ResourceShape shape = task_slot.active_mask.to_shape(); - auto *popped = sched.ready_queues[static_cast(shape)].pop(); - EXPECT_EQ(popped, &task_slot); -} - -TEST_F(WiringTest, DrainWiringQueueBackoffDefers) { - alignas(64) PTO2TaskSlotState task_slot; - alignas(64) PTO2TaskPayload payload; - memset(&payload, 0, sizeof(payload)); - PTO2TaskDescriptor desc{}; - - init_slot(task_slot, PTO2_TASK_PENDING, 0, 1); - payload.fanin_actual_count = 0; - task_slot.payload = &payload; - task_slot.task = &desc; - - sched.wiring.queue.push(&task_slot); - - // Without force_drain, single item < BATCH_SIZE → backoff - sched.wiring.backoff_counter = 0; - int wired = sched.drain_wiring_queue(false); - EXPECT_EQ(wired, 0) << "Backoff should defer when queue < BATCH_SIZE"; - EXPECT_EQ(sched.wiring.backoff_counter, 1); -} - -TEST_F(WiringTest, DrainWiringQueueBackoffLimitForcesProcess) { - alignas(64) PTO2TaskSlotState task_slot; - alignas(64) PTO2TaskPayload payload; - memset(&payload, 0, sizeof(payload)); - PTO2TaskDescriptor desc{}; - - init_slot(task_slot, PTO2_TASK_PENDING, 0, 1); - payload.fanin_actual_count = 0; - task_slot.payload = &payload; - task_slot.task = &desc; - - sched.wiring.queue.push(&task_slot); - - // Set backoff at limit → should process - sched.wiring.backoff_counter = PTO2SchedulerState::WiringState::BACKOFF_LIMIT; - int wired = sched.drain_wiring_queue(false); - EXPECT_EQ(wired, 1) << "Backoff limit reached should force processing"; -}