From 87d001de480974c33c2d8170cd4e48a6e48847e5 Mon Sep 17 00:00:00 2001 From: Chao Wang <26245345+ChaoWao@users.noreply.github.com> Date: Thu, 2 Jul 2026 09:56:27 +0800 Subject: [PATCH] refactor(tmr): remove PTO2LocalReadyBuffer local-first dispatch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The tensormap_and_ringbuffer scheduler carried a per-thread, per-CoreType thread-local ready buffer (PTO2LocalReadyBuffer) as a "local-first dispatch" fast path: a newly-ready consumer was try_push'd into the producing thread's local buffer (peer-invisible, zero-atomic) before falling back to the shared MPMC ready_queues[]. A spill/flush layer (capacity-gated overflow spill, flush_local_bufs, FlushGuard) existed only to stop a thread from hoarding work and starving peers. Remove it on both arches. The scheduler now always routes ready tasks straight to the shared ready_queues[] (DUMMY tasks still go to dummy_ready_queue via the existing push_ready_routed()). Correctness never depended on the local buffer — every consumer already had a nullptr/!try_push -> shared-queue fallback. Changes (a2a3 + a5, mirrored): - Delete struct PTO2LocalReadyBuffer, PTO2_LOCAL_DISPATCH_TYPE_NUM, LOCAL_READY_CAP_PER_TYPE, the local_bufs[] stack allocation. - Drop the PTO2LocalReadyBuffer param from release_fanin_and_check_ready, get_ready_tasks_batch, on_task_complete, pop_ready_tasks_batch, dispatch_shape, dispatch_ready_tasks, complete_slot_task, check_running_cores_for_completion, has_residual_mix, poll_and_complete (+ DrainCompletionSink field). - Delete the spill/flush machinery in dispatch_ready_tasks. - a2a3 speculative-release / early-dispatch machinery is untouched (orthogonal). C++ unit tests (tests/ut/cpp/{a2a3,a5}): delete the LocalReadyBufferTest suite in test_ready_queue.cpp and rewrite test_scheduler_state.cpp's GetReadyTasksBatch test to drain the shared queue (was: local-buffer-first); update the file/header comments. Profiling cleanup (platform + Python): drop local_depth_* from L2SwimlaneAicpuSchedPhaseRecord, the local_at_* params of l2_swimlane_aicpu_record_sched_phase, the host JSON local_at_* emit, and the swimlane_converter.py local_at_*/local_ready_buf handling. Shared-queue depth tracking is retained. Fix a latent kernel bug the timing change unmasked (examples/workers/l3/ ep_dispatch_combine): dispatch wrote recv_count_out with a raw scalar GM store and never flushed it to HBM, so the downstream local_expert task read it as 0 under the new dispatch timing -> zero rows processed -> all-zero output. Add a single-cache-line dcci after the recv_count_out write. Root cause + evidence in docs/investigations/2026-07-local-buffer-removal-ep-combine-regression.md. Also restore the tmr benchmark set in tools/benchmark_rounds.sh: #1177 silently dropped the 6 cases #1157 had just added, leaving only alternating_matmul_add. Restored benchmark_bgemm / paged_attention_unroll(+manual_scope) / batch_paged_attention / qwen3_14b_decode; spmd_paged_attention is commented out pending a pre-existing onboard stall (reproduces on baseline, unrelated). Verified on real a2a3 silicon (task-submit locked): - Sim: a2a3sim + a5sim tmr suites pass; ep_dispatch_combine sim passes. - Onboard: tmr suite 33 passed / 1 skipped; ep_dispatch_combine 3/3 pass after the dcci fix (was 3/3 fail); l2_swimlane dfx (incl. swimlane_converter smoke over the new JSON) passes. - C++ UT sources compile clean after the test updates. - Benchmark (Effective, orch∪sched window, 100 rounds, same session, same locked device): all measurable tmr cases improve -4% to -18% (qwen3_14b_decode -4.4%, batch_paged_attention -18%). Co-Authored-By: Claude Opus 4.8 (1M context) --- ...al-buffer-removal-ep-combine-regression.md | 157 +++++++++++ docs/investigations/README.md | 1 + docs/scheduler.md | 4 +- .../kernels/aiv/dispatch.cpp | 6 + simpler_setup/tools/swimlane_converter.py | 30 +-- .../aicpu/l2_swimlane_collector_aicpu.h | 14 +- .../include/common/l2_swimlane_profiling.h | 29 +-- .../aicpu/l2_swimlane_collector_aicpu.cpp | 6 +- .../shared/host/l2_swimlane_collector.cpp | 2 - .../runtime/pto_async_wait.h | 4 +- .../runtime/scheduler/pto_scheduler.h | 117 ++------- .../scheduler/scheduler_completion.cpp | 13 +- .../runtime/scheduler/scheduler_context.h | 33 +-- .../runtime/scheduler/scheduler_dispatch.cpp | 245 +++++------------- .../aicpu/l2_swimlane_collector_aicpu.h | 14 +- .../include/common/l2_swimlane_profiling.h | 29 +-- .../aicpu/l2_swimlane_collector_aicpu.cpp | 6 +- .../shared/host/l2_swimlane_collector.cpp | 2 - .../runtime/pto_async_wait.h | 4 +- .../runtime/scheduler/pto_scheduler.h | 117 ++------- .../scheduler/scheduler_completion.cpp | 13 +- .../runtime/scheduler/scheduler_context.h | 32 +-- .../runtime/scheduler/scheduler_dispatch.cpp | 185 +++++-------- tests/ut/cpp/a2a3/test_ready_queue.cpp | 77 +----- tests/ut/cpp/a2a3/test_scheduler_state.cpp | 22 +- tests/ut/cpp/a5/test_ready_queue.cpp | 77 +----- tests/ut/cpp/a5/test_scheduler_state.cpp | 22 +- tools/benchmark_rounds.sh | 14 + 28 files changed, 465 insertions(+), 810 deletions(-) create mode 100644 docs/investigations/2026-07-local-buffer-removal-ep-combine-regression.md diff --git a/docs/investigations/2026-07-local-buffer-removal-ep-combine-regression.md b/docs/investigations/2026-07-local-buffer-removal-ep-combine-regression.md new file mode 100644 index 0000000000..b47857f078 --- /dev/null +++ b/docs/investigations/2026-07-local-buffer-removal-ep-combine-regression.md @@ -0,0 +1,157 @@ +# 2026-07 — Removing PTO2LocalReadyBuffer exposed a missing dcci in EP dispatch + +## Status + +**RESOLVED** in PR #1245. Root cause was a latent kernel bug (dispatch never +flushed `recv_count_out` to HBM), unmasked by the dispatch-timing change from +removing `PTO2LocalReadyBuffer`. Fixed by a one-line `dcci` in the example +kernel; the local-buffer removal itself is correct. Onboard a2a3 3/3 pass after +the fix. See "Fix" below. + +## Symptom + +`examples/workers/l3/ep_dispatch_combine` (a2a3 onboard, `device_count=2`) +fails its golden check **deterministically** after the local-buffer removal: + +- baseline (local buffer present): onboard **3/3 PASS** +- current (local buffer removed): onboard **3/3 FAIL**, sim **PASS** + +No runtime-error signatures at all (zero 507018 / HandleTaskTimeout / deadlock / +stall in the device log). It is a **wrong result**, not a hang. + +## Isolation (what the diagnostics prove) + +The example verifies two stages; only the second fails: + +- **dispatch stage** (`_verify_recv_outputs`: recv_x / recv_w / recv_idx / + recv_count) — **passes** (prints nothing; it only prints on mismatch). +- **combine stage** (`_verify_routed_y`) — **fails**, and the chip's `routed_y` + is **all zero**: `got min=0 median=0 max=0`, `bad=524287/524288`, `rel=1.0`, + bad-d span = full [0,4095], bad rows = 128/128 on both chips. + +So dispatch (which does the same class of cross-rank TPUT/TNOTIFY) is fine; the +combine kernel produces an entirely empty result. + +## Hypothesis history + +### REJECTED: dispatch→combine pub_counts visibility gap + +Original theory: combine's push loop no-ops because `pub_counts` (written by +dispatch into window scratch, read by combine) reads as zero under the new +dispatch timing, leaving `routed_y_buf` untouched → all-zero `routed_y`. + +**Disproved by direct measurement.** A temporary probe summed the whole +`pub_counts` table combine reads and broadcast it into `routed_y[0]` (host- +visible); on current-onboard it read **non-zero** (chip0 sum surfaced as +`got max=20476.7`). So the dispatch→combine scratch dataflow IS visible — combine +sees correct counts. This edge is NOT the bug. + +### OPEN: root cause is in combine's push→reduce section + +With pub_counts confirmed visible, the all-zero `routed_y` must come from +downstream of that read, inside combine.cpp: + +1. **push phase** — the cross-rank `TPUT` of `recv_y` rows into the peer's + `routed_y_buf` never lands, or +2. **combine_done barrier** (TNOTIFY/TWAIT) — mis-synchronizes, or +3. **reduce phase** — reads `routed_y_buf` as zero. + +A second probe (sum `routed_y_buf` after the barrier, before reduce) would +disambiguate push-landed vs reduce-read, but the ad-hoc GM-scalar-loop probe +**failed to compile** on AICore (bisheng frontend error 70). Next approach: +runtime-side timing comparison (baseline vs current [STRACE] / PTO2_SCHED_PROFILING +of the combine task) rather than more in-kernel probes. + +### NARROWED: the empty result originates at local_expert, not combine + +Python-side probes (main.py, host-visible OUTPUT_EXISTING tensors, no kernel +recompile) on current-onboard show: + +| tensor | producer | state after run | +| ------ | -------- | --------------- | +| recv_x_out | dispatch | **correct** (nz≈3.2M, absmax 5472) | +| recv_w_out | dispatch | **correct** (nz=782) | +| recv_count_out | dispatch | **correct** (sum 782 / 754) | +| recv_y | local_expert | **ALL ZERO** (nz=0) | +| routed_y | combine | all zero (combine pushes recv_y's zeros) | + +So combine is a red herring — it correctly propagates an already-empty input. +The regression is at the **dispatch → local_expert** task edge: local_expert +had all three inputs correct in the *final* state, yet produced nothing. + +local_expert (`kernels/aiv/local_expert.cpp`) bounds its work by +`n_rows = recv_count[e]` and skips the row loop entirely when it reads 0. +**Leading hypothesis:** under the new dispatch timing, local_expert's AICore +reads `recv_count` (dispatch's HBM OUTPUT_EXISTING output, reused as +local_expert INPUT) as stale/zero → every expert does 0 rows → recv_y stays +zero. The host sees recv_count=782 afterwards because dispatch's write *did* +land — just not before local_expert consumed it. + +Two overlapping-sched-window facts, established via device_wall.sched markers: +the two ranks' combine tasks DO run concurrently (rules out "one rank exits +before the other's push"), consistent with the fault being upstream in +local_expert, not the cross-rank combine. + +A kernel sentinel probe to confirm "did local_expert read count=0" crashed the +AICore (507015 — bad UB slot addresses in the ad-hoc probe) and was reverted. + +### CONFIRMED ROOT CAUSE: dispatch never flushes recv_count_out to HBM + +A compile-safe probe settled it: forcing local_expert to process **all R rows** +(`n_rows = R`, ignoring `recv_count[e]`) makes the whole test **PASS** on +current-onboard. So recv_x / recv_w are visible and correct — the *only* wrong +input is `recv_count`, which local_expert reads as **0**. + +Why: in `dispatch.cpp`, `recv_count_out[e] = sum;` (line ~304) is a **raw scalar +GM store** from the AICore scalar unit, followed only by `pipe_barrier(PIPE_ALL)`. +`pipe_barrier` orders on-core pipelines but does **NOT** flush the cache line to +HBM — that needs a `dcci(..., CACHELINE_OUT)`. recv_x/recv_w/recv_idx are safe +because they go out through `TSTORE` (a real vector→GM write). `recv_count_out` +sits in cache; whether the downstream `local_expert` task's AICore sees it in +HBM depended on incidental timing. + +Removing `PTO2LocalReadyBuffer` changed AICPU dispatch timing enough that +local_expert now reads `recv_count_out` before dispatch's cached scalar store +lands in HBM → every expert runs 0 rows → recv_y all-zero → combine faithfully +pushes zeros → routed_y all-zero. Baseline timing happened to let the store +land first; sim has no cache so it never reproduced. + +This is a **latent kernel bug**, not a runtime regression: the local-buffer +removal is correct; it merely unmasked a missing `dcci` in the example kernel. + +## Fix + +In `examples/workers/l3/ep_dispatch_combine/kernels/aiv/dispatch.cpp`, after the +`recv_count_out[e] = sum;` loop, flush the written range to HBM with `dcci` +(pattern already used in `qwen3_14b_decode/fa_work_build.cpp` and +`deferred_notify_demo/kernel_producer.cpp`: `dcci(ptr, ENTIRE_DATA_CACHE, +CACHELINE_OUT)` / `dcci(ptr, SINGLE_CACHE_LINE, CACHELINE_OUT)`). The runtime's +completion-before-dispatch invariant then carries the flushed value to +local_expert correctly. Codegen-owned area (`examples/`); no runtime change. + +Robustness note: the same raw-scalar-store-without-dcci pattern should be +audited elsewhere in dispatch.cpp (any non-TSTORE GM write consumed by a later +task), since all of them were relying on the same incidental timing. + +## Superseded ambiguity (kept for history) + +- If the dispatch→combine **window-scratch visibility** is a guarantee the + runtime is supposed to provide across a task dependency edge, the real fix is + in the **runtime** (`src/{arch}/runtime/tensormap_and_ringbuffer/`) — restore + the cross-task visibility the local buffer was incidentally providing — and + the local-buffer removal is fine once that guarantee is explicit. This is the + Runtime-owned area. +- If combine is simply **missing a synchronization** it always needed (and only + passed before by timing luck), the fix is in the **kernel** + (`examples/.../combine.cpp`, Codegen-owned) — e.g. a barrier / acquire on + `pub_counts` before the push loop. + +Deciding requires the pub_counts dump above. Do not edit either area before the +dump confirms which edge actually fails. + +## Do NOT re-derive + +- sim passes; the bug is hardware-timing-only. Reproducing needs a2a3 onboard, + `device_count=2`, via `task-submit`. +- dispatch is not the culprit — it verifies clean. Focus on the dispatch→combine + window edge (`pub_counts`) and the combine push loop's `n == 0` early-out. diff --git a/docs/investigations/README.md b/docs/investigations/README.md index d8b1d6aa25..20eac6fb26 100644 --- a/docs/investigations/README.md +++ b/docs/investigations/README.md @@ -84,6 +84,7 @@ that ...". Newest first. +- [2026-07 — Removing PTO2LocalReadyBuffer exposed a missing dcci in EP dispatch](2026-07-local-buffer-removal-ep-combine-regression.md) — RESOLVED in #1245: local-buffer removal changed dispatch timing and unmasked a latent kernel bug (dispatch never dcci'd `recv_count_out` to HBM → local_expert read count=0 → all-zero output); fixed with a one-line dcci in the example kernel - [2026-06 — Gating the two residual profiling enable() calls on the orch/scheduler hot path](2026-06-orch-profiling-enable-gates-hot-path.md) — gated under existing `PTO2_PROFILING`; magnitude unmeasured, no new macro - [2026-06 — Replacing COND with GM+dcci for AICore→AICPU notification](2026-06-cond-vs-gm-notification.md) - [2026-06 — Letting AICore directly read or write the SPR MMIO window](2026-06-aicore-mmio-to-spr.md) diff --git a/docs/scheduler.md b/docs/scheduler.md index eb6e5fabdc..75b5a8458f 100644 --- a/docs/scheduler.md +++ b/docs/scheduler.md @@ -84,8 +84,8 @@ ReadyQueue ready_next_level_queue_; // WorkerType::NEXT_LEVEL tasks ReadyQueue ready_sub_queue_; // WorkerType::SUB tasks ``` -Matching L2's per-shape ready buffer (`PTO2_LocalReadyBuffer` fan-out to -AIC / AIV / MIX queues), with the L3+ exception that we use `std::queue` +Matching L2's per-shape ready queues (the shared MPMC `ready_queues[]` split +by AIC / AIV / MIX), with the L3+ exception that we use `std::queue` (Allowed Exception 3: dynamic data structures on host) and only two worker types (Allowed Exception 2: `NEXT_LEVEL` + `SUB` at L3+, not AIC / AIV / MIX). `Orchestrator::submit_*` routes each slot to the queue diff --git a/examples/workers/l3/ep_dispatch_combine/kernels/aiv/dispatch.cpp b/examples/workers/l3/ep_dispatch_combine/kernels/aiv/dispatch.cpp index a6828afd31..743fe9e9dc 100644 --- a/examples/workers/l3/ep_dispatch_combine/kernels/aiv/dispatch.cpp +++ b/examples/workers/l3/ep_dispatch_combine/kernels/aiv/dispatch.cpp @@ -303,6 +303,12 @@ extern "C" __aicore__ __attribute__((always_inline)) void kernel_entry(__gm__ in } recv_count_out[e] = sum; } + // recv_count_out is written by the scalar unit (raw GM store), not TSTORE, + // so it stays in cache; the downstream local_expert task reads it to bound + // its row loop. Flush the one cache line ([L=16,1] INT32 = 64 B) so its HBM + // value is visible regardless of dispatch/consumer timing. (recv_x/recv_w/ + // recv_idx go out via TSTORE and need no dcci.) + dcci((__gm__ void *)recv_count_out, SINGLE_CACHE_LINE, CACHELINE_OUT); // ------------------------------------------------------------------ // payload_push: push x / weight / idx payloads via TPUT. diff --git a/simpler_setup/tools/swimlane_converter.py b/simpler_setup/tools/swimlane_converter.py index 5e8fe7ccd6..0dd8022041 100644 --- a/simpler_setup/tools/swimlane_converter.py +++ b/simpler_setup/tools/swimlane_converter.py @@ -1549,17 +1549,11 @@ def _find_containing_complete(thread_idx: int, finish_us: float): # Queue-depth snapshot fields. Layout per # L2SwimlaneAicpuSchedPhaseRecord docstring: [AIC, AIV, MIX]. - local_at_start = record.get("local_at_start") - local_at_end = record.get("local_at_end") shared_at_start = record.get("shared_at_start") shared_at_end = record.get("shared_at_end") depths_valid = ( - isinstance(local_at_start, list) - and isinstance(local_at_end, list) - and isinstance(shared_at_start, list) + isinstance(shared_at_start, list) and isinstance(shared_at_end, list) - and len(local_at_start) == 3 - and len(local_at_end) == 3 and len(shared_at_start) == 3 and len(shared_at_end) == 3 ) @@ -1579,8 +1573,6 @@ def _find_containing_complete(thread_idx: int, finish_us: float): # parser-safe while still self-documenting. phase_args.update( { - f"T{thread_idx}_local_at_start (aic,aiv,mix)": list(local_at_start), - f"T{thread_idx}_local_at_end (aic,aiv,mix)": list(local_at_end), "shared_at_start (aic,aiv,mix)": list(shared_at_start), "shared_at_end (aic,aiv,mix)": list(shared_at_end), } @@ -1611,27 +1603,13 @@ def _find_containing_complete(thread_idx: int, finish_us: float): # start, so emitting both is redundant. Two samples at the # SAME ts (e.g. final-drain emit where start_time==end_time) # also breaks Perfetto's rate calc (divide-by-zero → NULL). - # Track name carries thread index so it reads standalone - # even with the thread tree collapsed. Only complete/dispatch - # carry real queue depths; release/resolve/early_dispatch zero- - # fill them, so skip their counter samples to avoid spurious 0 - # dips. + # Only complete/dispatch carry real queue depths; release/ + # resolve/early_dispatch zero-fill them, so skip their counter + # samples to avoid spurious 0 dips. if phase not in ("complete", "dispatch"): continue if not depths_valid: continue - local_track_name = f"local_ready_buf_T{thread_idx}" - events.append( - { - "args": {"AIC": local_at_end[0], "AIV": local_at_end[1], "MIX": local_at_end[2]}, - "cat": "queue", - "name": local_track_name, - "ph": "C", - "pid": 2, - "tid": tid, - "ts": end_us, - } - ) # Shared queue: dedicated tid 3999 so all 3 schedulers' # snapshots compose onto one timeline (it's the same global # queue regardless of who sampled it). Samples from different diff --git a/src/a2a3/platform/include/aicpu/l2_swimlane_collector_aicpu.h b/src/a2a3/platform/include/aicpu/l2_swimlane_collector_aicpu.h index 3182eb1c19..8c8b56d75b 100644 --- a/src/a2a3/platform/include/aicpu/l2_swimlane_collector_aicpu.h +++ b/src/a2a3/platform/include/aicpu/l2_swimlane_collector_aicpu.h @@ -166,11 +166,9 @@ void l2_swimlane_aicpu_init_phase(int worker_count, int num_sched_phase_threads, * pool. Silently drops records when the buffer is full or the pool was not * primed (init failed for this thread). * - * Queue-depth snapshots distinguish "task hidden in T0's local_buf" from - * "shared queue has it but peers spin on the wrong shape" — the former shows - * `local_depth > 0, shared_depth == 0` for the owning thread while peers see - * `shared_depth == 0` until overflow. Pass nullptr for any of the four arrays - * when not capturing (the record's corresponding slot is zero-filled). + * Queue-depth snapshots record the per-shape shared ready-queue occupancy at + * phase boundaries. Pass nullptr for either array when not capturing (the + * record's corresponding slot is zero-filled). * * @param thread_idx Scheduler thread index * @param kind Complete or Dispatch @@ -180,16 +178,12 @@ void l2_swimlane_aicpu_init_phase(int worker_count, int num_sched_phase_threads, * @param tasks_processed Tasks processed in this phase batch * @param pop_hit Dispatch delta since last emit (0 for Complete) * @param pop_miss Dispatch delta since last emit (0 for Complete) - * @param local_at_start Per-shape PTO2LocalReadyBuffer.count at phase start (size L2SWIMLANE_NUM_QUEUE_SHAPES; may be - * nullptr) * @param shared_at_start Per-shape sched.ready_queues[shape].size() at phase start (may be nullptr) - * @param local_at_end Per-shape PTO2LocalReadyBuffer.count at phase end (may be nullptr) * @param shared_at_end Per-shape sched.ready_queues[shape].size() at phase end (may be nullptr) */ void l2_swimlane_aicpu_record_sched_phase( int thread_idx, L2SwimlaneSchedPhaseKind kind, uint64_t start_time, uint64_t end_time, uint32_t loop_iter, - uint32_t tasks_processed, uint32_t pop_hit = 0, uint32_t pop_miss = 0, const int16_t *local_at_start = nullptr, - const int16_t *shared_at_start = nullptr, const int16_t *local_at_end = nullptr, + uint32_t tasks_processed, uint32_t pop_hit = 0, uint32_t pop_miss = 0, const int16_t *shared_at_start = nullptr, const int16_t *shared_at_end = nullptr ); diff --git a/src/a2a3/platform/include/common/l2_swimlane_profiling.h b/src/a2a3/platform/include/common/l2_swimlane_profiling.h index c5bd912610..b7a4fc354b 100644 --- a/src/a2a3/platform/include/common/l2_swimlane_profiling.h +++ b/src/a2a3/platform/include/common/l2_swimlane_profiling.h @@ -535,27 +535,22 @@ constexpr int L2SWIMLANE_NUM_QUEUE_SHAPES = 3; * (zero for Complete). Kept named, not "extra1"/"extra2", so the device-side * commit and the host-side JSON emit don't drift on which extra means which. * - * Queue-depth snapshots (local_depth_*, shared_depth_*) record the per-shape - * scheduler queue occupancy at phase boundaries. They surface the - * dep-release-then-discovery latency that head OH alone can't distinguish from - * register-write latency: a phase whose start sees `local_depth=N, shared=0` - * and end sees `local_depth=N-K` shows that K tasks were popped from this - * thread's private buffer (invisible to peer threads) — peers must spin until - * those tasks overflow into shared. Filled with 0 below SCHED_PHASES. + * Queue-depth snapshots (shared_depth_*) record the per-shape scheduler ready + * queue occupancy at phase boundaries. They surface the dep-release-then- + * discovery latency that head OH alone can't distinguish from register-write + * latency. Filled with 0 below SCHED_PHASES. */ struct L2SwimlaneAicpuSchedPhaseRecord { - uint64_t start_time; // Phase start timestamp - uint64_t end_time; // Phase end timestamp - uint32_t loop_iter; // Scheduler-loop iteration number on this thread - L2SwimlaneSchedPhaseKind kind; // see enum above - uint32_t tasks_processed; // Tasks processed in this phase batch - uint32_t pop_hit; // SCHED_DISPATCH delta since last emit (0 for Complete) - uint32_t pop_miss; // SCHED_DISPATCH delta since last emit (0 for Complete) - int16_t local_depth_at_start[L2SWIMLANE_NUM_QUEUE_SHAPES]; // this thread's PTO2LocalReadyBuffer.count - int16_t local_depth_at_end[L2SWIMLANE_NUM_QUEUE_SHAPES]; + uint64_t start_time; // Phase start timestamp + uint64_t end_time; // Phase end timestamp + uint32_t loop_iter; // Scheduler-loop iteration number on this thread + L2SwimlaneSchedPhaseKind kind; // see enum above + uint32_t tasks_processed; // Tasks processed in this phase batch + uint32_t pop_hit; // SCHED_DISPATCH delta since last emit (0 for Complete) + uint32_t pop_miss; // SCHED_DISPATCH delta since last emit (0 for Complete) int16_t shared_depth_at_start[L2SWIMLANE_NUM_QUEUE_SHAPES]; // sched->ready_queues[shape].size() int16_t shared_depth_at_end[L2SWIMLANE_NUM_QUEUE_SHAPES]; - uint32_t _pad; // 64B alignment padding + uint32_t _pad[4]; // 64B alignment padding }; static_assert(sizeof(L2SwimlaneAicpuSchedPhaseRecord) == 64, "L2SwimlaneAicpuSchedPhaseRecord layout drift"); diff --git a/src/a2a3/platform/shared/aicpu/l2_swimlane_collector_aicpu.cpp b/src/a2a3/platform/shared/aicpu/l2_swimlane_collector_aicpu.cpp index aafffa8ee4..cfb188065e 100644 --- a/src/a2a3/platform/shared/aicpu/l2_swimlane_collector_aicpu.cpp +++ b/src/a2a3/platform/shared/aicpu/l2_swimlane_collector_aicpu.cpp @@ -796,8 +796,8 @@ static Record *acquire_phase_slot( void l2_swimlane_aicpu_record_sched_phase( int thread_idx, L2SwimlaneSchedPhaseKind kind, uint64_t start_time, uint64_t end_time, uint32_t loop_iter, - uint32_t tasks_processed, uint32_t pop_hit, uint32_t pop_miss, const int16_t *local_at_start, - const int16_t *shared_at_start, const int16_t *local_at_end, const int16_t *shared_at_end + uint32_t tasks_processed, uint32_t pop_hit, uint32_t pop_miss, const int16_t *shared_at_start, + const int16_t *shared_at_end ) { if (!s_phase_initialized) return; auto *state = s_sched_phase_pools[thread_idx]; @@ -829,9 +829,7 @@ void l2_swimlane_aicpu_record_sched_phase( dst[i] = src[i]; } }; - copy_snapshot(record->local_depth_at_start, local_at_start); copy_snapshot(record->shared_depth_at_start, shared_at_start); - copy_snapshot(record->local_depth_at_end, local_at_end); copy_snapshot(record->shared_depth_at_end, shared_at_end); } diff --git a/src/a2a3/platform/shared/host/l2_swimlane_collector.cpp b/src/a2a3/platform/shared/host/l2_swimlane_collector.cpp index b0a9123b26..05e5f45938 100644 --- a/src/a2a3/platform/shared/host/l2_swimlane_collector.cpp +++ b/src/a2a3/platform/shared/host/l2_swimlane_collector.cpp @@ -846,9 +846,7 @@ int L2SwimlaneCollector::export_swimlane_json() { outfile << ", \"pop_hit\": " << pr.pop_hit << ", \"pop_miss\": " << pr.pop_miss; } // Queue-depth snapshots — [AIC, AIV, MIX] per L2SwimlaneAicpuSchedPhaseRecord docstring. - emit_depth_array("local_at_start", pr.local_depth_at_start); emit_depth_array("shared_at_start", pr.shared_depth_at_start); - emit_depth_array("local_at_end", pr.local_depth_at_end); emit_depth_array("shared_at_end", pr.shared_depth_at_end); outfile << "}"; first = false; diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_async_wait.h b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_async_wait.h index 42a9474187..d1da3ae892 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_async_wait.h +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/pto_async_wait.h @@ -24,7 +24,6 @@ #include "pto_runtime2_types.h" struct PTO2SchedulerState; -struct PTO2LocalReadyBuffer; struct CompletionStats; inline constexpr int32_t MAX_ASYNC_WAITS = 64; @@ -177,7 +176,6 @@ struct AsyncWaitList { // entries[]). struct DrainCompletionSink { PTO2SchedulerState *sched{nullptr}; - PTO2LocalReadyBuffer *local_bufs{nullptr}; PTO2TaskSlotState **deferred_release_slot_states{nullptr}; int32_t *deferred_release_count{nullptr}; int32_t deferred_release_capacity{0}; @@ -296,7 +294,7 @@ struct AsyncWaitList { template AsyncPollResult poll_and_complete( - AICoreCompletionMailbox *aicore_mailbox, PTO2SchedulerState *sched, PTO2LocalReadyBuffer *local_bufs, + AICoreCompletionMailbox *aicore_mailbox, PTO2SchedulerState *sched, PTO2TaskSlotState **deferred_release_slot_states, int32_t &deferred_release_count, int32_t deferred_release_capacity #if PTO2_SCHED_PROFILING 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 a37eb0d43d..3ed6058ce2 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 @@ -62,42 +62,6 @@ struct PTO2ReadyQueueSlot { PTO2TaskSlotState *slot_state; }; -/** - * Thread-local ready buffer for local-first dispatch optimization. - * - * Two buffers per scheduling thread, one per CoreType (AIC=0, AIV=1). - * Initialized once before the scheduling loop; must be empty at - * the start of each iteration (verified by always_assert). - * - * Phase 1 fills per-CoreType buffers via on_task_complete(). - * The dispatch stage drains them local-first via get_ready_tasks_batch, - * with any remaining tasks pushed to the global ready queue. - */ -// Number of CoreType values eligible for local dispatch (AIC=0, AIV=1) -static constexpr int PTO2_LOCAL_DISPATCH_TYPE_NUM = 2; - -struct PTO2LocalReadyBuffer { - PTO2TaskSlotState **slot_states = nullptr; - int count = 0; - int capacity = 0; - - void reset(PTO2TaskSlotState **buf, int cap) { - slot_states = buf; - count = 0; - capacity = cap; - } - - bool try_push(PTO2TaskSlotState *s) { - if (slot_states && count < capacity) { - slot_states[count++] = s; - return true; - } - return false; - } - - PTO2TaskSlotState *pop() { return (count > 0) ? slot_states[--count] : nullptr; } -}; - /** * Lock-free bounded MPMC queue (Dmitry Vyukov design) * @@ -834,9 +798,7 @@ struct PTO2SchedulerState { // 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[]. Used by paths that do not have a thread-local - // ready buffer (e.g. wiring). See push_ready_routed_local for the - // dispatch-time fast path. + // per-shape ready_queues[]. void push_ready_routed(PTO2TaskSlotState *slot_state) { PTO2ResourceShape shape = slot_state->active_mask.to_shape(); if (shape == PTO2ResourceShape::DUMMY) { @@ -1157,9 +1119,7 @@ struct PTO2SchedulerState { return slot_state.next_block_idx.load(std::memory_order_seq_cst) >= slot_state.logical_block_num; } - bool release_fanin_and_check_ready( - PTO2TaskSlotState &slot_state, PTO2LocalReadyBuffer *local_bufs = nullptr, SpecReleaseSink *sink = nullptr - ) { + bool release_fanin_and_check_ready(PTO2TaskSlotState &slot_state, SpecReleaseSink *sink = nullptr) { // Atomically increment fanin_refcount and check if all producers are done // ACQ_REL on fanin_refcount already synchronizes with the orchestrator's // init release, making fanin_count visible — plain load suffices. @@ -1169,16 +1129,7 @@ struct PTO2SchedulerState { // Speculative early-dispatch: pre-staged tasks are released by doorbell // here, skipping the ready-queue round-trip entirely. if (try_speculative_release(slot_state, sink)) return true; - // Local-first: try per-CoreType thread-local buffer before global queue - // Route by active_mask: AIC-containing tasks → buf[0], AIV-only → buf[1] - // DUMMY shape is out of range for local_bufs (sized PTO2_NUM_RESOURCE_SHAPES); - // dummy slots bypass the local fast path and go straight to dummy_ready_queue. - PTO2ResourceShape shape = slot_state.active_mask.to_shape(); - if (shape == PTO2ResourceShape::DUMMY) { - dummy_ready_queue.push(&slot_state); - } else if (!local_bufs || !local_bufs[static_cast(shape)].try_push(&slot_state)) { - ready_queues[static_cast(shape)].push(&slot_state); - } + push_ready_routed(&slot_state); return true; } return false; @@ -1186,8 +1137,7 @@ struct PTO2SchedulerState { #if PTO2_ORCH_PROFILING || PTO2_SCHED_PROFILING bool release_fanin_and_check_ready( - PTO2TaskSlotState &slot_state, uint64_t &atomic_count, uint64_t &push_wait, - PTO2LocalReadyBuffer *local_bufs = nullptr, SpecReleaseSink *sink = nullptr + PTO2TaskSlotState &slot_state, uint64_t &atomic_count, uint64_t &push_wait, SpecReleaseSink *sink = nullptr ) { int32_t new_refcount = slot_state.fanin_refcount.fetch_add(1, std::memory_order_acq_rel) + 1; atomic_count += 1; // fanin_refcount.fetch_add @@ -1196,14 +1146,13 @@ struct PTO2SchedulerState { // Speculative early-dispatch: pre-staged tasks are released by doorbell // here, skipping the ready-queue round-trip entirely. if (try_speculative_release(slot_state, sink)) return true; - // Local-first: try per-CoreType thread-local buffer before global queue. - // Dummy slots bypass local_bufs (out-of-range for PTO2_NUM_RESOURCE_SHAPES) - // and go straight to dummy_ready_queue; use the profiling-aware push so - // atomic_count / push_wait stay consistent with the non-dummy path. + // Dummy slots go to dummy_ready_queue; everything else to the per-shape + // ready_queues[]. Use the profiling-aware push so atomic_count / push_wait + // stay consistent with the non-dummy path. PTO2ResourceShape shape = slot_state.active_mask.to_shape(); if (shape == PTO2ResourceShape::DUMMY) { dummy_ready_queue.push(&slot_state, atomic_count, push_wait); - } else if (!local_bufs || !local_bufs[static_cast(shape)].try_push(&slot_state)) { + } else { ready_queues[static_cast(shape)].push(&slot_state, atomic_count, push_wait); } return true; @@ -1212,35 +1161,15 @@ struct PTO2SchedulerState { } #endif - int get_ready_tasks_batch( - PTO2ResourceShape shape, PTO2LocalReadyBuffer &local_buf, PTO2TaskSlotState **out, int max_count - ) { - int count = 0; - while (count < max_count && local_buf.count > 0) { - out[count++] = local_buf.slot_states[--local_buf.count]; - } - int remaining = max_count - count; - if (remaining > 0) { - count += ready_queues[static_cast(shape)].pop_batch(out + count, remaining); - } - return count; + int get_ready_tasks_batch(PTO2ResourceShape shape, PTO2TaskSlotState **out, int max_count) { + return ready_queues[static_cast(shape)].pop_batch(out, max_count); } #if PTO2_SCHED_PROFILING int get_ready_tasks_batch( - PTO2ResourceShape shape, PTO2LocalReadyBuffer &local_buf, PTO2TaskSlotState **out, int max_count, - uint64_t &atomic_count, uint64_t &wait_cycle + PTO2ResourceShape shape, PTO2TaskSlotState **out, int max_count, uint64_t &atomic_count, uint64_t &wait_cycle ) { - int count = 0; - while (count < max_count && local_buf.count > 0) { - out[count++] = local_buf.slot_states[--local_buf.count]; - } - int remaining = max_count - count; - if (remaining > 0) { - count += - ready_queues[static_cast(shape)].pop_batch(out + count, remaining, atomic_count, wait_cycle); - } - return count; + return ready_queues[static_cast(shape)].pop_batch(out, max_count, atomic_count, wait_cycle); } #endif @@ -1292,12 +1221,11 @@ struct PTO2SchedulerState { uint32_t #endif on_task_complete( - PTO2TaskSlotState &slot_state, + PTO2TaskSlotState &slot_state #if PTO2_SCHED_PROFILING - int thread_idx, + , + int thread_idx #endif - - PTO2LocalReadyBuffer *local_bufs = nullptr ) { #if PTO2_SCHED_PROFILING CompletionStats stats = {0, 0, 0, true}; @@ -1352,12 +1280,12 @@ struct PTO2SchedulerState { PTO2TaskSlotState &consumer_slot = *current->slot_state; #if PTO2_SCHED_PROFILING stats.fanout_edges++; - if (release_fanin_and_check_ready(consumer_slot, fanout_atomics, push_wait, local_bufs, &rel_sink)) { + if (release_fanin_and_check_ready(consumer_slot, fanout_atomics, push_wait, &rel_sink)) { stats.tasks_enqueued++; } #else consumer_walk_count++; - release_fanin_and_check_ready(consumer_slot, local_bufs, &rel_sink); + release_fanin_and_check_ready(consumer_slot, &rel_sink); #endif current = current->next; } @@ -1465,9 +1393,9 @@ AsyncWaitList::try_inline_complete_locked(AsyncWaitList::DrainCompletionSink &si // Return value (CompletionStats / consumer-walk count) discarded: // async-wait drain path has no Resolve swimlane bar attached. #if PTO2_SCHED_PROFILING - (void)sink.sched->on_task_complete(slot_state, sink.thread_idx, sink.local_bufs); + (void)sink.sched->on_task_complete(slot_state, sink.thread_idx); #else - (void)sink.sched->on_task_complete(slot_state, sink.local_bufs); + (void)sink.sched->on_task_complete(slot_state); #endif if (*sink.deferred_release_count >= sink.deferred_release_capacity) { while (*sink.deferred_release_count > 0) { @@ -1487,7 +1415,7 @@ AsyncWaitList::try_inline_complete_locked(AsyncWaitList::DrainCompletionSink &si template inline AsyncPollResult AsyncWaitList::poll_and_complete( - AICoreCompletionMailbox *aicore_mailbox, PTO2SchedulerState *sched, PTO2LocalReadyBuffer *local_bufs, + AICoreCompletionMailbox *aicore_mailbox, PTO2SchedulerState *sched, PTO2TaskSlotState **deferred_release_slot_states, int32_t &deferred_release_count, int32_t deferred_release_capacity #if PTO2_SCHED_PROFILING , @@ -1499,7 +1427,6 @@ inline AsyncPollResult AsyncWaitList::poll_and_complete( AsyncWaitList::DrainCompletionSink sink{}; sink.sched = sched; - sink.local_bufs = local_bufs; sink.deferred_release_slot_states = deferred_release_slot_states; sink.deferred_release_count = &deferred_release_count; sink.deferred_release_capacity = deferred_release_capacity; @@ -1547,9 +1474,9 @@ inline AsyncPollResult AsyncWaitList::poll_and_complete( // Return value (CompletionStats / consumer-walk count) discarded: // deferred-completion drain has no Resolve swimlane bar attached. #if PTO2_SCHED_PROFILING - (void)sched->on_task_complete(*entry.slot_state, thread_idx, local_bufs); + (void)sched->on_task_complete(*entry.slot_state, thread_idx); #else - (void)sched->on_task_complete(*entry.slot_state, local_bufs); + (void)sched->on_task_complete(*entry.slot_state); #endif // Drain deferred_release in place when the buffer fills — same // overflow-drain idiom used by complete_slot_task's inline path diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_completion.cpp b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_completion.cpp index 7745898656..4f0f40ff21 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_completion.cpp +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_completion.cpp @@ -85,7 +85,7 @@ SlotTransition SchedulerContext::decide_slot_transition( void SchedulerContext::complete_slot_task( PTO2TaskSlotState &slot_state, int32_t expected_reg_task_id, [[maybe_unused]] PTO2SubtaskSlot subslot, int32_t thread_idx, int32_t core_id, Handshake *hank, int32_t &completed_this_turn, - PTO2TaskSlotState *deferred_release_slot_states[], int32_t &deferred_release_count, PTO2LocalReadyBuffer *local_bufs + PTO2TaskSlotState *deferred_release_slot_states[], int32_t &deferred_release_count #if PTO2_PROFILING , uint64_t dispatch_ts, uint64_t finish_ts @@ -200,9 +200,9 @@ void SchedulerContext::complete_slot_task( // counter side-effects (g_sched_*_atomic_count[thread_idx], consumed // by the otc_* log lines). It returns CompletionStats whose // `fanout_edges` is the consumer-walk count. - consumers_resolved = sched_->on_task_complete(slot_state, thread_idx, local_bufs).fanout_edges; + consumers_resolved = sched_->on_task_complete(slot_state, thread_idx).fanout_edges; #else - consumers_resolved = sched_->on_task_complete(slot_state, local_bufs); + consumers_resolved = sched_->on_task_complete(slot_state); #endif #if PTO2_PROFILING if (resolve_t0 != 0) { @@ -294,8 +294,7 @@ void SchedulerContext::clear_running_slot(CoreExecState &core) { void SchedulerContext::check_running_cores_for_completion( int32_t thread_idx, Handshake *hank, int32_t &completed_this_turn, int32_t &cur_thread_completed, - bool &made_progress, PTO2TaskSlotState *deferred_release_slot_states[], int32_t &deferred_release_count, - PTO2LocalReadyBuffer *local_bufs + bool &made_progress, PTO2TaskSlotState *deferred_release_slot_states[], int32_t &deferred_release_count ) { #if PTO2_SCHED_PROFILING auto &l2_swimlane = sched_l2_swimlane_[thread_idx]; @@ -377,7 +376,7 @@ void SchedulerContext::check_running_cores_for_completion( if (t.pending_done) { complete_slot_task( *core.pending_slot_state, core.pending_reg_task_id, core.pending_subslot, thread_idx, core_id, hank, - completed_this_turn, deferred_release_slot_states, deferred_release_count, local_bufs + completed_this_turn, deferred_release_slot_states, deferred_release_count #if PTO2_PROFILING , core.pending_dispatch_timestamp, finish_ts @@ -388,7 +387,7 @@ void SchedulerContext::check_running_cores_for_completion( if (t.running_done) { complete_slot_task( *core.running_slot_state, core.running_reg_task_id, core.running_subslot, thread_idx, core_id, hank, - completed_this_turn, deferred_release_slot_states, deferred_release_count, local_bufs + completed_this_turn, deferred_release_slot_states, deferred_release_count #if PTO2_PROFILING , core.running_dispatch_timestamp, finish_ts 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 02962864d1..9ce98ac5fa 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 @@ -208,10 +208,7 @@ class SchedulerContext { return "?"; } - int pop_ready_tasks_batch( - PTO2ResourceShape shape, int32_t thread_idx, PTO2LocalReadyBuffer &local_buf, PTO2TaskSlotState **out, - int max_count - ); + int pop_ready_tasks_batch(PTO2ResourceShape shape, int32_t thread_idx, PTO2TaskSlotState **out, int max_count); void build_payload( PTO2DispatchPayload &dispatch_payload, PTO2TaskSlotState &slot_state, PTO2SubtaskSlot subslot, @@ -254,8 +251,8 @@ class SchedulerContext { ); void dispatch_shape( - int32_t thread_idx, PTO2ResourceShape shape, CoreTracker::DispatchPhase phase, PTO2LocalReadyBuffer &local_buf, - CoreTracker &tracker, bool &entered_drain, bool &made_progress, bool &try_pushed + int32_t thread_idx, PTO2ResourceShape shape, CoreTracker::DispatchPhase phase, CoreTracker &tracker, + bool &entered_drain, bool &made_progress, bool &try_pushed ); // Speculative early-dispatch (Hook 1). After normal dispatch leaves idle @@ -276,10 +273,10 @@ class SchedulerContext { ); // One pass of "Phase 4" in the resolve_and_dispatch loop: IDLE-stage dispatch - // for MIX then (if no mix residual) AIC/AIV; mid-flush of local buffers; then - // PENDING-stage dispatch with cross-thread idle gating. MIX is strictly - // prioritized — when mix residual is detected after MIX-IDLE, AIC/AIV are - // skipped for the whole pass but MIX-PENDING still runs. + // for MIX then (if no mix residual) AIC/AIV; then PENDING-stage dispatch with + // cross-thread idle gating. MIX is strictly prioritized — when mix residual is + // detected after MIX-IDLE, AIC/AIV are skipped for the whole pass but + // MIX-PENDING still runs. // // Forward-progress argument for AIC/AIV: skip_aic_aiv is sticky for the // current pass only. The next loop iteration re-evaluates after Phase 1 @@ -288,8 +285,7 @@ class SchedulerContext { // not unbounded — once mix completes on at least one cluster, the next // pass either drains the residual or admits AIC/AIV. void dispatch_ready_tasks( - int32_t thread_idx, CoreTracker &tracker, PTO2LocalReadyBuffer (&local_bufs)[PTO2_NUM_RESOURCE_SHAPES], - bool pmu_active, bool &made_progress, bool &try_pushed + int32_t thread_idx, CoreTracker &tracker, bool pmu_active, bool &made_progress, bool &try_pushed ); // Returns true if any *other* scheduler thread currently has an idle core @@ -298,15 +294,14 @@ class SchedulerContext { // rationale and the safety argument against the drain worker. bool has_idle_in_other_threads(int32_t self_thread_idx, PTO2ResourceShape shape) const; - // True if mix tasks remain anywhere this thread could see them: the caller's - // MIX local LIFO stack or the global MIX ready queue. Approximate — + // 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 // extra/missed AIC/AIV skip and self-corrects on the next loop iteration. - bool has_residual_mix(const PTO2LocalReadyBuffer &mix_local_buf) const { - return mix_local_buf.count > 0 || sched_->ready_queues[static_cast(PTO2ResourceShape::MIX)].size() > 0; + bool has_residual_mix() const { + return sched_->ready_queues[static_cast(PTO2ResourceShape::MIX)].size() > 0; } // ========================================================================= @@ -320,8 +315,7 @@ class SchedulerContext { void complete_slot_task( PTO2TaskSlotState &slot_state, int32_t expected_reg_task_id, PTO2SubtaskSlot subslot, int32_t thread_idx, int32_t core_id, Handshake *hank, int32_t &completed_this_turn, - PTO2TaskSlotState *deferred_release_slot_states[], int32_t &deferred_release_count, - PTO2LocalReadyBuffer *local_bufs + PTO2TaskSlotState *deferred_release_slot_states[], int32_t &deferred_release_count #if PTO2_PROFILING , uint64_t dispatch_ts, uint64_t finish_ts @@ -333,8 +327,7 @@ class SchedulerContext { void check_running_cores_for_completion( int32_t thread_idx, Handshake *hank, int32_t &completed_this_turn, int32_t &cur_thread_completed, - bool &made_progress, PTO2TaskSlotState *deferred_release_slot_states[], int32_t &deferred_release_count, - PTO2LocalReadyBuffer *local_bufs + bool &made_progress, PTO2TaskSlotState *deferred_release_slot_states[], int32_t &deferred_release_count ); bool enter_drain_mode(PTO2TaskSlotState *slot_state, int32_t block_num); 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 c4a10369d2..b7e3fa0f7a 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 @@ -84,7 +84,7 @@ bool SchedulerContext::has_idle_in_other_threads(int32_t self_thread_idx, PTO2Re } int SchedulerContext::pop_ready_tasks_batch( - PTO2ResourceShape shape, int32_t thread_idx, PTO2LocalReadyBuffer &local_buf, PTO2TaskSlotState **out, int max_count + PTO2ResourceShape shape, int32_t thread_idx, PTO2TaskSlotState **out, int max_count ) { #if PTO2_PROFILING auto &l2_swimlane = sched_l2_swimlane_[thread_idx]; @@ -92,11 +92,11 @@ int SchedulerContext::pop_ready_tasks_batch( extern uint64_t g_sched_pop_atomic_count[], g_sched_pop_wait_cycle[]; uint64_t t_pop_start = get_sys_cnt_aicpu(); int count = sched_->get_ready_tasks_batch( - shape, local_buf, out, max_count, g_sched_pop_atomic_count[thread_idx], g_sched_pop_wait_cycle[thread_idx] + shape, out, max_count, g_sched_pop_atomic_count[thread_idx], g_sched_pop_wait_cycle[thread_idx] ); l2_swimlane.sched_dispatch_pop_cycle += (get_sys_cnt_aicpu() - t_pop_start); #else - int count = sched_->get_ready_tasks_batch(shape, local_buf, out, max_count); + int count = sched_->get_ready_tasks_batch(shape, out, max_count); #endif if (l2_swimlane_level_ >= L2SwimlaneLevel::SCHED_PHASES) { if (count > 0) { @@ -107,7 +107,7 @@ int SchedulerContext::pop_ready_tasks_batch( } #else (void)thread_idx; - int count = sched_->get_ready_tasks_batch(shape, local_buf, out, max_count); + int count = sched_->get_ready_tasks_batch(shape, out, max_count); #endif return count; } @@ -271,8 +271,8 @@ int SchedulerContext::prepare_block_for_dispatch( } void SchedulerContext::dispatch_shape( - int32_t thread_idx, PTO2ResourceShape shape, CoreTracker::DispatchPhase phase, PTO2LocalReadyBuffer &local_buf, - CoreTracker &tracker, bool &entered_drain, bool &made_progress, bool &try_pushed + int32_t thread_idx, PTO2ResourceShape shape, CoreTracker::DispatchPhase phase, CoreTracker &tracker, + bool &entered_drain, bool &made_progress, bool &try_pushed ) { #if PTO2_SCHED_PROFILING auto &l2_swimlane = sched_l2_swimlane_[thread_idx]; @@ -287,7 +287,7 @@ void SchedulerContext::dispatch_shape( while (cores.has_value() && !entered_drain) { int want = cores.count(); PTO2TaskSlotState *batch[CoreTracker::MAX_CLUSTERS * 3]; - int got = pop_ready_tasks_batch(shape, thread_idx, local_buf, batch, want); + int got = pop_ready_tasks_batch(shape, thread_idx, batch, want); if (got == 0) break; // sync_start exclusion gate. @@ -469,11 +469,9 @@ void SchedulerContext::dispatch_shape( } void SchedulerContext::dispatch_ready_tasks( - int32_t thread_idx, CoreTracker &tracker, PTO2LocalReadyBuffer (&local_bufs)[PTO2_NUM_RESOURCE_SHAPES], - bool pmu_active, bool &made_progress, bool &try_pushed + int32_t thread_idx, CoreTracker &tracker, bool pmu_active, bool &made_progress, bool &try_pushed ) { using Phase = CoreTracker::DispatchPhase; - constexpr int32_t MIX_I = static_cast(PTO2ResourceShape::MIX); // MIX is handled explicitly at the top of each stage; only AIC/AIV cycle // through this 2-elem array, with order toggled by thread parity for @@ -484,124 +482,49 @@ void SchedulerContext::dispatch_ready_tasks( }; const PTO2ResourceShape *aic_aiv = kAicAivOrder[thread_idx & 1]; - // Spill overflow from local_bufs to the shared ready queue BEFORE we start - // dispatching. release_fanin's fast path packs all newly-ready consumers - // into the producing thread's local_bufs (zero atomic, peer-invisible). For - // batch releases (e.g. attn_fence → 50 out_proj consumers) that - // overshoots this thread's slot budget so peers are starving while we - // hoard. The cross-thread invisibility window between "complete pushes 50 - // to local" and "IDLE-AIC's mid-phase flush exposes overflow to shared" - // is what shows up in the swimlane as the multi-microsecond inter-thread - // stagger on out_proj's first wave. - // - // Gate conditions: - // (a) local count exceeds this thread's per-shape block budget — we - // can't dispatch them all even with both RUNNING+PENDING slots; - // (b) at least one peer has idle cores in this shape — they want work. - // Both must hold to avoid wasting a CAS push when we could profitably - // self-dispatch the overflow. Condition (b) reads peer CoreTracker - // (plain 8-byte load on a rarely-contended cache line, ~5 ns) — we - // deliberately avoid ready_queues[s].size() here, which is two atomic - // loads on lines pushers + poppers actively bounce. - // - // Capacity derives from how cores are partitioned across sched threads: - // per-shape budget = (PLATFORM_MAX_BLOCKDIM / active_sched_threads_) - // × cores_per_blockdim_for_that_shape - // MIX is 1 cluster per block dim, so its budget equals the block-dim - // share without multiplying. - // - // Push the trailing `excess` slot pointers — O(1) count decrement, no - // memmove. push_batch is one CAS for the whole excess; peers see the - // batch immediately and can race for them. - const int32_t bd_per_thread = PLATFORM_MAX_BLOCKDIM / active_sched_threads_; - const int32_t thread_capacity[PTO2_NUM_RESOURCE_SHAPES] = { - /*AIC=*/bd_per_thread * PLATFORM_AIC_CORES_PER_BLOCKDIM, - /*AIV=*/bd_per_thread * PLATFORM_AIV_CORES_PER_BLOCKDIM, - /*MIX=*/bd_per_thread, - }; - for (int32_t s = 0; s < PTO2_NUM_RESOURCE_SHAPES; s++) { - auto &lb = local_bufs[s]; - int32_t excess = lb.count - thread_capacity[s]; - if (excess <= 0) continue; - if (!has_idle_in_other_threads(thread_idx, static_cast(s))) continue; - sched_->ready_queues[s].push_batch(&lb.slot_states[lb.count - excess], excess); - lb.count -= excess; - } - - auto flush_local_bufs = [&]() { - for (int32_t s = 0; s < PTO2_NUM_RESOURCE_SHAPES; s++) { - auto &lb = local_bufs[s]; - if (lb.count > 0) { - sched_->ready_queues[s].push_batch(lb.slot_states, lb.count); - lb.count = 0; - } - } - }; - // Every return path below must flush; wrap in RAII so we cannot forget. - // The mid-function flush between IDLE and PENDING is still called - // explicitly — guard only covers exit. - struct FlushGuard { - decltype(flush_local_bufs) &flush_fn; - ~FlushGuard() { flush_fn(); } - } flush_guard{flush_local_bufs}; - bool entered_drain = false; // ===== IDLE stage ===== - dispatch_shape( - thread_idx, PTO2ResourceShape::MIX, Phase::IDLE, local_bufs[MIX_I], tracker, entered_drain, made_progress, - try_pushed - ); + dispatch_shape(thread_idx, PTO2ResourceShape::MIX, Phase::IDLE, tracker, entered_drain, made_progress, try_pushed); if (entered_drain) return; // MIX-IDLE residual: AIC/AIV (both IDLE and PENDING) yield for this pass. // MIX-PENDING below still runs — that is the core of "mix strict priority": // pending slots are spent on mix before AIC/AIV get any chance. - bool skip_aic_aiv = has_residual_mix(local_bufs[MIX_I]); + bool skip_aic_aiv = has_residual_mix(); if (!skip_aic_aiv) { for (int i = 0; i < 2; i++) { PTO2ResourceShape s = aic_aiv[i]; - dispatch_shape( - thread_idx, s, Phase::IDLE, local_bufs[static_cast(s)], tracker, entered_drain, made_progress, - try_pushed - ); + dispatch_shape(thread_idx, s, Phase::IDLE, tracker, entered_drain, made_progress, try_pushed); if (entered_drain) return; } } - // Flush between IDLE and PENDING so PENDING-stage queue-size checks and any - // peer-thread reads see the IDLE-stage release_fanin output. - flush_local_bufs(); - if (pmu_active) return; // ===== PENDING stage ===== // MIX-PENDING gate: skip when a peer has an idle MIX-capable cluster — that // peer's next IDLE-MIX iteration will pull the mix task from the global - // queue (already flushed above) at lower latency than us pre-loading a - // pending slot here. Forward progress for MIX is preserved: at least one - // thread will run MIX-IDLE next pass and consume the residual. + // queue at lower latency than us pre-loading a pending slot here. Forward + // progress for MIX is preserved: at least one thread will run MIX-IDLE next + // pass and consume the residual. // // The gate is NOT subject to skip_aic_aiv — residual mix continues to drain // via pending slots on this thread when no peer is idle. if (!has_idle_in_other_threads(thread_idx, PTO2ResourceShape::MIX)) { dispatch_shape( - thread_idx, PTO2ResourceShape::MIX, Phase::PENDING, local_bufs[MIX_I], tracker, entered_drain, - made_progress, try_pushed + thread_idx, PTO2ResourceShape::MIX, Phase::PENDING, tracker, entered_drain, made_progress, try_pushed ); if (entered_drain) return; } // Re-check after MIX-PENDING. If MIX-IDLE already set skip_aic_aiv, leave // it set; otherwise, escalate iff PENDING-MIX left residual. - if (!skip_aic_aiv && has_residual_mix(local_bufs[MIX_I])) { + if (!skip_aic_aiv && has_residual_mix()) { skip_aic_aiv = true; } - // PENDING-MIX may have re-populated AIC/AIV local_bufs via release_fanin - // during in-flight completions; flush_guard ensures these don't carry - // across to the next iteration's IDLE stage. if (skip_aic_aiv) return; // AIC/AIV-PENDING gate: a peer-idle skip is a delay, not a loss — the peer @@ -609,10 +532,7 @@ void SchedulerContext::dispatch_ready_tasks( for (int i = 0; i < 2; i++) { PTO2ResourceShape s = aic_aiv[i]; if (has_idle_in_other_threads(thread_idx, s)) continue; - dispatch_shape( - thread_idx, s, Phase::PENDING, local_bufs[static_cast(s)], tracker, entered_drain, made_progress, - try_pushed - ); + dispatch_shape(thread_idx, s, Phase::PENDING, tracker, entered_drain, made_progress, try_pushed); if (entered_drain) return; } } @@ -795,12 +715,6 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ l2_swimlane.l2_swimlane_enabled = (l2_swimlane_level_ != L2SwimlaneLevel::DISABLED); #endif - constexpr int LOCAL_READY_CAP_PER_TYPE = 64; - PTO2TaskSlotState *local_ptrs[PTO2_NUM_RESOURCE_SHAPES][LOCAL_READY_CAP_PER_TYPE]; - PTO2LocalReadyBuffer local_bufs[PTO2_NUM_RESOURCE_SHAPES]; - for (int32_t i = 0; i < PTO2_NUM_RESOURCE_SHAPES; i++) { - local_bufs[i].reset(local_ptrs[i], LOCAL_READY_CAP_PER_TYPE); - } PTO2TaskSlotState *deferred_release_slot_states[PTO2_DEFERRED_RELEASE_CAP]; int32_t deferred_release_count = 0; @@ -822,34 +736,27 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ #if PTO2_PROFILING // Queue-depth snapshot carried across the iteration boundary: each phase - // emit consumes (phase_start_*) and refreshes them with its own end snapshot - // so the next phase's "at_start" equals the previous phase's "at_end". + // emit consumes (phase_start_shared) and refreshes it with its own end + // snapshot so the next phase's "at_start" equals the previous phase's + // "at_end". // // L2SWIMLANE_NUM_QUEUE_SHAPES (3) matches PTO2_NUM_RESOURCE_SHAPES: AIC/AIV/MIX. // - // **Hot-path cost discipline.** Local depth (this thread's PTO2LocalReadyBuffer) - // is a single int read on a register-cached stack — free. Shared depth - // (PTO2ReadyQueue::size) is two atomic relaxed loads against cache lines - // that all peer sched threads also write to (enqueue_pos and dequeue_pos - // bounce on every flush_local_bufs + every pop). With both phases emitting - // per iter that's 12 cross-core loads × thousands of iters per run, a - // measurable AICPU slowdown. Mitigation: lazy + per-iter cached shared - // snapshot, refreshed at most once per iteration. The complete-emit and - // dispatch-emit in the same iter both reuse the same shared sample; the - // big transitions (local→shared flush) still show up across iter boundaries. + // **Hot-path cost discipline.** Shared depth (PTO2ReadyQueue::size) is two + // atomic relaxed loads against cache lines that all peer sched threads also + // write to (enqueue_pos and dequeue_pos bounce on every push + every pop). + // With both phases emitting per iter that's cross-core loads × thousands of + // iters per run, a measurable AICPU slowdown. Mitigation: lazy + per-iter + // cached shared snapshot, refreshed at most once per iteration. The + // complete-emit and dispatch-emit in the same iter both reuse the same + // shared sample. static_assert( L2SWIMLANE_NUM_QUEUE_SHAPES == PTO2_NUM_RESOURCE_SHAPES, "queue snapshot width must match runtime resource shape count" ); - int16_t phase_start_local[L2SWIMLANE_NUM_QUEUE_SHAPES] = {0}; int16_t phase_start_shared[L2SWIMLANE_NUM_QUEUE_SHAPES] = {0}; int16_t iter_shared_snapshot[L2SWIMLANE_NUM_QUEUE_SHAPES] = {0}; bool iter_shared_sampled = false; - auto capture_local_snapshot = [&](int16_t local_out[L2SWIMLANE_NUM_QUEUE_SHAPES]) { - for (int s = 0; s < L2SWIMLANE_NUM_QUEUE_SHAPES; s++) { - local_out[s] = static_cast(local_bufs[s].count); - } - }; auto get_or_sample_shared = [&]() -> const int16_t * { if (!iter_shared_sampled) { // Clamp to int16_t max before narrowing. PTO2_PROF_READYQUEUE_SIZE @@ -865,15 +772,22 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ } return iter_shared_snapshot; }; - auto capture_phase_end = [&](int16_t local_out[L2SWIMLANE_NUM_QUEUE_SHAPES], - int16_t shared_out[L2SWIMLANE_NUM_QUEUE_SHAPES]) { - capture_local_snapshot(local_out); + auto capture_phase_end = [&](int16_t shared_out[L2SWIMLANE_NUM_QUEUE_SHAPES]) { const int16_t *shared_cached = get_or_sample_shared(); for (int s = 0; s < L2SWIMLANE_NUM_QUEUE_SHAPES; s++) shared_out[s] = shared_cached[s]; }; + // Queue-mutating phases (Complete / Wire / Dummy) push newly-ready consumers + // straight into the shared ready_queues[] (the local-first buffer is gone), + // so their end-of-phase shared depth differs from their start. Force a fresh + // re-sample for those emits — this also refreshes the per-iter cache so the + // next phase's start snapshot is not stale. + auto capture_phase_end_fresh = [&](int16_t shared_out[L2SWIMLANE_NUM_QUEUE_SHAPES]) { + iter_shared_sampled = false; + capture_phase_end(shared_out); + }; if (l2_swimlane_level_ >= L2SwimlaneLevel::SCHED_PHASES) { - capture_phase_end(phase_start_local, phase_start_shared); + capture_phase_end(phase_start_shared); } #endif @@ -930,7 +844,7 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ if (try_completed) { check_running_cores_for_completion( thread_idx, hank, completed_this_turn, cur_thread_completed, made_progress, - deferred_release_slot_states, deferred_release_count, local_bufs + deferred_release_slot_states, deferred_release_count ); } if (completed_this_turn > 0) { @@ -954,7 +868,7 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ if (rt_ != nullptr && rt_->aicore_mailbox != nullptr && (sched_->async_wait_list.count > 0 || rt_->aicore_mailbox->has_pending())) { AsyncPollResult poll_result = sched_->async_wait_list.poll_and_complete( - rt_->aicore_mailbox, sched_, local_bufs, deferred_release_slot_states, deferred_release_count, + rt_->aicore_mailbox, sched_, deferred_release_slot_states, deferred_release_count, PTO2_DEFERRED_RELEASE_CAP #if PTO2_SCHED_PROFILING , @@ -991,24 +905,17 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ // iteration; on a pure-retire iteration phase_complete_count is 0). if (l2_swimlane_level_ >= L2SwimlaneLevel::SCHED_PHASES && (l2_swimlane.phase_complete_count > 0 || l2_swimlane.phase_subretire_count > 0)) { - // Local depth is cheap (this thread's own buffer counter). - // Shared depth is NOT sampled here: complete's release_fanin - // pushes to local_bufs in the fast path (try_push succeeds - // until cap=64). Shared only changes on dispatch's flush - // path. Carrying phase_start_shared forward as end_shared - // is the right answer 99% of the time AND skips three - // contended atomic loads per emit. - int16_t phase_end_local[L2SWIMLANE_NUM_QUEUE_SHAPES]; - capture_local_snapshot(phase_end_local); + // Complete's release_fanin pushes newly-ready consumers into the + // shared ready_queues[], so the end depth differs from the start. + 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::Complete, _t0_phase, _t1, l2_swimlane.sched_loop_count, l2_swimlane.phase_complete_count + l2_swimlane.phase_subretire_count, /*pop_hit=*/0, - /*pop_miss=*/0, phase_start_local, phase_start_shared, phase_end_local, phase_start_shared + /*pop_miss=*/0, phase_start_shared, phase_end_shared ); - for (int s = 0; s < L2SWIMLANE_NUM_QUEUE_SHAPES; s++) { - phase_start_local[s] = phase_end_local[s]; - // phase_start_shared unchanged — carried forward - } + for (int s = 0; s < L2SWIMLANE_NUM_QUEUE_SHAPES; s++) + phase_start_shared[s] = phase_end_shared[s]; _t0_phase = _t1; l2_swimlane.phase_complete_count = 0; l2_swimlane.phase_subretire_count = 0; @@ -1042,16 +949,14 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ // 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_local[L2SWIMLANE_NUM_QUEUE_SHAPES]; - capture_local_snapshot(phase_end_local); + 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_local, phase_start_shared, - phase_end_local, phase_start_shared + 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_local[s] = phase_end_local[s]; - } + for (int s = 0; s < L2SWIMLANE_NUM_QUEUE_SHAPES; s++) + phase_start_shared[s] = phase_end_shared[s]; _t0_phase = _t1; } #endif @@ -1108,9 +1013,9 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ // OFF and the Resolve emit below is excluded. [[maybe_unused]] uint32_t dummy_consumers = 0; #if PTO2_SCHED_PROFILING - dummy_consumers = sched_->on_task_complete(dummy_slot, thread_idx, local_bufs).fanout_edges; + dummy_consumers = sched_->on_task_complete(dummy_slot, thread_idx).fanout_edges; #else - dummy_consumers = sched_->on_task_complete(dummy_slot, local_bufs); + dummy_consumers = sched_->on_task_complete(dummy_slot); #endif #if PTO2_PROFILING if (dummy_resolve_t0 != 0) { @@ -1153,17 +1058,16 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ // following Dispatch / EarlyDispatch / second-Complete bars start // at this end. if (dummy_outer_t0 != 0) { - int16_t phase_end_local[L2SWIMLANE_NUM_QUEUE_SHAPES]; - capture_local_snapshot(phase_end_local); uint64_t dummy_outer_t1 = get_sys_cnt_aicpu(); + 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::Dummy, dummy_outer_t0, dummy_outer_t1, l2_swimlane.sched_loop_count, static_cast(dummy_got), /*pop_hit=*/0, - /*pop_miss=*/0, phase_start_local, phase_start_shared, phase_end_local, phase_start_shared + /*pop_miss=*/0, phase_start_shared, phase_end_shared ); - for (int s = 0; s < L2SWIMLANE_NUM_QUEUE_SHAPES; s++) { - phase_start_local[s] = phase_end_local[s]; - } + for (int s = 0; s < L2SWIMLANE_NUM_QUEUE_SHAPES; s++) + phase_start_shared[s] = phase_end_shared[s]; _t0_phase = dummy_outer_t1; // We do NOT re-sync _t0/_t1 — the dummy span will be absorbed // into the next CYCLE_COUNT_LAP accumulator. The phase-model @@ -1178,7 +1082,7 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ #if PTO2_PROFILING uint64_t dispatch_t0 = (l2_swimlane_level_ >= L2SwimlaneLevel::SCHED_PHASES) ? get_sys_cnt_aicpu() : 0; #endif - dispatch_ready_tasks(thread_idx, tracker, local_bufs, pmu_active, made_progress, try_pushed); + dispatch_ready_tasks(thread_idx, tracker, pmu_active, made_progress, try_pushed); #if PTO2_PROFILING // Emit Dispatch IMMEDIATELY after dispatch_ready_tasks so its span // covers the actual publish work — not the trailing second-poll / @@ -1192,17 +1096,14 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ uint64_t pop_miss_delta = l2_swimlane.pop_miss - l2_swimlane.pop_miss_at_last_emit; debug_assert(pop_hit_delta < (1ULL << 32)); debug_assert(pop_miss_delta < (1ULL << 32)); - int16_t phase_end_local[L2SWIMLANE_NUM_QUEUE_SHAPES]; int16_t phase_end_shared[L2SWIMLANE_NUM_QUEUE_SHAPES]; - capture_phase_end(phase_end_local, phase_end_shared); + capture_phase_end(phase_end_shared); l2_swimlane_aicpu_record_sched_phase( thread_idx, L2SwimlaneSchedPhaseKind::Dispatch, _t0_phase, dispatch_t1, l2_swimlane.sched_loop_count, l2_swimlane.phase_dispatch_count, static_cast(pop_hit_delta), - static_cast(pop_miss_delta), phase_start_local, phase_start_shared, phase_end_local, - phase_end_shared + static_cast(pop_miss_delta), phase_start_shared, phase_end_shared ); for (int s = 0; s < L2SWIMLANE_NUM_QUEUE_SHAPES; s++) { - phase_start_local[s] = phase_end_local[s]; phase_start_shared[s] = phase_end_shared[s]; } _t0_phase = dispatch_t1; @@ -1220,7 +1121,7 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ // per-iteration cost (the discovery walk only runs on genuinely idle passes). bool any_ready_work = try_pushed; for (int s = 0; !any_ready_work && s < PTO2_NUM_RESOURCE_SHAPES; s++) { - if (sched_->ready_queues[s].size() > 0 || local_bufs[s].count > 0) any_ready_work = true; + if (sched_->ready_queues[s].size() > 0) any_ready_work = true; } #if PTO2_PROFILING bool early_dispatch_record = l2_swimlane_level_ >= L2SwimlaneLevel::SCHED_PHASES; @@ -1266,7 +1167,7 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ int32_t completed_2nd = 0; check_running_cores_for_completion( thread_idx, hank, completed_2nd, cur_thread_completed, made_progress, deferred_release_slot_states, - deferred_release_count, local_bufs + deferred_release_count ); if (completed_2nd > 0) { #if PTO2_SCHED_PROFILING @@ -1294,17 +1195,16 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ // Scheduler View). if (complete2_t0 != 0 && (l2_swimlane.phase_complete_count > 0 || l2_swimlane.phase_subretire_count > 0)) { uint64_t complete2_t1 = get_sys_cnt_aicpu(); - int16_t phase_end_local[L2SWIMLANE_NUM_QUEUE_SHAPES]; - capture_local_snapshot(phase_end_local); + 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::Complete, complete2_t0, complete2_t1, l2_swimlane.sched_loop_count, l2_swimlane.phase_complete_count + l2_swimlane.phase_subretire_count, /*pop_hit=*/0, - /*pop_miss=*/0, phase_start_local, phase_start_shared, phase_end_local, phase_start_shared + /*pop_miss=*/0, phase_start_shared, phase_end_shared ); - for (int s = 0; s < L2SWIMLANE_NUM_QUEUE_SHAPES; s++) { - phase_start_local[s] = phase_end_local[s]; - } + for (int s = 0; s < L2SWIMLANE_NUM_QUEUE_SHAPES; s++) + phase_start_shared[s] = phase_end_shared[s]; _t0_phase = complete2_t1; l2_swimlane.phase_complete_count = 0; l2_swimlane.phase_subretire_count = 0; @@ -1442,13 +1342,12 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ debug_assert(final_pop_miss_delta < (1ULL << 32)); if (final_pop_hit_delta != 0 || final_pop_miss_delta != 0) { uint64_t t_now = get_sys_cnt_aicpu(); - int16_t phase_end_local[L2SWIMLANE_NUM_QUEUE_SHAPES]; int16_t phase_end_shared[L2SWIMLANE_NUM_QUEUE_SHAPES]; - capture_phase_end(phase_end_local, phase_end_shared); + capture_phase_end(phase_end_shared); l2_swimlane_aicpu_record_sched_phase( thread_idx, L2SwimlaneSchedPhaseKind::Dispatch, t_now, t_now, l2_swimlane.sched_loop_count, 0, static_cast(final_pop_hit_delta), static_cast(final_pop_miss_delta), - phase_end_local, phase_end_shared, phase_end_local, phase_end_shared + phase_end_shared, phase_end_shared ); l2_swimlane.pop_hit_at_last_emit = l2_swimlane.pop_hit; l2_swimlane.pop_miss_at_last_emit = l2_swimlane.pop_miss; diff --git a/src/a5/platform/include/aicpu/l2_swimlane_collector_aicpu.h b/src/a5/platform/include/aicpu/l2_swimlane_collector_aicpu.h index 3182eb1c19..8c8b56d75b 100644 --- a/src/a5/platform/include/aicpu/l2_swimlane_collector_aicpu.h +++ b/src/a5/platform/include/aicpu/l2_swimlane_collector_aicpu.h @@ -166,11 +166,9 @@ void l2_swimlane_aicpu_init_phase(int worker_count, int num_sched_phase_threads, * pool. Silently drops records when the buffer is full or the pool was not * primed (init failed for this thread). * - * Queue-depth snapshots distinguish "task hidden in T0's local_buf" from - * "shared queue has it but peers spin on the wrong shape" — the former shows - * `local_depth > 0, shared_depth == 0` for the owning thread while peers see - * `shared_depth == 0` until overflow. Pass nullptr for any of the four arrays - * when not capturing (the record's corresponding slot is zero-filled). + * Queue-depth snapshots record the per-shape shared ready-queue occupancy at + * phase boundaries. Pass nullptr for either array when not capturing (the + * record's corresponding slot is zero-filled). * * @param thread_idx Scheduler thread index * @param kind Complete or Dispatch @@ -180,16 +178,12 @@ void l2_swimlane_aicpu_init_phase(int worker_count, int num_sched_phase_threads, * @param tasks_processed Tasks processed in this phase batch * @param pop_hit Dispatch delta since last emit (0 for Complete) * @param pop_miss Dispatch delta since last emit (0 for Complete) - * @param local_at_start Per-shape PTO2LocalReadyBuffer.count at phase start (size L2SWIMLANE_NUM_QUEUE_SHAPES; may be - * nullptr) * @param shared_at_start Per-shape sched.ready_queues[shape].size() at phase start (may be nullptr) - * @param local_at_end Per-shape PTO2LocalReadyBuffer.count at phase end (may be nullptr) * @param shared_at_end Per-shape sched.ready_queues[shape].size() at phase end (may be nullptr) */ void l2_swimlane_aicpu_record_sched_phase( int thread_idx, L2SwimlaneSchedPhaseKind kind, uint64_t start_time, uint64_t end_time, uint32_t loop_iter, - uint32_t tasks_processed, uint32_t pop_hit = 0, uint32_t pop_miss = 0, const int16_t *local_at_start = nullptr, - const int16_t *shared_at_start = nullptr, const int16_t *local_at_end = nullptr, + uint32_t tasks_processed, uint32_t pop_hit = 0, uint32_t pop_miss = 0, const int16_t *shared_at_start = nullptr, const int16_t *shared_at_end = nullptr ); diff --git a/src/a5/platform/include/common/l2_swimlane_profiling.h b/src/a5/platform/include/common/l2_swimlane_profiling.h index e9abdbf937..04c6a1837c 100644 --- a/src/a5/platform/include/common/l2_swimlane_profiling.h +++ b/src/a5/platform/include/common/l2_swimlane_profiling.h @@ -535,27 +535,22 @@ constexpr int L2SWIMLANE_NUM_QUEUE_SHAPES = 3; * (zero for Complete). Kept named, not "extra1"/"extra2", so the device-side * commit and the host-side JSON emit don't drift on which extra means which. * - * Queue-depth snapshots (local_depth_*, shared_depth_*) record the per-shape - * scheduler queue occupancy at phase boundaries. They surface the - * dep-release-then-discovery latency that head OH alone can't distinguish from - * register-write latency: a phase whose start sees `local_depth=N, shared=0` - * and end sees `local_depth=N-K` shows that K tasks were popped from this - * thread's private buffer (invisible to peer threads) — peers must spin until - * those tasks overflow into shared. Filled with 0 below SCHED_PHASES. + * Queue-depth snapshots (shared_depth_*) record the per-shape scheduler ready + * queue occupancy at phase boundaries. They surface the dep-release-then- + * discovery latency that head OH alone can't distinguish from register-write + * latency. Filled with 0 below SCHED_PHASES. */ struct L2SwimlaneAicpuSchedPhaseRecord { - uint64_t start_time; // Phase start timestamp - uint64_t end_time; // Phase end timestamp - uint32_t loop_iter; // Scheduler-loop iteration number on this thread - L2SwimlaneSchedPhaseKind kind; // see enum above - uint32_t tasks_processed; // Tasks processed in this phase batch - uint32_t pop_hit; // SCHED_DISPATCH delta since last emit (0 for Complete) - uint32_t pop_miss; // SCHED_DISPATCH delta since last emit (0 for Complete) - int16_t local_depth_at_start[L2SWIMLANE_NUM_QUEUE_SHAPES]; // this thread's PTO2LocalReadyBuffer.count - int16_t local_depth_at_end[L2SWIMLANE_NUM_QUEUE_SHAPES]; + uint64_t start_time; // Phase start timestamp + uint64_t end_time; // Phase end timestamp + uint32_t loop_iter; // Scheduler-loop iteration number on this thread + L2SwimlaneSchedPhaseKind kind; // see enum above + uint32_t tasks_processed; // Tasks processed in this phase batch + uint32_t pop_hit; // SCHED_DISPATCH delta since last emit (0 for Complete) + uint32_t pop_miss; // SCHED_DISPATCH delta since last emit (0 for Complete) int16_t shared_depth_at_start[L2SWIMLANE_NUM_QUEUE_SHAPES]; // sched->ready_queues[shape].size() int16_t shared_depth_at_end[L2SWIMLANE_NUM_QUEUE_SHAPES]; - uint32_t _pad; // 64B alignment padding + uint32_t _pad[4]; // 64B alignment padding }; static_assert(sizeof(L2SwimlaneAicpuSchedPhaseRecord) == 64, "L2SwimlaneAicpuSchedPhaseRecord layout drift"); diff --git a/src/a5/platform/shared/aicpu/l2_swimlane_collector_aicpu.cpp b/src/a5/platform/shared/aicpu/l2_swimlane_collector_aicpu.cpp index aafffa8ee4..cfb188065e 100644 --- a/src/a5/platform/shared/aicpu/l2_swimlane_collector_aicpu.cpp +++ b/src/a5/platform/shared/aicpu/l2_swimlane_collector_aicpu.cpp @@ -796,8 +796,8 @@ static Record *acquire_phase_slot( void l2_swimlane_aicpu_record_sched_phase( int thread_idx, L2SwimlaneSchedPhaseKind kind, uint64_t start_time, uint64_t end_time, uint32_t loop_iter, - uint32_t tasks_processed, uint32_t pop_hit, uint32_t pop_miss, const int16_t *local_at_start, - const int16_t *shared_at_start, const int16_t *local_at_end, const int16_t *shared_at_end + uint32_t tasks_processed, uint32_t pop_hit, uint32_t pop_miss, const int16_t *shared_at_start, + const int16_t *shared_at_end ) { if (!s_phase_initialized) return; auto *state = s_sched_phase_pools[thread_idx]; @@ -829,9 +829,7 @@ void l2_swimlane_aicpu_record_sched_phase( dst[i] = src[i]; } }; - copy_snapshot(record->local_depth_at_start, local_at_start); copy_snapshot(record->shared_depth_at_start, shared_at_start); - copy_snapshot(record->local_depth_at_end, local_at_end); copy_snapshot(record->shared_depth_at_end, shared_at_end); } diff --git a/src/a5/platform/shared/host/l2_swimlane_collector.cpp b/src/a5/platform/shared/host/l2_swimlane_collector.cpp index ee1b031273..e0c9f1cee7 100644 --- a/src/a5/platform/shared/host/l2_swimlane_collector.cpp +++ b/src/a5/platform/shared/host/l2_swimlane_collector.cpp @@ -877,9 +877,7 @@ int L2SwimlaneCollector::export_swimlane_json() { outfile << ", \"pop_hit\": " << pr.pop_hit << ", \"pop_miss\": " << pr.pop_miss; } // Queue-depth snapshots — [AIC, AIV, MIX] per L2SwimlaneAicpuSchedPhaseRecord docstring. - emit_depth_array("local_at_start", pr.local_depth_at_start); emit_depth_array("shared_at_start", pr.shared_depth_at_start); - emit_depth_array("local_at_end", pr.local_depth_at_end); emit_depth_array("shared_at_end", pr.shared_depth_at_end); outfile << "}"; first = false; diff --git a/src/a5/runtime/tensormap_and_ringbuffer/runtime/pto_async_wait.h b/src/a5/runtime/tensormap_and_ringbuffer/runtime/pto_async_wait.h index 42a9474187..d1da3ae892 100644 --- a/src/a5/runtime/tensormap_and_ringbuffer/runtime/pto_async_wait.h +++ b/src/a5/runtime/tensormap_and_ringbuffer/runtime/pto_async_wait.h @@ -24,7 +24,6 @@ #include "pto_runtime2_types.h" struct PTO2SchedulerState; -struct PTO2LocalReadyBuffer; struct CompletionStats; inline constexpr int32_t MAX_ASYNC_WAITS = 64; @@ -177,7 +176,6 @@ struct AsyncWaitList { // entries[]). struct DrainCompletionSink { PTO2SchedulerState *sched{nullptr}; - PTO2LocalReadyBuffer *local_bufs{nullptr}; PTO2TaskSlotState **deferred_release_slot_states{nullptr}; int32_t *deferred_release_count{nullptr}; int32_t deferred_release_capacity{0}; @@ -296,7 +294,7 @@ struct AsyncWaitList { template AsyncPollResult poll_and_complete( - AICoreCompletionMailbox *aicore_mailbox, PTO2SchedulerState *sched, PTO2LocalReadyBuffer *local_bufs, + AICoreCompletionMailbox *aicore_mailbox, PTO2SchedulerState *sched, PTO2TaskSlotState **deferred_release_slot_states, int32_t &deferred_release_count, int32_t deferred_release_capacity #if PTO2_SCHED_PROFILING diff --git a/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/pto_scheduler.h b/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/pto_scheduler.h index ffa47a3009..3a2966e972 100644 --- a/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/pto_scheduler.h +++ b/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/pto_scheduler.h @@ -61,42 +61,6 @@ struct PTO2ReadyQueueSlot { PTO2TaskSlotState *slot_state; }; -/** - * Thread-local ready buffer for local-first dispatch optimization. - * - * Two buffers per scheduling thread, one per CoreType (AIC=0, AIV=1). - * Initialized once before the scheduling loop; must be empty at - * the start of each iteration (verified by always_assert). - * - * Phase 1 fills per-CoreType buffers via on_task_complete(). - * The dispatch stage drains them local-first via get_ready_tasks_batch, - * with any remaining tasks pushed to the global ready queue. - */ -// Number of CoreType values eligible for local dispatch (AIC=0, AIV=1) -static constexpr int PTO2_LOCAL_DISPATCH_TYPE_NUM = 2; - -struct PTO2LocalReadyBuffer { - PTO2TaskSlotState **slot_states = nullptr; - int count = 0; - int capacity = 0; - - void reset(PTO2TaskSlotState **buf, int cap) { - slot_states = buf; - count = 0; - capacity = cap; - } - - bool try_push(PTO2TaskSlotState *s) { - if (slot_states && count < capacity) { - slot_states[count++] = s; - return true; - } - return false; - } - - PTO2TaskSlotState *pop() { return (count > 0) ? slot_states[--count] : nullptr; } -}; - /** * Lock-free bounded MPMC queue (Dmitry Vyukov design) * @@ -832,9 +796,7 @@ struct PTO2SchedulerState { // 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[]. Used by paths that do not have a thread-local - // ready buffer (e.g. wiring). See push_ready_routed_local for the - // dispatch-time fast path. + // per-shape ready_queues[]. void push_ready_routed(PTO2TaskSlotState *slot_state) { PTO2ResourceShape shape = slot_state->active_mask.to_shape(); if (shape == PTO2ResourceShape::DUMMY) { @@ -993,45 +955,32 @@ struct PTO2SchedulerState { } #endif - bool release_fanin_and_check_ready(PTO2TaskSlotState &slot_state, PTO2LocalReadyBuffer *local_bufs = nullptr) { + bool release_fanin_and_check_ready(PTO2TaskSlotState &slot_state) { // Atomically increment fanin_refcount and check if all producers are done // ACQ_REL on fanin_refcount already synchronizes with the orchestrator's // init release, making fanin_count visible — plain load suffices. int32_t new_refcount = slot_state.fanin_refcount.fetch_add(1, std::memory_order_acq_rel) + 1; if (new_refcount == slot_state.fanin_count) { - // Local-first: try per-CoreType thread-local buffer before global queue - // Route by active_mask: AIC-containing tasks → buf[0], AIV-only → buf[1] - // DUMMY shape is out of range for local_bufs (sized PTO2_NUM_RESOURCE_SHAPES); - // dummy slots bypass the local fast path and go straight to dummy_ready_queue. - PTO2ResourceShape shape = slot_state.active_mask.to_shape(); - if (shape == PTO2ResourceShape::DUMMY) { - dummy_ready_queue.push(&slot_state); - } else if (!local_bufs || !local_bufs[static_cast(shape)].try_push(&slot_state)) { - ready_queues[static_cast(shape)].push(&slot_state); - } + push_ready_routed(&slot_state); return true; } return false; } #if PTO2_ORCH_PROFILING || PTO2_SCHED_PROFILING - bool release_fanin_and_check_ready( - PTO2TaskSlotState &slot_state, uint64_t &atomic_count, uint64_t &push_wait, - PTO2LocalReadyBuffer *local_bufs = nullptr - ) { + bool release_fanin_and_check_ready(PTO2TaskSlotState &slot_state, uint64_t &atomic_count, uint64_t &push_wait) { int32_t new_refcount = slot_state.fanin_refcount.fetch_add(1, std::memory_order_acq_rel) + 1; atomic_count += 1; // fanin_refcount.fetch_add if (new_refcount == slot_state.fanin_count) { - // Local-first: try per-CoreType thread-local buffer before global queue. - // Dummy slots bypass local_bufs (out-of-range for PTO2_NUM_RESOURCE_SHAPES) - // and go straight to dummy_ready_queue; use the profiling-aware push so - // atomic_count / push_wait stay consistent with the non-dummy path. + // Dummy slots go to dummy_ready_queue; everything else to the per-shape + // ready_queues[]. Use the profiling-aware push so atomic_count / push_wait + // stay consistent with the non-dummy path. PTO2ResourceShape shape = slot_state.active_mask.to_shape(); if (shape == PTO2ResourceShape::DUMMY) { dummy_ready_queue.push(&slot_state, atomic_count, push_wait); - } else if (!local_bufs || !local_bufs[static_cast(shape)].try_push(&slot_state)) { + } else { ready_queues[static_cast(shape)].push(&slot_state, atomic_count, push_wait); } return true; @@ -1040,35 +989,15 @@ struct PTO2SchedulerState { } #endif - int get_ready_tasks_batch( - PTO2ResourceShape shape, PTO2LocalReadyBuffer &local_buf, PTO2TaskSlotState **out, int max_count - ) { - int count = 0; - while (count < max_count && local_buf.count > 0) { - out[count++] = local_buf.slot_states[--local_buf.count]; - } - int remaining = max_count - count; - if (remaining > 0) { - count += ready_queues[static_cast(shape)].pop_batch(out + count, remaining); - } - return count; + int get_ready_tasks_batch(PTO2ResourceShape shape, PTO2TaskSlotState **out, int max_count) { + return ready_queues[static_cast(shape)].pop_batch(out, max_count); } #if PTO2_SCHED_PROFILING int get_ready_tasks_batch( - PTO2ResourceShape shape, PTO2LocalReadyBuffer &local_buf, PTO2TaskSlotState **out, int max_count, - uint64_t &atomic_count, uint64_t &wait_cycle + PTO2ResourceShape shape, PTO2TaskSlotState **out, int max_count, uint64_t &atomic_count, uint64_t &wait_cycle ) { - int count = 0; - while (count < max_count && local_buf.count > 0) { - out[count++] = local_buf.slot_states[--local_buf.count]; - } - int remaining = max_count - count; - if (remaining > 0) { - count += - ready_queues[static_cast(shape)].pop_batch(out + count, remaining, atomic_count, wait_cycle); - } - return count; + return ready_queues[static_cast(shape)].pop_batch(out, max_count, atomic_count, wait_cycle); } #endif @@ -1114,12 +1043,11 @@ struct PTO2SchedulerState { void #endif on_task_complete( - PTO2TaskSlotState &slot_state, + PTO2TaskSlotState &slot_state #if PTO2_SCHED_PROFILING - int thread_idx, + , + int thread_idx #endif - - PTO2LocalReadyBuffer *local_bufs = nullptr ) { #if PTO2_SCHED_PROFILING CompletionStats stats = {0, 0, 0, true}; @@ -1156,11 +1084,11 @@ struct PTO2SchedulerState { PTO2TaskSlotState &consumer_slot = *current->slot_state; #if PTO2_SCHED_PROFILING stats.fanout_edges++; - if (release_fanin_and_check_ready(consumer_slot, fanout_atomics, push_wait, local_bufs)) { + if (release_fanin_and_check_ready(consumer_slot, fanout_atomics, push_wait)) { stats.tasks_enqueued++; } #else - release_fanin_and_check_ready(consumer_slot, local_bufs); + release_fanin_and_check_ready(consumer_slot); #endif current = current->next; } @@ -1254,9 +1182,9 @@ struct PTO2SchedulerState { inline bool AsyncWaitList::try_inline_complete_locked(AsyncWaitList::DrainCompletionSink &sink, PTO2TaskSlotState &slot_state) { #if PTO2_SCHED_PROFILING - sink.sched->on_task_complete(slot_state, sink.thread_idx, sink.local_bufs); + sink.sched->on_task_complete(slot_state, sink.thread_idx); #else - sink.sched->on_task_complete(slot_state, sink.local_bufs); + sink.sched->on_task_complete(slot_state); #endif if (*sink.deferred_release_count >= sink.deferred_release_capacity) { while (*sink.deferred_release_count > 0) { @@ -1276,7 +1204,7 @@ AsyncWaitList::try_inline_complete_locked(AsyncWaitList::DrainCompletionSink &si template inline AsyncPollResult AsyncWaitList::poll_and_complete( - AICoreCompletionMailbox *aicore_mailbox, PTO2SchedulerState *sched, PTO2LocalReadyBuffer *local_bufs, + AICoreCompletionMailbox *aicore_mailbox, PTO2SchedulerState *sched, PTO2TaskSlotState **deferred_release_slot_states, int32_t &deferred_release_count, int32_t deferred_release_capacity #if PTO2_SCHED_PROFILING , @@ -1288,7 +1216,6 @@ inline AsyncPollResult AsyncWaitList::poll_and_complete( AsyncWaitList::DrainCompletionSink sink{}; sink.sched = sched; - sink.local_bufs = local_bufs; sink.deferred_release_slot_states = deferred_release_slot_states; sink.deferred_release_count = &deferred_release_count; sink.deferred_release_capacity = deferred_release_capacity; @@ -1334,9 +1261,9 @@ inline AsyncPollResult AsyncWaitList::poll_and_complete( if (entry.normal_done && entry.waiting_completion_count <= 0) { #if PTO2_SCHED_PROFILING - sched->on_task_complete(*entry.slot_state, thread_idx, local_bufs); + sched->on_task_complete(*entry.slot_state, thread_idx); #else - sched->on_task_complete(*entry.slot_state, local_bufs); + sched->on_task_complete(*entry.slot_state); #endif if (deferred_release_count >= deferred_release_capacity) { while (deferred_release_count > 0) { diff --git a/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_completion.cpp b/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_completion.cpp index 7d83249aba..276a04d3de 100644 --- a/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_completion.cpp +++ b/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_completion.cpp @@ -70,7 +70,7 @@ SlotTransition SchedulerContext::decide_slot_transition( void SchedulerContext::complete_slot_task( PTO2TaskSlotState &slot_state, int32_t expected_reg_task_id, [[maybe_unused]] PTO2SubtaskSlot subslot, int32_t thread_idx, int32_t core_id, Handshake *hank, int32_t &completed_this_turn, - PTO2TaskSlotState *deferred_release_slot_states[], int32_t &deferred_release_count, PTO2LocalReadyBuffer *local_bufs + PTO2TaskSlotState *deferred_release_slot_states[], int32_t &deferred_release_count #if PTO2_PROFILING , uint64_t dispatch_ts, uint64_t finish_ts @@ -153,9 +153,9 @@ void SchedulerContext::complete_slot_task( // SCHED_PROFILING variant takes thread_idx for its per-thread atomic // counter side-effects (g_sched_*_atomic_count[thread_idx], consumed // by the otc_* log lines). Its return value is unused. - (void)sched_->on_task_complete(slot_state, thread_idx, local_bufs); + (void)sched_->on_task_complete(slot_state, thread_idx); #else - sched_->on_task_complete(slot_state, local_bufs); + sched_->on_task_complete(slot_state); #endif #if PTO2_PROFILING l2_swimlane.phase_complete_count++; @@ -238,8 +238,7 @@ void SchedulerContext::clear_running_slot(CoreExecState &core) { void SchedulerContext::check_running_cores_for_completion( int32_t thread_idx, Handshake *hank, int32_t &completed_this_turn, int32_t &cur_thread_completed, - bool &made_progress, PTO2TaskSlotState *deferred_release_slot_states[], int32_t &deferred_release_count, - PTO2LocalReadyBuffer *local_bufs + bool &made_progress, PTO2TaskSlotState *deferred_release_slot_states[], int32_t &deferred_release_count ) { #if PTO2_SCHED_PROFILING auto &l2_swimlane = sched_l2_swimlane_[thread_idx]; @@ -296,7 +295,7 @@ void SchedulerContext::check_running_cores_for_completion( if (t.pending_done) { complete_slot_task( *core.pending_slot_state, core.pending_reg_task_id, core.pending_subslot, thread_idx, core_id, hank, - completed_this_turn, deferred_release_slot_states, deferred_release_count, local_bufs + completed_this_turn, deferred_release_slot_states, deferred_release_count #if PTO2_PROFILING , core.pending_dispatch_timestamp, finish_ts @@ -307,7 +306,7 @@ void SchedulerContext::check_running_cores_for_completion( if (t.running_done) { complete_slot_task( *core.running_slot_state, core.running_reg_task_id, core.running_subslot, thread_idx, core_id, hank, - completed_this_turn, deferred_release_slot_states, deferred_release_count, local_bufs + completed_this_turn, deferred_release_slot_states, deferred_release_count #if PTO2_PROFILING , core.running_dispatch_timestamp, finish_ts diff --git a/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_context.h b/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_context.h index a6c0bdfa75..2ec4645876 100644 --- a/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_context.h +++ b/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_context.h @@ -208,10 +208,7 @@ class SchedulerContext { return "?"; } - int pop_ready_tasks_batch( - PTO2ResourceShape shape, int32_t thread_idx, PTO2LocalReadyBuffer &local_buf, PTO2TaskSlotState **out, - int max_count - ); + int pop_ready_tasks_batch(PTO2ResourceShape shape, int32_t thread_idx, PTO2TaskSlotState **out, int max_count); void build_payload( PTO2DispatchPayload &dispatch_payload, PTO2TaskSlotState &slot_state, PTO2SubtaskSlot subslot, @@ -235,15 +232,14 @@ class SchedulerContext { void dispatch_shape( Runtime *runtime, int32_t thread_idx, PTO2ResourceShape shape, CoreTracker::DispatchPhase phase, - PTO2LocalReadyBuffer &local_buf, CoreTracker &tracker, bool &entered_drain, bool &made_progress, - bool &try_pushed + CoreTracker &tracker, bool &entered_drain, bool &made_progress, bool &try_pushed ); // One pass of "Phase 4" in the resolve_and_dispatch loop: IDLE-stage dispatch - // for MIX then (if no mix residual) AIC/AIV; mid-flush of local buffers; then - // PENDING-stage dispatch with cross-thread idle gating. MIX is strictly - // prioritized — when mix residual is detected after MIX-IDLE, AIC/AIV are - // skipped for the whole pass but MIX-PENDING still runs. + // for MIX then (if no mix residual) AIC/AIV; then PENDING-stage dispatch with + // cross-thread idle gating. MIX is strictly prioritized — when mix residual is + // detected after MIX-IDLE, AIC/AIV are skipped for the whole pass but + // MIX-PENDING still runs. // // Forward-progress argument for AIC/AIV: skip_aic_aiv is sticky for the // current pass only. The next loop iteration re-evaluates after Phase 1 @@ -252,8 +248,7 @@ class SchedulerContext { // not unbounded — once mix completes on at least one cluster, the next // pass either drains the residual or admits AIC/AIV. void dispatch_ready_tasks( - Runtime *runtime, int32_t thread_idx, CoreTracker &tracker, - PTO2LocalReadyBuffer (&local_bufs)[PTO2_NUM_RESOURCE_SHAPES], bool pmu_active, bool &made_progress, + Runtime *runtime, int32_t thread_idx, CoreTracker &tracker, bool pmu_active, bool &made_progress, bool &try_pushed ); @@ -263,15 +258,14 @@ class SchedulerContext { // rationale and the safety argument against the drain worker. bool has_idle_in_other_threads(int32_t self_thread_idx, PTO2ResourceShape shape) const; - // True if mix tasks remain anywhere this thread could see them: the caller's - // MIX local LIFO stack or the global MIX ready queue. Approximate — + // 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 // extra/missed AIC/AIV skip and self-corrects on the next loop iteration. - bool has_residual_mix(const PTO2LocalReadyBuffer &mix_local_buf) const { - return mix_local_buf.count > 0 || sched_->ready_queues[static_cast(PTO2ResourceShape::MIX)].size() > 0; + bool has_residual_mix() const { + return sched_->ready_queues[static_cast(PTO2ResourceShape::MIX)].size() > 0; } // ========================================================================= @@ -284,8 +278,7 @@ class SchedulerContext { void complete_slot_task( PTO2TaskSlotState &slot_state, int32_t expected_reg_task_id, PTO2SubtaskSlot subslot, int32_t thread_idx, int32_t core_id, Handshake *hank, int32_t &completed_this_turn, - PTO2TaskSlotState *deferred_release_slot_states[], int32_t &deferred_release_count, - PTO2LocalReadyBuffer *local_bufs + PTO2TaskSlotState *deferred_release_slot_states[], int32_t &deferred_release_count #if PTO2_PROFILING , uint64_t dispatch_ts, uint64_t finish_ts @@ -297,8 +290,7 @@ class SchedulerContext { void check_running_cores_for_completion( int32_t thread_idx, Handshake *hank, int32_t &completed_this_turn, int32_t &cur_thread_completed, - bool &made_progress, PTO2TaskSlotState *deferred_release_slot_states[], int32_t &deferred_release_count, - PTO2LocalReadyBuffer *local_bufs + bool &made_progress, PTO2TaskSlotState *deferred_release_slot_states[], int32_t &deferred_release_count ); bool enter_drain_mode(PTO2TaskSlotState *slot_state, int32_t block_num); diff --git a/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_dispatch.cpp b/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_dispatch.cpp index beea5ada44..1b6ad25d33 100644 --- a/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_dispatch.cpp +++ b/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_dispatch.cpp @@ -73,7 +73,7 @@ bool SchedulerContext::has_idle_in_other_threads(int32_t self_thread_idx, PTO2Re } int SchedulerContext::pop_ready_tasks_batch( - PTO2ResourceShape shape, int32_t thread_idx, PTO2LocalReadyBuffer &local_buf, PTO2TaskSlotState **out, int max_count + PTO2ResourceShape shape, int32_t thread_idx, PTO2TaskSlotState **out, int max_count ) { #if PTO2_PROFILING auto &l2_swimlane = sched_l2_swimlane_[thread_idx]; @@ -81,11 +81,11 @@ int SchedulerContext::pop_ready_tasks_batch( extern uint64_t g_sched_pop_atomic_count[], g_sched_pop_wait_cycle[]; uint64_t t_pop_start = get_sys_cnt_aicpu(); int count = sched_->get_ready_tasks_batch( - shape, local_buf, out, max_count, g_sched_pop_atomic_count[thread_idx], g_sched_pop_wait_cycle[thread_idx] + shape, out, max_count, g_sched_pop_atomic_count[thread_idx], g_sched_pop_wait_cycle[thread_idx] ); l2_swimlane.sched_dispatch_pop_cycle += (get_sys_cnt_aicpu() - t_pop_start); #else - int count = sched_->get_ready_tasks_batch(shape, local_buf, out, max_count); + int count = sched_->get_ready_tasks_batch(shape, out, max_count); #endif if (l2_swimlane_level_ >= L2SwimlaneLevel::SCHED_PHASES) { if (count > 0) { @@ -96,7 +96,7 @@ int SchedulerContext::pop_ready_tasks_batch( } #else (void)thread_idx; - int count = sched_->get_ready_tasks_batch(shape, local_buf, out, max_count); + int count = sched_->get_ready_tasks_batch(shape, out, max_count); #endif return count; } @@ -263,7 +263,7 @@ void SchedulerContext::dispatch_block( void SchedulerContext::dispatch_shape( Runtime *runtime, int32_t thread_idx, PTO2ResourceShape shape, CoreTracker::DispatchPhase phase, - PTO2LocalReadyBuffer &local_buf, CoreTracker &tracker, bool &entered_drain, bool &made_progress, bool &try_pushed + CoreTracker &tracker, bool &entered_drain, bool &made_progress, bool &try_pushed ) { #if PTO2_SCHED_PROFILING auto &l2_swimlane = sched_l2_swimlane_[thread_idx]; @@ -278,7 +278,7 @@ void SchedulerContext::dispatch_shape( while (cores.has_value() && !entered_drain) { int want = cores.count(); PTO2TaskSlotState *batch[CoreTracker::MAX_CLUSTERS * 3]; - int got = pop_ready_tasks_batch(shape, thread_idx, local_buf, batch, want); + int got = pop_ready_tasks_batch(shape, thread_idx, batch, want); if (got == 0) break; bool dispatched_any = false; @@ -369,11 +369,9 @@ void SchedulerContext::dispatch_shape( } void SchedulerContext::dispatch_ready_tasks( - Runtime *runtime, int32_t thread_idx, CoreTracker &tracker, - PTO2LocalReadyBuffer (&local_bufs)[PTO2_NUM_RESOURCE_SHAPES], bool pmu_active, bool &made_progress, bool &try_pushed + Runtime *runtime, int32_t thread_idx, CoreTracker &tracker, bool pmu_active, bool &made_progress, bool &try_pushed ) { using Phase = CoreTracker::DispatchPhase; - constexpr int32_t MIX_I = static_cast(PTO2ResourceShape::MIX); // MIX is handled explicitly at the top of each stage; only AIC/AIV cycle // through this 2-elem array, with order toggled by thread parity for @@ -384,80 +382,52 @@ void SchedulerContext::dispatch_ready_tasks( }; const PTO2ResourceShape *aic_aiv = kAicAivOrder[thread_idx & 1]; - auto flush_local_bufs = [&]() { - for (int32_t s = 0; s < PTO2_NUM_RESOURCE_SHAPES; s++) { - auto &lb = local_bufs[s]; - if (lb.count > 0) { - sched_->ready_queues[s].push_batch(lb.slot_states, lb.count); - lb.count = 0; - } - } - }; - // Every return path below must flush; wrap in RAII so we cannot forget. - // The mid-function flush between IDLE and PENDING is still called - // explicitly — guard only covers exit. - struct FlushGuard { - decltype(flush_local_bufs) &flush_fn; - ~FlushGuard() { flush_fn(); } - } flush_guard{flush_local_bufs}; - bool entered_drain = false; // ===== IDLE stage ===== dispatch_shape( - runtime, thread_idx, PTO2ResourceShape::MIX, Phase::IDLE, local_bufs[MIX_I], tracker, entered_drain, - made_progress, try_pushed + runtime, thread_idx, PTO2ResourceShape::MIX, Phase::IDLE, tracker, entered_drain, made_progress, try_pushed ); if (entered_drain) return; // MIX-IDLE residual: AIC/AIV (both IDLE and PENDING) yield for this pass. // MIX-PENDING below still runs — that is the core of "mix strict priority": // pending slots are spent on mix before AIC/AIV get any chance. - bool skip_aic_aiv = has_residual_mix(local_bufs[MIX_I]); + bool skip_aic_aiv = has_residual_mix(); if (!skip_aic_aiv) { for (int i = 0; i < 2; i++) { PTO2ResourceShape s = aic_aiv[i]; - dispatch_shape( - runtime, thread_idx, s, Phase::IDLE, local_bufs[static_cast(s)], tracker, entered_drain, - made_progress, try_pushed - ); + dispatch_shape(runtime, thread_idx, s, Phase::IDLE, tracker, entered_drain, made_progress, try_pushed); if (entered_drain) return; } } - // Flush between IDLE and PENDING so PENDING-stage queue-size checks and any - // peer-thread reads see the IDLE-stage release_fanin output. - flush_local_bufs(); - if (pmu_active) return; // ===== PENDING stage ===== // MIX-PENDING gate: skip when a peer has an idle MIX-capable cluster — that // peer's next IDLE-MIX iteration will pull the mix task from the global - // queue (already flushed above) at lower latency than us pre-loading a - // pending slot here. Forward progress for MIX is preserved: at least one - // thread will run MIX-IDLE next pass and consume the residual. + // queue at lower latency than us pre-loading a pending slot here. Forward + // progress for MIX is preserved: at least one thread will run MIX-IDLE next + // pass and consume the residual. // // The gate is NOT subject to skip_aic_aiv — residual mix continues to drain // via pending slots on this thread when no peer is idle. if (!has_idle_in_other_threads(thread_idx, PTO2ResourceShape::MIX)) { dispatch_shape( - runtime, thread_idx, PTO2ResourceShape::MIX, Phase::PENDING, local_bufs[MIX_I], tracker, entered_drain, - made_progress, try_pushed + runtime, thread_idx, PTO2ResourceShape::MIX, Phase::PENDING, tracker, entered_drain, made_progress, + try_pushed ); if (entered_drain) return; } // Re-check after MIX-PENDING. If MIX-IDLE already set skip_aic_aiv, leave // it set; otherwise, escalate iff PENDING-MIX left residual. - if (!skip_aic_aiv && has_residual_mix(local_bufs[MIX_I])) { + if (!skip_aic_aiv && has_residual_mix()) { skip_aic_aiv = true; } - // PENDING-MIX may have re-populated AIC/AIV local_bufs via release_fanin - // during in-flight completions; flush_guard ensures these don't carry - // across to the next iteration's IDLE stage. if (skip_aic_aiv) return; // AIC/AIV-PENDING gate: a peer-idle skip is a delay, not a loss — the peer @@ -465,10 +435,7 @@ void SchedulerContext::dispatch_ready_tasks( for (int i = 0; i < 2; i++) { PTO2ResourceShape s = aic_aiv[i]; if (has_idle_in_other_threads(thread_idx, s)) continue; - dispatch_shape( - runtime, thread_idx, s, Phase::PENDING, local_bufs[static_cast(s)], tracker, entered_drain, - made_progress, try_pushed - ); + dispatch_shape(runtime, thread_idx, s, Phase::PENDING, tracker, entered_drain, made_progress, try_pushed); if (entered_drain) return; } } @@ -512,12 +479,6 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ l2_swimlane.l2_swimlane_enabled = (l2_swimlane_level_ != L2SwimlaneLevel::DISABLED); #endif - constexpr int LOCAL_READY_CAP_PER_TYPE = 64; - PTO2TaskSlotState *local_ptrs[PTO2_NUM_RESOURCE_SHAPES][LOCAL_READY_CAP_PER_TYPE]; - PTO2LocalReadyBuffer local_bufs[PTO2_NUM_RESOURCE_SHAPES]; - for (int32_t i = 0; i < PTO2_NUM_RESOURCE_SHAPES; i++) { - local_bufs[i].reset(local_ptrs[i], LOCAL_READY_CAP_PER_TYPE); - } PTO2TaskSlotState *deferred_release_slot_states[PTO2_DEFERRED_RELEASE_CAP]; int32_t deferred_release_count = 0; @@ -540,34 +501,27 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ #if PTO2_PROFILING // Queue-depth snapshot carried across the iteration boundary: each phase - // emit consumes (phase_start_*) and refreshes them with its own end snapshot - // so the next phase's "at_start" equals the previous phase's "at_end". + // emit consumes (phase_start_shared) and refreshes it with its own end + // snapshot so the next phase's "at_start" equals the previous phase's + // "at_end". // // L2SWIMLANE_NUM_QUEUE_SHAPES (3) matches PTO2_NUM_RESOURCE_SHAPES: AIC/AIV/MIX. // - // **Hot-path cost discipline.** Local depth (this thread's PTO2LocalReadyBuffer) - // is a single int read on a register-cached stack — free. Shared depth - // (PTO2ReadyQueue::size) is two atomic relaxed loads against cache lines - // that all peer sched threads also write to (enqueue_pos and dequeue_pos - // bounce on every flush_local_bufs + every pop). With both phases emitting - // per iter that's 12 cross-core loads × thousands of iters per run, a - // measurable AICPU slowdown. Mitigation: lazy + per-iter cached shared - // snapshot, refreshed at most once per iteration. The complete-emit and - // dispatch-emit in the same iter both reuse the same shared sample; the - // big transitions (local→shared flush) still show up across iter boundaries. + // **Hot-path cost discipline.** Shared depth (PTO2ReadyQueue::size) is two + // atomic relaxed loads against cache lines that all peer sched threads also + // write to (enqueue_pos and dequeue_pos bounce on every push + every pop). + // With both phases emitting per iter that's cross-core loads × thousands of + // iters per run, a measurable AICPU slowdown. Mitigation: lazy + per-iter + // cached shared snapshot, refreshed at most once per iteration. The + // complete-emit and dispatch-emit in the same iter both reuse the same + // shared sample. static_assert( L2SWIMLANE_NUM_QUEUE_SHAPES == PTO2_NUM_RESOURCE_SHAPES, "queue snapshot width must match runtime resource shape count" ); - int16_t phase_start_local[L2SWIMLANE_NUM_QUEUE_SHAPES] = {0}; int16_t phase_start_shared[L2SWIMLANE_NUM_QUEUE_SHAPES] = {0}; int16_t iter_shared_snapshot[L2SWIMLANE_NUM_QUEUE_SHAPES] = {0}; bool iter_shared_sampled = false; - auto capture_local_snapshot = [&](int16_t local_out[L2SWIMLANE_NUM_QUEUE_SHAPES]) { - for (int s = 0; s < L2SWIMLANE_NUM_QUEUE_SHAPES; s++) { - local_out[s] = static_cast(local_bufs[s].count); - } - }; auto get_or_sample_shared = [&]() -> const int16_t * { if (!iter_shared_sampled) { // Clamp to int16_t max before narrowing. PTO2_PROF_READYQUEUE_SIZE @@ -583,15 +537,22 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ } return iter_shared_snapshot; }; - auto capture_phase_end = [&](int16_t local_out[L2SWIMLANE_NUM_QUEUE_SHAPES], - int16_t shared_out[L2SWIMLANE_NUM_QUEUE_SHAPES]) { - capture_local_snapshot(local_out); + auto capture_phase_end = [&](int16_t shared_out[L2SWIMLANE_NUM_QUEUE_SHAPES]) { const int16_t *shared_cached = get_or_sample_shared(); for (int s = 0; s < L2SWIMLANE_NUM_QUEUE_SHAPES; s++) shared_out[s] = shared_cached[s]; }; + // Queue-mutating phases (Complete / Wire / Dummy) push newly-ready consumers + // straight into the shared ready_queues[] (the local-first buffer is gone), + // so their end-of-phase shared depth differs from their start. Force a fresh + // re-sample for those emits — this also refreshes the per-iter cache so the + // next phase's start snapshot is not stale. + auto capture_phase_end_fresh = [&](int16_t shared_out[L2SWIMLANE_NUM_QUEUE_SHAPES]) { + iter_shared_sampled = false; + capture_phase_end(shared_out); + }; if (l2_swimlane_level_ >= L2SwimlaneLevel::SCHED_PHASES) { - capture_phase_end(phase_start_local, phase_start_shared); + capture_phase_end(phase_start_shared); } #endif @@ -643,7 +604,7 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ if (try_completed) { check_running_cores_for_completion( thread_idx, hank, completed_this_turn, cur_thread_completed, made_progress, - deferred_release_slot_states, deferred_release_count, local_bufs + deferred_release_slot_states, deferred_release_count ); } if (completed_this_turn > 0) { @@ -667,7 +628,7 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ if (rt_ != nullptr && rt_->aicore_mailbox != nullptr && (sched_->async_wait_list.count > 0 || rt_->aicore_mailbox->has_pending())) { AsyncPollResult poll_result = sched_->async_wait_list.poll_and_complete( - rt_->aicore_mailbox, sched_, local_bufs, deferred_release_slot_states, deferred_release_count, + rt_->aicore_mailbox, sched_, deferred_release_slot_states, deferred_release_count, PTO2_DEFERRED_RELEASE_CAP #if PTO2_SCHED_PROFILING , @@ -699,24 +660,17 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ } else { CYCLE_COUNT_LAP(l2_swimlane.sched_complete_cycle); if (l2_swimlane_level_ >= L2SwimlaneLevel::SCHED_PHASES && l2_swimlane.phase_complete_count > 0) { - // Local depth is cheap (this thread's own buffer counter). - // Shared depth is NOT sampled here: complete's release_fanin - // pushes to local_bufs in the fast path (try_push succeeds - // until cap=64). Shared only changes on dispatch's flush - // path. Carrying phase_start_shared forward as end_shared - // is the right answer 99% of the time AND skips three - // contended atomic loads per emit. - int16_t phase_end_local[L2SWIMLANE_NUM_QUEUE_SHAPES]; - capture_local_snapshot(phase_end_local); + // Complete's release_fanin pushes newly-ready consumers into the + // shared ready_queues[], so the end depth differs from the start. + 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::Complete, _t0_phase, _t1, l2_swimlane.sched_loop_count, - l2_swimlane.phase_complete_count, /*pop_hit=*/0, /*pop_miss=*/0, phase_start_local, - phase_start_shared, phase_end_local, phase_start_shared + l2_swimlane.phase_complete_count, /*pop_hit=*/0, /*pop_miss=*/0, phase_start_shared, + phase_end_shared ); - for (int s = 0; s < L2SWIMLANE_NUM_QUEUE_SHAPES; s++) { - phase_start_local[s] = phase_end_local[s]; - // phase_start_shared unchanged — carried forward - } + for (int s = 0; s < L2SWIMLANE_NUM_QUEUE_SHAPES; s++) + phase_start_shared[s] = phase_end_shared[s]; _t0_phase = _t1; l2_swimlane.phase_complete_count = 0; } @@ -747,16 +701,14 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ // Wire outer phase: emit one bar covering this iter's drain_wiring_queue // pass when it wired any tasks. tasks_processed = wired count. if (l2_swimlane_level_ >= L2SwimlaneLevel::SCHED_PHASES && wired > 0) { - int16_t phase_end_local[L2SWIMLANE_NUM_QUEUE_SHAPES]; - capture_local_snapshot(phase_end_local); + 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_local, phase_start_shared, - phase_end_local, phase_start_shared + 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_local[s] = phase_end_local[s]; - } + for (int s = 0; s < L2SWIMLANE_NUM_QUEUE_SHAPES; s++) + phase_start_shared[s] = phase_end_shared[s]; _t0_phase = _t1; } #endif @@ -780,9 +732,9 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ for (int di = 0; di < dummy_got; di++) { PTO2TaskSlotState &dummy_slot = *dummy_batch[di]; #if PTO2_SCHED_PROFILING - sched_->on_task_complete(dummy_slot, thread_idx, local_bufs); + sched_->on_task_complete(dummy_slot, thread_idx); #else - sched_->on_task_complete(dummy_slot, local_bufs); + sched_->on_task_complete(dummy_slot); #endif // Dummy tasks have no subtasks to retire and no fanout pre-conditions // beyond their own producers; release self-reference so the slot can @@ -808,17 +760,16 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ } #if PTO2_PROFILING if (dummy_outer_t0 != 0) { - int16_t phase_end_local[L2SWIMLANE_NUM_QUEUE_SHAPES]; - capture_local_snapshot(phase_end_local); uint64_t dummy_outer_t1 = get_sys_cnt_aicpu(); + 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::Dummy, dummy_outer_t0, dummy_outer_t1, l2_swimlane.sched_loop_count, static_cast(dummy_got), /*pop_hit=*/0, - /*pop_miss=*/0, phase_start_local, phase_start_shared, phase_end_local, phase_start_shared + /*pop_miss=*/0, phase_start_shared, phase_end_shared ); - for (int s = 0; s < L2SWIMLANE_NUM_QUEUE_SHAPES; s++) { - phase_start_local[s] = phase_end_local[s]; - } + for (int s = 0; s < L2SWIMLANE_NUM_QUEUE_SHAPES; s++) + phase_start_shared[s] = phase_end_shared[s]; _t0_phase = dummy_outer_t1; } #endif @@ -827,7 +778,7 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ // Phase 4: MIX-strict-priority dispatch with phase-split and // cross-thread idle gating. See dispatch_ready_tasks for the policy. // pmu_active is cached at function scope above (loop-invariant). - dispatch_ready_tasks(runtime, thread_idx, tracker, local_bufs, pmu_active, made_progress, try_pushed); + dispatch_ready_tasks(runtime, thread_idx, tracker, pmu_active, made_progress, try_pushed); #if PTO2_PROFILING if (!try_pushed) { @@ -844,17 +795,14 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ // realistic dispatch cadence and silently truncates without this guard. debug_assert(pop_hit_delta < (1ULL << 32)); debug_assert(pop_miss_delta < (1ULL << 32)); - int16_t phase_end_local[L2SWIMLANE_NUM_QUEUE_SHAPES]; int16_t phase_end_shared[L2SWIMLANE_NUM_QUEUE_SHAPES]; - capture_phase_end(phase_end_local, phase_end_shared); + capture_phase_end(phase_end_shared); l2_swimlane_aicpu_record_sched_phase( thread_idx, L2SwimlaneSchedPhaseKind::Dispatch, _t0_phase, _t1, l2_swimlane.sched_loop_count, l2_swimlane.phase_dispatch_count, static_cast(pop_hit_delta), - static_cast(pop_miss_delta), phase_start_local, phase_start_shared, phase_end_local, - phase_end_shared + static_cast(pop_miss_delta), phase_start_shared, phase_end_shared ); for (int s = 0; s < L2SWIMLANE_NUM_QUEUE_SHAPES; s++) { - phase_start_local[s] = phase_end_local[s]; phase_start_shared[s] = phase_end_shared[s]; } _t0_phase = _t1; @@ -967,13 +915,12 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ debug_assert(final_pop_miss_delta < (1ULL << 32)); if (final_pop_hit_delta != 0 || final_pop_miss_delta != 0) { uint64_t t_now = get_sys_cnt_aicpu(); - int16_t phase_end_local[L2SWIMLANE_NUM_QUEUE_SHAPES]; int16_t phase_end_shared[L2SWIMLANE_NUM_QUEUE_SHAPES]; - capture_phase_end(phase_end_local, phase_end_shared); + capture_phase_end(phase_end_shared); l2_swimlane_aicpu_record_sched_phase( thread_idx, L2SwimlaneSchedPhaseKind::Dispatch, t_now, t_now, l2_swimlane.sched_loop_count, 0, static_cast(final_pop_hit_delta), static_cast(final_pop_miss_delta), - phase_end_local, phase_end_shared, phase_end_local, phase_end_shared + phase_end_shared, phase_end_shared ); l2_swimlane.pop_hit_at_last_emit = l2_swimlane.pop_hit; l2_swimlane.pop_miss_at_last_emit = l2_swimlane.pop_miss; diff --git a/tests/ut/cpp/a2a3/test_ready_queue.cpp b/tests/ut/cpp/a2a3/test_ready_queue.cpp index 886b613be5..5cf411d75f 100644 --- a/tests/ut/cpp/a2a3/test_ready_queue.cpp +++ b/tests/ut/cpp/a2a3/test_ready_queue.cpp @@ -9,10 +9,9 @@ * ----------------------------------------------------------------------------------------------------------- */ /** - * Unit tests for PTO2ReadyQueue and PTO2LocalReadyBuffer from pto_scheduler.h + * Unit tests for PTO2ReadyQueue from pto_scheduler.h * - * Tests the lock-free bounded MPMC queue (Vyukov design) and the thread-local - * ready buffer used for local-first dispatch optimization. + * Tests the lock-free bounded MPMC queue (Vyukov design). * * Design contracts: * @@ -30,10 +29,6 @@ * - size() relaxed ordering: size() reads both positions with * memory_order_relaxed and is a hint, not a snapshot. If a stale read * produces d > e the guard returns 0. - * - * - LocalReadyBuffer LIFO dispatch: try_push appends at count++, pop returns - * slot_states[--count]. LIFO reversal is intentional for cache-locality - * when a producer immediately dispatches its fanout. */ #include @@ -406,71 +401,3 @@ INSTANTIATE_TEST_SUITE_P( MPMCConfig{4, 4, 1250} // HighContentionStress ) ); - -// ============================================================================= -// LocalReadyBuffer -// ============================================================================= - -class LocalReadyBufferTest : public ::testing::Test { -protected: - static constexpr int CAPACITY = 8; - - PTO2LocalReadyBuffer buffer; - PTO2TaskSlotState *backing[CAPACITY]; - - void SetUp() override { buffer.reset(backing, CAPACITY); } -}; - -// --- Normal path --- - -TEST_F(LocalReadyBufferTest, PopEmptyReturnsNullptr) { EXPECT_EQ(buffer.pop(), nullptr); } - -// LIFO dispatch: try_push appends at count++, pop returns slot_states[--count]. -TEST_F(LocalReadyBufferTest, LIFOOrdering) { - PTO2TaskSlotState a, b; - - ASSERT_TRUE(buffer.try_push(&a)); - ASSERT_TRUE(buffer.try_push(&b)); - - EXPECT_EQ(buffer.pop(), &b); - EXPECT_EQ(buffer.pop(), &a); - EXPECT_EQ(buffer.pop(), nullptr); -} - -TEST_F(LocalReadyBufferTest, TryPushFullReturnsFalse) { - PTO2TaskSlotState items[CAPACITY + 1]; - - for (int i = 0; i < CAPACITY; i++) { - ASSERT_TRUE(buffer.try_push(&items[i])); - } - - EXPECT_FALSE(buffer.try_push(&items[CAPACITY])); -} - -TEST_F(LocalReadyBufferTest, ResetSetsCleanState) { - EXPECT_EQ(buffer.pop(), nullptr) << "Fresh buffer is empty"; - - PTO2TaskSlotState a, b; - ASSERT_TRUE(buffer.try_push(&a)); - ASSERT_TRUE(buffer.try_push(&b)); - - buffer.reset(backing, CAPACITY); - EXPECT_EQ(buffer.pop(), nullptr) << "Buffer is empty after reset"; - - PTO2TaskSlotState items[CAPACITY]; - for (int i = 0; i < CAPACITY; i++) { - EXPECT_TRUE(buffer.try_push(&items[i])); - } - EXPECT_FALSE(buffer.try_push(&a)) << "Full after pushing capacity items post-reset"; -} - -// --- Boundary conditions --- - -TEST_F(LocalReadyBufferTest, NullBackingBuffer) { - PTO2LocalReadyBuffer buf; - buf.reset(nullptr, 0); - - PTO2TaskSlotState item{}; - EXPECT_FALSE(buf.try_push(&item)) << "Push fails with null backing"; - EXPECT_EQ(buf.pop(), nullptr) << "Pop returns null with null backing"; -} diff --git a/tests/ut/cpp/a2a3/test_scheduler_state.cpp b/tests/ut/cpp/a2a3/test_scheduler_state.cpp index 51e0c9ab76..a2c8e96086 100644 --- a/tests/ut/cpp/a2a3/test_scheduler_state.cpp +++ b/tests/ut/cpp/a2a3/test_scheduler_state.cpp @@ -188,27 +188,25 @@ TEST_F(SchedulerStateTest, ScopeEndBatchRelease) { } // ============================================================================= -// get_ready_tasks_batch: local buffer first +// get_ready_tasks_batch: drains the shared ready queue // ============================================================================= -TEST_F(SchedulerStateTest, GetReadyTasksBatchLocalFirst) { +TEST_F(SchedulerStateTest, GetReadyTasksBatchDrainsSharedQueue) { alignas(64) PTO2TaskSlotState slot_a, slot_b; - init_slot(slot_a, PTO2_TASK_PENDING, 0, 1); + // fanin_count = 1 so a single release_fanin_and_check_ready call drives each + // slot to ready (new_refcount 0->1 == fanin_count) and enqueues it. + init_slot(slot_a, PTO2_TASK_PENDING, 1, 1); init_slot(slot_b, PTO2_TASK_PENDING, 1, 1); - PTO2TaskSlotState *local_buf_storage[4]; - PTO2LocalReadyBuffer local_buf; - local_buf.reset(local_buf_storage, 4); - local_buf.try_push(&slot_a); - - // Use src API to route slot_b into the global ready queue - sched.release_fanin_and_check_ready(slot_b); + // Route both slots into the global ready queue via the src API. + ASSERT_TRUE(sched.release_fanin_and_check_ready(slot_a)); + ASSERT_TRUE(sched.release_fanin_and_check_ready(slot_b)); PTO2TaskSlotState *out[4]; - int count = sched.get_ready_tasks_batch(PTO2ResourceShape::AIC, local_buf, out, 4); + int count = sched.get_ready_tasks_batch(PTO2ResourceShape::AIC, out, 4); EXPECT_EQ(count, 2); - // Local buffer drains first (LIFO), so slot_a comes first + // Shared queue is FIFO, so slot_a (pushed first) comes first. EXPECT_EQ(out[0], &slot_a); EXPECT_EQ(out[1], &slot_b); } diff --git a/tests/ut/cpp/a5/test_ready_queue.cpp b/tests/ut/cpp/a5/test_ready_queue.cpp index 886b613be5..5cf411d75f 100644 --- a/tests/ut/cpp/a5/test_ready_queue.cpp +++ b/tests/ut/cpp/a5/test_ready_queue.cpp @@ -9,10 +9,9 @@ * ----------------------------------------------------------------------------------------------------------- */ /** - * Unit tests for PTO2ReadyQueue and PTO2LocalReadyBuffer from pto_scheduler.h + * Unit tests for PTO2ReadyQueue from pto_scheduler.h * - * Tests the lock-free bounded MPMC queue (Vyukov design) and the thread-local - * ready buffer used for local-first dispatch optimization. + * Tests the lock-free bounded MPMC queue (Vyukov design). * * Design contracts: * @@ -30,10 +29,6 @@ * - size() relaxed ordering: size() reads both positions with * memory_order_relaxed and is a hint, not a snapshot. If a stale read * produces d > e the guard returns 0. - * - * - LocalReadyBuffer LIFO dispatch: try_push appends at count++, pop returns - * slot_states[--count]. LIFO reversal is intentional for cache-locality - * when a producer immediately dispatches its fanout. */ #include @@ -406,71 +401,3 @@ INSTANTIATE_TEST_SUITE_P( MPMCConfig{4, 4, 1250} // HighContentionStress ) ); - -// ============================================================================= -// LocalReadyBuffer -// ============================================================================= - -class LocalReadyBufferTest : public ::testing::Test { -protected: - static constexpr int CAPACITY = 8; - - PTO2LocalReadyBuffer buffer; - PTO2TaskSlotState *backing[CAPACITY]; - - void SetUp() override { buffer.reset(backing, CAPACITY); } -}; - -// --- Normal path --- - -TEST_F(LocalReadyBufferTest, PopEmptyReturnsNullptr) { EXPECT_EQ(buffer.pop(), nullptr); } - -// LIFO dispatch: try_push appends at count++, pop returns slot_states[--count]. -TEST_F(LocalReadyBufferTest, LIFOOrdering) { - PTO2TaskSlotState a, b; - - ASSERT_TRUE(buffer.try_push(&a)); - ASSERT_TRUE(buffer.try_push(&b)); - - EXPECT_EQ(buffer.pop(), &b); - EXPECT_EQ(buffer.pop(), &a); - EXPECT_EQ(buffer.pop(), nullptr); -} - -TEST_F(LocalReadyBufferTest, TryPushFullReturnsFalse) { - PTO2TaskSlotState items[CAPACITY + 1]; - - for (int i = 0; i < CAPACITY; i++) { - ASSERT_TRUE(buffer.try_push(&items[i])); - } - - EXPECT_FALSE(buffer.try_push(&items[CAPACITY])); -} - -TEST_F(LocalReadyBufferTest, ResetSetsCleanState) { - EXPECT_EQ(buffer.pop(), nullptr) << "Fresh buffer is empty"; - - PTO2TaskSlotState a, b; - ASSERT_TRUE(buffer.try_push(&a)); - ASSERT_TRUE(buffer.try_push(&b)); - - buffer.reset(backing, CAPACITY); - EXPECT_EQ(buffer.pop(), nullptr) << "Buffer is empty after reset"; - - PTO2TaskSlotState items[CAPACITY]; - for (int i = 0; i < CAPACITY; i++) { - EXPECT_TRUE(buffer.try_push(&items[i])); - } - EXPECT_FALSE(buffer.try_push(&a)) << "Full after pushing capacity items post-reset"; -} - -// --- Boundary conditions --- - -TEST_F(LocalReadyBufferTest, NullBackingBuffer) { - PTO2LocalReadyBuffer buf; - buf.reset(nullptr, 0); - - PTO2TaskSlotState item{}; - EXPECT_FALSE(buf.try_push(&item)) << "Push fails with null backing"; - EXPECT_EQ(buf.pop(), nullptr) << "Pop returns null with null backing"; -} diff --git a/tests/ut/cpp/a5/test_scheduler_state.cpp b/tests/ut/cpp/a5/test_scheduler_state.cpp index 8f7f0b514e..ede85564e5 100644 --- a/tests/ut/cpp/a5/test_scheduler_state.cpp +++ b/tests/ut/cpp/a5/test_scheduler_state.cpp @@ -178,27 +178,25 @@ TEST_F(SchedulerStateTest, ScopeEndBatchRelease) { } // ============================================================================= -// get_ready_tasks_batch: local buffer first +// get_ready_tasks_batch: drains the shared ready queue // ============================================================================= -TEST_F(SchedulerStateTest, GetReadyTasksBatchLocalFirst) { +TEST_F(SchedulerStateTest, GetReadyTasksBatchDrainsSharedQueue) { alignas(64) PTO2TaskSlotState slot_a, slot_b; - init_slot(slot_a, PTO2_TASK_PENDING, 0, 1); + // fanin_count = 1 so a single release_fanin_and_check_ready call drives each + // slot to ready (new_refcount 0->1 == fanin_count) and enqueues it. + init_slot(slot_a, PTO2_TASK_PENDING, 1, 1); init_slot(slot_b, PTO2_TASK_PENDING, 1, 1); - PTO2TaskSlotState *local_buf_storage[4]; - PTO2LocalReadyBuffer local_buf; - local_buf.reset(local_buf_storage, 4); - local_buf.try_push(&slot_a); - - // Use src API to route slot_b into the global ready queue - sched.release_fanin_and_check_ready(slot_b); + // Route both slots into the global ready queue via the src API. + ASSERT_TRUE(sched.release_fanin_and_check_ready(slot_a)); + ASSERT_TRUE(sched.release_fanin_and_check_ready(slot_b)); PTO2TaskSlotState *out[4]; - int count = sched.get_ready_tasks_batch(PTO2ResourceShape::AIC, local_buf, out, 4); + int count = sched.get_ready_tasks_batch(PTO2ResourceShape::AIC, out, 4); EXPECT_EQ(count, 2); - // Local buffer drains first (LIFO), so slot_a comes first + // Shared queue is FIFO, so slot_a (pushed first) comes first. EXPECT_EQ(out[0], &slot_a); EXPECT_EQ(out[1], &slot_b); } diff --git a/tools/benchmark_rounds.sh b/tools/benchmark_rounds.sh index 9f36568161..1f818381e6 100755 --- a/tools/benchmark_rounds.sh +++ b/tools/benchmark_rounds.sh @@ -36,9 +36,23 @@ PROJECT_ROOT="$(cd "$SCRIPT_DIR/.." && pwd)" # --- tensormap_and_ringbuffer --- declare -A TMR_EXAMPLE_CASES=( [alternating_matmul_add]="Case1" + [benchmark_bgemm]="Case0" + [paged_attention_unroll]="Case1,Case2" + [paged_attention_unroll_manual_scope]="Case1,Case2" + [batch_paged_attention]="Case1" + # spmd_paged_attention temporarily disabled: pre-existing onboard stall + # (507018 S1:running-stalled), reproduces on baseline — see KNOWN_ISSUES.md. + # [spmd_paged_attention]="Case1,Case2" + [qwen3_14b_decode]="StressBatch16Seq3500" ) TMR_EXAMPLE_ORDER=( alternating_matmul_add + benchmark_bgemm + paged_attention_unroll + paged_attention_unroll_manual_scope + batch_paged_attention + # spmd_paged_attention # temporarily disabled — see KNOWN_ISSUES.md + qwen3_14b_decode ) # ---------------------------------------------------------------------------