From 9429177faa191d877f93ab09cb63d6e41b960c0c Mon Sep 17 00:00:00 2001 From: Leaf-Salix <2503954024@qq.com> Date: Mon, 27 Jul 2026 15:24:20 +0800 Subject: [PATCH] Fix: make sync_start drain generation-safe without election - Tag per-thread tree acknowledgements with the drain attempt - Use scheduler thread 0 as the fixed staging coordinator - Remove the deterministic runtime hook and retain hook-free regression coverage - Preserve the 64-byte shared-state layout and record the investigation --- docs/dynamic-linking.md | 4 +- .../2026-07-sync-start-drain-retry-aba.md | 237 ++++++++++++++++++ docs/investigations/README.md | 1 + .../runtime/scheduler/scheduler_cold_path.cpp | 8 +- .../scheduler/scheduler_completion.cpp | 117 ++++----- .../runtime/scheduler/scheduler_context.h | 1 + .../runtime/scheduler/scheduler_dispatch.cpp | 2 +- .../runtime/scheduler/scheduler_types.h | 29 ++- .../docs/RUNTIME_LOGIC.md | 17 +- .../runtime/scheduler/scheduler_cold_path.cpp | 8 +- .../scheduler/scheduler_completion.cpp | 122 +++++---- .../runtime/scheduler/scheduler_context.h | 1 + .../runtime/scheduler/scheduler_dispatch.cpp | 2 +- .../runtime/scheduler/scheduler_types.h | 29 ++- .../docs/RUNTIME_LOGIC.md | 42 ++-- .../runtime/scheduler/scheduler_cold_path.cpp | 8 +- .../scheduler/scheduler_completion.cpp | 67 +++-- .../runtime/scheduler/scheduler_context.h | 1 + .../runtime/scheduler/scheduler_types.h | 23 +- tests/ut/cpp/a2a3/test_scheduler_state.cpp | 27 ++ tests/ut/cpp/a5/test_scheduler_state.cpp | 27 ++ tests/ut/cpp/common/test_error_code_names.cpp | 2 +- 22 files changed, 552 insertions(+), 223 deletions(-) create mode 100644 docs/investigations/2026-07-sync-start-drain-retry-aba.md diff --git a/docs/dynamic-linking.md b/docs/dynamic-linking.md index fb2b629060..04cd24037f 100644 --- a/docs/dynamic-linking.md +++ b/docs/dynamic-linking.md @@ -223,8 +223,8 @@ SchedulerContext owns its own teardown: - `SchedulerContext::deinit()` resets every scheduler-owned field — per-core states, payloads, sync-start drain coordination - (`sync_start_pending` / `drain_worker_elected` / `drain_ack_mask` / - `pending_task`), task counters, worker-id lists, + (`sync_start_pending` / `drain_attempt` / `drain_ack_tokens_` / + `pending_task` / parallel-stage state), task counters, worker-id lists, core trackers, `cores_total_num_` / `aic_count_` / `aiv_count_`, `regs_`, `sched_`, `func_id_to_addr_`, and the `pto2_init_*` flags. - `AicpuExecutor::deinit()` calls `sched_ctx_.deinit()` first, then resets diff --git a/docs/investigations/2026-07-sync-start-drain-retry-aba.md b/docs/investigations/2026-07-sync-start-drain-retry-aba.md new file mode 100644 index 0000000000..52b9e09ea2 --- /dev/null +++ b/docs/investigations/2026-07-sync-start-drain-retry-aba.md @@ -0,0 +1,237 @@ +# `sync_start` drain retry ABA across reusable barrier state + +**Date**: 2026-07-27 +**Verdict**: fixed by tagging drain attempts, reducing acknowledgements through +a generation-tagged tree, and using scheduler thread 0 as the coordinator. The +deterministic runtime hook was used to validate the interleaving and then +removed; hook-free unit and scene tests remain as regression coverage. + +## Question + +Issue [#1455](https://github.com/hw-native-sys/simpler/issues/1455) +identified an ABA race in the `sync_start` drain retry protocol. The original +protocol reused these values across attempts without a generation: + +- `drain_ack_mask` +- `drain_worker_elected` +- `drain_stage_go` +- `drain_stage_done_mask` + +When the elected drain worker found fewer available resources than the pending +task required, it cleared the acknowledgement and election state so scheduler +threads could resume completion polling. A thread from the old attempt could +already have crossed the acknowledgement barrier and then be delayed before +election. If it resumed after the reset, every value it observed was locally +valid, but some values belonged to the next attempt. + +The failure ended in a split wait: the stale owner waited for the old attempt's +`stage_done` mask while its peers waited for the new attempt's `ack_mask`. +Neither side could make progress. + +The investigation asked: + +1. Can that stale-participant interleaving be reproduced deterministically? +2. Which attempt-tagged protocol prevents state from different retries from + being combined? +3. Can the fix avoid adding a shared atomic hot spot or a per-spin O(N) scan? + +## What was tried + +### Deterministic runtime hook + +The first hook was tested on upstream commit +`d0bc661a5b41b4925f2d72c77e86228cc351cf23`. It was enabled with +`SIMPLER_DRAIN_ABA_TEST=1` and used four AICPU threads, which give the +tensormap-and-ringbuffer runtime three active scheduler threads. + +The scene submitted: + +1. one long-running MIX holder block; +2. one 24-block MIX task with `sync_start`. + +The holder left only 23 clusters available, forcing the first drain attempt to +retry. The hook parked scheduler thread 1 after the old attempt's +acknowledgement barrier but before election. After another scheduler reset the +drain state, the hook released thread 1 into the original protocol. Once the +resource count reached 24, that stale thread could become the owner and publish +`stage_go` using state assembled from two attempts. + +The important property was that the hook did not synthesize the final timeout. +It only selected the interleaving; the original election and staging protocol +created the split barrier. + +The hook was adapted as the production protocol changed. In the final +generation-tagged version it: + +- published an independent per-thread arrival token; +- parked thread 1 after the acknowledgement barrier; +- waited for another scheduler to advance `drain_attempt`; +- waited until the peer schedulers had acknowledged the new attempt; +- released thread 1 while it still held the old local attempt. + +The fixed protocol then had to reject that stale local attempt before it could +coordinate staging. A dedicated `NOT_EXERCISED` failure made the test fail if +the resource retry never occurred, preventing a silent pass where the intended +interleaving was not reached. + +### Coordination designs + +The following production designs were built and measured: + +| Design | Coordination shape | Outcome | +| ------ | ------------------ | ------- | +| Per-thread generation tokens with O(N) scan | Every scheduler publishes independently; the coordinator scans all tokens | Correct, but O(N) work on each barrier spin | +| Packed generation and ack mask | One shared atomic stores the attempt and mask | Rejected because shared LL/SC updates contended | +| One-way tree without root broadcast | Fixed coordinator; followers proceed without waiting for the root token | Rejected because polling moved to the shared `stage_go` cache line | +| Generation-tagged tree with rotating coordinator | Tree reduction plus `attempt % N` coordinator | No stable performance benefit | +| Generation-tagged tree with fixed coordinator | Tree reduction and root broadcast; scheduler thread 0 coordinates staging | Retained | + +The retained tree has one token per scheduler. Each scheduler waits for at most +two child subtree tokens, publishes one subtree token, and waits for the root +token before staging. Fan-in work per scheduler is O(1), the critical path is +O(log N), and no shared read-modify-write operation is required. + +Scheduler indices are contiguous in all three runtime variants, so thread 0 is +present whenever a scheduler exists. Fixing the coordinator also removes the +election state and prevents a stale scheduler from becoming an owner in a later +attempt. + +### Validation and measurement + +Correctness checks covered the A2/A3 tensormap-and-ringbuffer and +host-build-graph runtimes and the A5 tensormap-and-ringbuffer runtime. +Generation-token unit tests were added for both architecture families. + +Hardware scene tests covered: + +- the deterministic ABA hook; +- normal and stress `sync_start` drains; +- one, two, three, and four active schedulers; +- a non-power-of-two three-scheduler acknowledgement tree. + +Performance comparisons used one locked A2/A3 device and ABBA ordering. Each +ABBA segment ran 100 rounds. The final stress follow-up ran five independent +`A100-B100-B100-A100` cycles, giving 1,000 samples per version. + +## Result + +### Reproduced failure + +The original hook run reached the intended cross-attempt state: + +```text +attempt 1 owner: available=23 < block_num=24, reset for retry +stale thread 1: local_attempt=1, current_attempt=2 +stale thread 1: publishes stage_go with ack_mask=0x5 +threads 0 and 2: wait for attempt 2 ack_mask=0x7 +stale thread 1: waits for attempt 1 stage_done=0x7 +``` + +The run then failed with AICPU timeout `507018`. This established that the +generation-less reusable fields could combine state from different attempts; +it was not only a theoretical stale-load concern. + +### Correctness after the fix + +With the generation-tagged tree and fixed coordinator: + +- the deterministic hook passed on A2/A3 hardware; +- normal and stress `sync_start` scene tests passed; +- one-, two-, three-, and four-scheduler configurations passed; +- A2/A3 and A5 scheduler-state unit tests passed; +- static review found no cross-attempt election, staging, teardown, or + memory-ordering gap. + +The final fixed-coordinator hardware tasks included: + +- ABA hook: `task_20260727_031658_60345132353` +- three-scheduler normal: `task_20260727_031726_6337306055` +- three-scheduler stress: `task_20260727_031758_65330616500` +- two-scheduler MIX spill: `task_20260727_031827_66165821039` +- one-scheduler normal: `task_20260727_031859_67079319390` +- four-scheduler host-build-graph: `task_20260727_031941_68308727426` + +### Performance + +Relative to the earlier PR version, the rejected alternatives changed mean +Device time as follows: + +| Design | Normal | Stress | +| ------ | -----: | -----: | +| Packed shared atomic | +16.42% | +19.02% | +| Tree without root broadcast | +10.34% | +7.60% | +| Fixed coordinator | -5.67% | +1.51% | +| Rotating coordinator | +3.02% | -1.32% | + +Positive values are slower. The rotating result did not show a consistent win +across normal and stress cases, while the fixed coordinator removed election +work and preserved the tree's distributed polling. + +After rebasing, a direct 100-round ABBA comparison against upstream runtime +commit `a53cc168` measured: + +| Case | Device | Effective | Scheduler | 10% trimmed Device | +| ---- | -----: | --------: | --------: | -----------------: | +| Normal | -0.91% | -0.83% | -0.52% | -0.14% | +| Stress | -2.93% | -3.02% | -2.99% | -4.56% | + +The final PR was subsequently rebased from `a53cc168` to `1c475a70`; the only +upstream difference was a CI workflow edit, so the runtime source and measured +binary were unchanged. + +The five-cycle, 1,000-sample stress follow-up compared upstream `1c475a70` +against PR commit `34a250e`. Its pooled Effective time was 4.07% slower, but +the five paired deltas ranged from -3.25% to +18.90%. Treating each process +cycle as the correlated unit gave a 95% interval of [-5.41%, +14.80%]. The +stress distribution was multimodal, so neither the earlier approximately 3% +improvement nor the later approximately 4% regression was established as a +stable effect. + +Plain non-drain cases ranged from -3.19% to +1.80% across Device, Effective, +Orchestrator, Scheduler, and trimmed Device measurements, with no consistent +regression. + +## Why not (now) + +The generation-less acknowledgement and election protocol is not safe to +restore: the deterministic run demonstrated a reachable split barrier, and the +attempt tag directly distinguishes the state that the old protocol conflated. + +The packed atomic and no-root-broadcast designs are not retained because they +introduced clear performance regressions. Rotation added state and +generation-dependent ownership without a stable gain. + +The deterministic hook itself was valuable during the investigation, but its +permanent implementation required an environment-controlled behavior gate, +runtime launch plumbing, scheduler hot-path branches, a production error code, +and test-only token scanning. The hook was A2/A3-specific while the production +fix spans three runtime variants. Once the interleaving and fix were recorded, +the permanent regression barrier was better provided by hook-free +generation-token unit tests plus normal and stress scene tests. + +## When to reconsider + +- A future scheduler change reintroduces reusable drain coordination state that + is not tagged by `drain_attempt`. +- A failure shows that unit tests and repeated normal/stress scene tests do not + cover a new cross-attempt transition. +- The project gains a fault-injection framework that can suspend scheduler + participants without adding product configuration, status codes, or + test-only branches to the runtime. +- Hardware with a substantially larger scheduler count makes the O(log N) + tree visible in profiles; compare a distributed tree against a contention- + free alternative on that hardware before changing the protocol. + +## References + +- [Issue #1455](https://github.com/hw-native-sys/simpler/issues/1455) - + original ABA analysis, deterministic reproducer, and split-barrier logs. +- [PR #1461](https://github.com/hw-native-sys/simpler/pull/1461) - production + fix, review discussion, and performance follow-ups. +- Upstream reproduction commit: + `d0bc661a5b41b4925f2d72c77e86228cc351cf23`. +- Final measured upstream base: + `1c475a7016f0bbc8a3d6cb7515acff6c14769905`. +- Final pre-hook-removal PR commit: + `34a250e4fa5a65a40ad2e839c87ab54f920f1623`. +- [Environment-variable and macro gating](../../.claude/rules/env-macro-gating.md). diff --git a/docs/investigations/README.md b/docs/investigations/README.md index 70d2ad53d5..87eb55d16c 100644 --- a/docs/investigations/README.md +++ b/docs/investigations/README.md @@ -84,6 +84,7 @@ that ...". Newest first. +- [2026-07 — `sync_start` drain retry ABA across reusable barrier state](2026-07-sync-start-drain-retry-aba.md) — reproduced #1455 deterministically by suspending an old-attempt scheduler between the ack barrier and election; retained a generation-tagged O(log N) tree with fixed thread-0 coordination after packed-atomic, no-root-broadcast, and rotating-coordinator experiments; removed the invasive runtime hook after hook-free UT/ST coverage was established - [2026-07 — Host worker dispatch latency: where the remaining ~50 µs goes](2026-07-host-dispatch-latency-budget.md) — measured & dropped: after #1499 a `Worker.run()` costs **43.3 µs fixed + 8.08 µs per task**, so the "94% of latency is our runtime, not the IPC" headline is an artifact of benchmarking a *single* task — at 256 tasks/run the fixed cost is 2%. Also settles #1498 against itself for latency: a pipe wake measures 4.4 µs one way against a 3.5 µs bare fork+shm round trip, so blocking *adds* more than the whole IPC floor to every dispatch; only the CPU-while-idle argument survives - [2026-07 — Containing A2/A3 SDMA stream teardown after an AICore fault](2026-07-a2a3-sdma-fault-teardown.md) — CANN exposes no remote-channel retirement fence: even device-confirmed stream frees followed by a soft-reset success that did not complete context teardown can add minutes to a later ordinary fault. Contained by making SDMA an explicit per-Worker opt-in (`enable_sdma`) so only opt-in Workers carry the slow teardown and ordinary Workers keep fast recovery; full closure needs CANN runtime + driver changes (#1425) - [2026-07 — AICore-side arg fill for ALL dispatches (not just the gated path)](2026-07-aicore-fills-all-args-ready-path.md) — measured & rejected for the ready path: making the AICore fill its own `args[]` for every task adds ~1.0 µs to each task's `receive→start` setup (paged_attention_unroll: 349 ns → 1356 ns), because a ready task has no idle doorbell gate to hide the fill. Offload is a win only on `not_ready` (early-dispatch), where the AICore already spins at the gate; shipped design keeps the AICPU filling ready tasks (#1328) diff --git a/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_cold_path.cpp b/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_cold_path.cpp index 3cc67f03bd..e41fd19f32 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_cold_path.cpp +++ b/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_cold_path.cpp @@ -983,10 +983,12 @@ void SchedulerContext::deinit() { // the count-gated consumer never reads entries[] past the fresh count. // Reset sync-start drain coordination — a previous run that aborted mid-drain - // would otherwise leave dirty pending/elected/ack state for the next reuse. + // would otherwise leave dirty pending/ack state for the next reuse. drain_state_.sync_start_pending.store(0, std::memory_order_release); - drain_state_.drain_worker_elected.store(0, std::memory_order_release); - drain_state_.drain_ack_mask.store(0, std::memory_order_release); + drain_state_.drain_attempt.store(0, std::memory_order_release); + for (int32_t t = 0; t < MAX_AICPU_THREADS; t++) { + drain_ack_tokens_[t].store(0, std::memory_order_release); + } drain_state_.drain_stage_go.store(0, std::memory_order_release); drain_state_.drain_stage_done_mask.store(0, std::memory_order_release); drain_state_.drain_running_staged.store(0, std::memory_order_release); diff --git a/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_completion.cpp b/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_completion.cpp index 96f17c18d9..03bba86f15 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_completion.cpp +++ b/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_completion.cpp @@ -467,20 +467,20 @@ void SchedulerContext::check_running_cores_for_completion( // Returns false if another thread already holds drain; caller must re-push slot_state. // // Two-phase protocol: CAS 0 -> -1 (sentinel) to claim ownership, store task and -// reset election flag, then release-store block_num. Other threads acquire-load +// reset staging state, then release-store block_num. Other threads acquire-load // sync_start_pending; seeing block_num > 0 ensures all relaxed stores are visible. bool SchedulerContext::enter_drain_mode(PTO2TaskSlotState *slot_state, int32_t block_num) { int32_t expected = 0; if (!drain_state_.sync_start_pending.compare_exchange_strong( - expected, -1, std::memory_order_relaxed, std::memory_order_relaxed + expected, -1, std::memory_order_acquire, std::memory_order_relaxed )) { return false; // Another thread already holds the drain slot. } - // We own the drain slot. Store the task and reset the coordination flags before making - // it visible. + // Advance the attempt before publishing the new task so delayed participants + // from the previous round cannot satisfy this round's ack tree. + uint64_t next_attempt = sync_start_drain_next_attempt(drain_state_.drain_attempt.load(std::memory_order_relaxed)); + drain_state_.drain_attempt.store(next_attempt, std::memory_order_release); drain_state_.pending_task.store(slot_state, std::memory_order_release); - drain_state_.drain_ack_mask.store(0, std::memory_order_relaxed); - drain_state_.drain_worker_elected.store(0, std::memory_order_relaxed); drain_state_.drain_stage_go.store(0, std::memory_order_relaxed); drain_state_.drain_stage_done_mask.store(0, std::memory_order_relaxed); drain_state_.drain_running_staged.store(0, std::memory_order_relaxed); @@ -639,21 +639,22 @@ SchedulerContext::drain_stage_cores(PTO2TaskSlotState *slot_state, int32_t block // Called by each scheduler thread when drain_state_.sync_start_pending != 0. // // Protocol: -// 1. Ack barrier: all threads signal they've stopped dispatch, spin until all acked. -// If this thread's ack bit gets cleared while waiting, a reset occurred -- return. -// 2. Election + availability: one thread wins the CAS. It checks global resources; if -// insufficient it resets ack/election so all threads resume completion polling to free -// cores, then retry. If sufficient it releases parallel staging (stage_go). +// 1. Ack barrier: all threads signal they've stopped dispatch through an O(log N) +// tree. Thread 0 publishes the completed root token before followers continue. +// Each ack carries the current attempt, so an attempt change invalidates the old cohort. +// 2. Availability: thread 0 checks global resources. If insufficient it advances the +// attempt so all threads resume completion polling and retry. If sufficient it +// releases parallel staging (stage_go). // 3. Parallel stage: EVERY thread stages its OWN cores concurrently (CAS-claimed block // indices), accumulates its running-slot cores, and marks its stage_done bit. -// 4. Finalize: the elected thread waits for all stage_done bits, seeds the rendezvous +// 4. Finalize: the coordinator waits for all stage_done bits, seeds the rendezvous // (running_slot_count) for a gated drain, and reopens the gate -// (a release-store the non-elected threads acquire, so the seed is visible before any -// completion promotes a pending block). Non-elected threads spin until the gate reopens. +// (a release-store the followers acquire, so the seed is visible before any +// completion promotes a pending block). Followers spin until the gate reopens. void SchedulerContext::handle_drain_mode(int32_t thread_idx, [[maybe_unused]] uint64_t *out_stage_wall_cycles) { #if SIMPLER_DFX bool drain_prof = (l2_swimlane_level_ >= L2SwimlaneLevel::SCHED_PHASES && out_stage_wall_cycles != nullptr); - uint64_t drain_acked_ts = 0; // set at ack-barrier end; used to measure the stage wall + uint64_t drain_acked_ts = 0; // set immediately before staging; used to measure the stage wall #endif // Every spin in this function honors is_completed(): once the run latches // completed_ (all tasks done, or a fatal error raised elsewhere), peers leave @@ -674,41 +675,35 @@ void SchedulerContext::handle_drain_mode(int32_t thread_idx, [[maybe_unused]] ui if (block_num == 0) return; uint32_t all_acked = (1u << active_sched_threads_) - 1; + uint64_t drain_attempt = drain_state_.drain_attempt.load(std::memory_order_acquire); + uint64_t subtree_token = sync_start_drain_ack_subtree_token(drain_attempt); - // Ack barrier -- signal this thread has stopped dispatch. - drain_state_.drain_ack_mask.fetch_or(1u << thread_idx, std::memory_order_release); - - // Spin until all threads have acked. - // If our bit is cleared while waiting, elected reset due to insufficient resources. - while (true) { - if (is_completed()) return; - uint32_t ack = drain_state_.drain_ack_mask.load(std::memory_order_acquire); - if ((ack & all_acked) == all_acked) break; - if ((ack & (1u << thread_idx)) == 0) return; - SPIN_WAIT_HINT(); + // O(log N) reduction with O(1) fan-out per thread and no shared RMW. + for (int32_t child : {thread_idx * 2 + 1, thread_idx * 2 + 2}) { + if (child >= active_sched_threads_) continue; + while (drain_ack_tokens_[child].load(std::memory_order_acquire) != subtree_token) { + if (is_completed()) return; + if (drain_state_.drain_attempt.load(std::memory_order_acquire) != drain_attempt) return; + SPIN_WAIT_HINT(); + } } - // Election -- exactly one thread wins the CAS. - int32_t expected = 0; - drain_state_.drain_worker_elected.compare_exchange_strong( - expected, thread_idx + 1, std::memory_order_acquire, std::memory_order_relaxed - ); - bool elected = drain_state_.drain_worker_elected.load(std::memory_order_relaxed) == thread_idx + 1; + drain_ack_tokens_[thread_idx].store(subtree_token, std::memory_order_release); + if (thread_idx != 0) { + while (drain_ack_tokens_[0].load(std::memory_order_acquire) != subtree_token) { + if (is_completed()) return; + if (drain_state_.drain_attempt.load(std::memory_order_acquire) != drain_attempt) return; + SPIN_WAIT_HINT(); + } + } + bool coordinator = thread_idx == 0; PTO2TaskSlotState *slot_state = drain_state_.pending_task.load(std::memory_order_acquire); // OWNER is acquired before the drain is published and persists through // completion, so every staging thread makes the same gate decision even if // producer release changes early_dispatch_state during the barrier. - bool gated = slot_state != nullptr && slot_state->payload != nullptr && - PTO2SchedulerState::owns_early_sync_drain(*slot_state->payload); - - if (elected) { - if (slot_state == nullptr) { - // pending_task observed null only when a concurrent drain completion already cleared - // it. Stale-elected: release the election lock and return. Do NOT clear drain_ack_mask - // / sync_start_pending -- a *new* drain run may already be accumulating acks. - drain_state_.drain_worker_elected.store(0, std::memory_order_release); - return; - } + bool gated = slot_state->payload != nullptr && PTO2SchedulerState::owns_early_sync_drain(*slot_state->payload); + + if (coordinator) { PTO2ResourceShape shape = slot_state->active_mask.to_shape(); // A gated drain may pre-stage onto pending slots too (idle+pending); the ready drain // needs block_num idle cores/clusters. @@ -716,8 +711,7 @@ void SchedulerContext::handle_drain_mode(int32_t thread_idx, [[maybe_unused]] ui count_global_available(shape, slot_state->active_mask.core_mask(), /*include_pending=*/gated); if (available < block_num) { // Insufficient -- reset so all threads resume completion polling to free cores, then retry. - drain_state_.drain_ack_mask.store(0, std::memory_order_release); - drain_state_.drain_worker_elected.store(0, std::memory_order_release); + drain_state_.drain_attempt.store(sync_start_drain_next_attempt(drain_attempt), std::memory_order_release); return; } // Release parallel staging: every thread (this one included) now stages its own cores. @@ -725,16 +719,13 @@ void SchedulerContext::handle_drain_mode(int32_t thread_idx, [[maybe_unused]] ui drain_state_.drain_stage_done_mask.store(0, std::memory_order_relaxed); drain_state_.drain_stage_go.store(1, std::memory_order_release); } else { - // Non-elected: wait for the go signal, or bail if the elected thread reset (stale / - // insufficient resources). + // Followers wait for the go signal, or bail if the coordinator + // advanced the attempt after finding insufficient resources. while (drain_state_.drain_stage_go.load(std::memory_order_acquire) == 0) { if (is_completed()) return; - if (drain_state_.drain_worker_elected.load(std::memory_order_acquire) == 0) return; + if (drain_state_.drain_attempt.load(std::memory_order_acquire) != drain_attempt) return; SPIN_WAIT_HINT(); } - slot_state = drain_state_.pending_task.load(std::memory_order_acquire); - if (slot_state == nullptr) return; - gated = slot_state->payload != nullptr && PTO2SchedulerState::owns_early_sync_drain(*slot_state->payload); } // Parallel stage this thread's own cores (CAS-claimed block indices), then mark done. @@ -750,20 +741,19 @@ void SchedulerContext::handle_drain_mode(int32_t thread_idx, [[maybe_unused]] ui drain_state_.drain_running_staged.fetch_add(my_running, std::memory_order_acq_rel); drain_state_.drain_stage_done_mask.fetch_or(1u << thread_idx, std::memory_order_release); - if (!elected) { - // Non-elected: staging done; wait for the elected thread to reopen the gate. Exiting via - // sync_start_pending==0 (release/acquire) or drain_worker_elected==0 both synchronize - // with the elected's finalize (its release fence sequences the seed before both stores), - // so the running_slot_count seed is visible before this thread resumes completions. + if (!coordinator) { + // Follower staging is done; wait for the coordinator to reopen the gate. + // sync_start_pending==0 synchronizes with finalize, so the running_slot_count seed is + // visible before this thread resumes completions. while (drain_state_.sync_start_pending.load(std::memory_order_acquire) != 0) { if (is_completed()) return; - if (drain_state_.drain_worker_elected.load(std::memory_order_acquire) == 0) return; + if (drain_state_.drain_attempt.load(std::memory_order_acquire) != drain_attempt) return; SPIN_WAIT_HINT(); } return; } - // Elected: wait for all threads to finish staging, then seed the rendezvous and reopen. + // Coordinator: wait for all threads to finish staging, then seed the rendezvous and reopen. while ((drain_state_.drain_stage_done_mask.load(std::memory_order_acquire) & all_acked) != all_acked) { if (is_completed()) return; SPIN_WAIT_HINT(); @@ -778,16 +768,15 @@ void SchedulerContext::handle_drain_mode(int32_t thread_idx, [[maybe_unused]] ui ); } // Clear drain state and reopen the gate FIRST, so the other threads resume immediately. - // Release fence sequences the seed + tracker mutations before every clear, so any thread - // that acquire-observes one of them (sync_start_pending==0 / drain_worker_elected==0) sees - // the seed. `slot_state` is a local holding the fa_fused slot (not drain_state_), so it stays - // valid for the propagate below even if a new drain reuses pending_task after reopen. + // Release fence sequences the seed + tracker mutations before sync_start_pending. The + // attempt remains published while the gate is open; the next drain owner advances it + // behind the -1 sentinel. Followers identify reopen by gate-open or attempt change. + // `slot_state` is a local holding the fa_fused slot (not drain_state_), so it stays valid for + // the propagate below even if a new drain reuses pending_task after reopen. std::atomic_thread_fence(std::memory_order_release); drain_state_.pending_task.store(nullptr, std::memory_order_release); drain_state_.drain_stage_go.store(0, std::memory_order_relaxed); drain_state_.drain_stage_done_mask.store(0, std::memory_order_relaxed); - drain_state_.drain_ack_mask.store(0, std::memory_order_relaxed); - drain_state_.drain_worker_elected.store(0, std::memory_order_relaxed); drain_state_.sync_start_pending.store(0, std::memory_order_release); // Recheck after publishing the drain seed. The producer-side rendezvous check can race diff --git a/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_context.h b/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_context.h index 78f2eb1cc9..76901dcdf6 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_context.h +++ b/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_context.h @@ -138,6 +138,7 @@ class SchedulerContext { // sync_start drain coordination SyncStartDrainState drain_state_; + std::atomic drain_ack_tokens_[MAX_AICPU_THREADS]{}; #if SIMPLER_DFX SchedL2SwimlaneCounters sched_l2_swimlane_[MAX_AICPU_THREADS]; diff --git a/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_dispatch.cpp b/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_dispatch.cpp index 17fc4738ed..29c90acf24 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_dispatch.cpp +++ b/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_dispatch.cpp @@ -1102,7 +1102,7 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ handle_drain_mode(thread_idx, &drain_stage_wall); // Record a Drain bar only when this thread actually did drain work (reached // drain_stage_cores). The many no-op entries — ack + availability-insufficient - // reset, stale-elected, non-elected bail before stage_go — never stage, so they + // reset or follower bail before stage_go — never stage, so they // would otherwise clutter the lane with zero-work drain(0) bars. if (l2_swimlane_level_ >= L2SwimlaneLevel::SCHED_PHASES && drain_stage_wall != 0) { l2_swimlane_aicpu_record_sched_phase( diff --git a/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_types.h b/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_types.h index 98e9654f01..91d2d226c5 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_types.h +++ b/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_types.h @@ -12,6 +12,7 @@ #define SCHEDULER_TYPES_H #include +#include #include #include "common/core_type.h" @@ -555,22 +556,36 @@ struct alignas(64) SchedL2SwimlaneCounters { // ============================================================================= // When sync_start_pending != 0, all scheduler threads skip dispatch -// (only process completions) until the drain worker finishes launching all blocks. +// (only process completions) until the fixed coordinator finishes launching all blocks. struct alignas(64) SyncStartDrainState { - std::atomic sync_start_pending{0}; // 0=normal; -1=initializing; >0=active (value=block_num) - std::atomic drain_worker_elected{0}; // 0=none; >0: elected thread's (thread_idx+1) - std::atomic drain_ack_mask{0}; // bit per thread; all-set = all threads reached ack barrier + std::atomic sync_start_pending{0}; // 0=normal; -1=initializing; >0=active (value=block_num) std::atomic pending_task{nullptr}; // held task (not re-queued) - // Parallel staging: after the elected thread confirms global availability it sets + std::atomic drain_attempt{0}; // incremented whenever an ack round is reset + // Parallel staging: after the coordinator confirms global availability it sets // stage_go, releasing every thread to stage its OWN cores concurrently (vs the old // single-thread serial fill). Each thread ORs its bit into stage_done_mask when it - // finishes and accumulates its running-slot cores into running_staged; the elected + // finishes and accumulates its running-slot cores into running_staged; the coordinator // thread waits for all bits, seeds the rendezvous, and reopens the gate. - std::atomic drain_stage_go{0}; // 0=hold; 1=elected released parallel staging + std::atomic drain_stage_go{0}; // 0=hold; 1=coordinator released parallel staging std::atomic drain_stage_done_mask{0}; // bit per thread; all-set = all threads done staging std::atomic drain_running_staged{0}; // sum of running-slot cores staged (rendezvous seed) int32_t _pad[7]; }; static_assert(sizeof(SyncStartDrainState) == 64); +static_assert(offsetof(SyncStartDrainState, pending_task) == 8); +static_assert(offsetof(SyncStartDrainState, drain_attempt) == 16); +static_assert(offsetof(SyncStartDrainState, drain_stage_go) == 24); + +constexpr uint64_t SYNC_START_DRAIN_ACK_SUBTREE_READY = uint64_t{1} << 63; +constexpr uint64_t SYNC_START_DRAIN_ATTEMPT_MASK = ~SYNC_START_DRAIN_ACK_SUBTREE_READY; + +inline uint64_t sync_start_drain_next_attempt(uint64_t attempt) { + uint64_t next = (attempt + 1) & SYNC_START_DRAIN_ATTEMPT_MASK; + return next == 0 ? 1 : next; +} + +inline uint64_t sync_start_drain_ack_subtree_token(uint64_t attempt) { + return attempt | SYNC_START_DRAIN_ACK_SUBTREE_READY; +} #endif // SCHEDULER_TYPES_H diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md b/src/a2a3/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md index fb3810f54d..99fd3ee517 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md @@ -647,19 +647,20 @@ feed the same drain. A sync_start cohort of `block_num` cores must occupy all its cores before any of them run. When it cannot fit inline, `enter_drain_mode` arms a stop-the-world drain: -1. **Single election** — a CAS on `sync_start_pending` (0 → −1) makes drains mutually +1. **Single ownership** — a CAS on `sync_start_pending` (0 → −1) makes drains mutually exclusive; only one cohort drains at a time, regardless of source. -2. **All-or-nothing** — the elected thread checks `count_global_available >= block_num` - *before* staging; if short it aborts (stages nothing) and retries after completions free - cores. A cohort is fully staged or not at all — never partial. -3. **Parallel stage** — all threads barrier, then each CAS-claims a block range and stages - its own cores with a non-zero `src_payload` gate: idle cores → running slots, busy cores → - pending slots. +2. **All-or-nothing** — scheduler thread 0 coordinates the generation-tagged ack tree and + checks `count_global_available >= block_num` *before* staging. If short, it advances the + attempt and retries after completions free cores. A cohort is fully staged or not at all — + never partial. +3. **Parallel stage** — after thread 0 broadcasts the completed root token, each scheduler + thread CAS-claims a block range and stages its own cores with a non-zero `src_payload` + gate: idle cores → running slots, busy cores → pending slots. 4. **Rendezvous launch** — `running_slot_count` counts staged running-slot cores; when it reaches `popcount(staged_core_mask)` **and** the producer has released, `maybe_rendezvous_ring` rings every gated core's doorbell together — the cohort starts as one. -Single-election + all-or-nothing make the drain deadlock-free across multiple cohorts: at most +Single ownership + all-or-nothing make the drain deadlock-free across multiple cohorts: at most one drains, and it fully stages or waits, so two cohorts can never each half-occupy the cluster set (see the completion path's `pending_gated` classification for why a promoted-but-still-gated block is not mistaken for a normal task). diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_cold_path.cpp b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_cold_path.cpp index 52d760e585..17290920dd 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_cold_path.cpp +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_cold_path.cpp @@ -1310,10 +1310,12 @@ void SchedulerContext::deinit() { // the count-gated consumer never reads entries[] past the fresh count. // Reset sync-start drain coordination — a previous run that aborted mid-drain - // would otherwise leave dirty pending/elected/ack state for the next reuse. + // would otherwise leave dirty pending/ack state for the next reuse. drain_state_.sync_start_pending.store(0, std::memory_order_release); - drain_state_.drain_worker_elected.store(0, std::memory_order_release); - drain_state_.drain_ack_mask.store(0, std::memory_order_release); + drain_state_.drain_attempt.store(0, std::memory_order_release); + for (int32_t t = 0; t < MAX_AICPU_THREADS; t++) { + drain_ack_tokens_[t].store(0, std::memory_order_release); + } drain_state_.drain_stage_go.store(0, std::memory_order_release); drain_state_.drain_stage_done_mask.store(0, std::memory_order_release); drain_state_.drain_running_staged.store(0, std::memory_order_release); 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 15d2b60cd9..1eab9dde5f 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 @@ -11,6 +11,7 @@ #include "scheduler_context.h" #include +#include #include "common/unified_log.h" #include "aicpu/device_time.h" @@ -496,20 +497,20 @@ void SchedulerContext::check_running_cores_for_completion( // Returns false if another thread already holds drain; caller must re-push slot_state. // // Two-phase protocol: CAS 0 -> -1 (sentinel) to claim ownership, store task and -// reset election flag, then release-store block_num. Other threads acquire-load +// reset staging state, then release-store block_num. Other threads acquire-load // sync_start_pending; seeing block_num > 0 ensures all relaxed stores are visible. bool SchedulerContext::enter_drain_mode(PTO2TaskSlotState *slot_state, int32_t block_num) { int32_t expected = 0; if (!drain_state_.sync_start_pending.compare_exchange_strong( - expected, -1, std::memory_order_relaxed, std::memory_order_relaxed + expected, -1, std::memory_order_acquire, std::memory_order_relaxed )) { return false; // Another thread already holds the drain slot. } - // We own the drain slot. Store the task and reset the coordination flags before making - // it visible. + // Advance the attempt before publishing the new task so delayed participants + // from the previous round cannot satisfy this round's ack tree. + uint64_t next_attempt = sync_start_drain_next_attempt(drain_state_.drain_attempt.load(std::memory_order_relaxed)); + drain_state_.drain_attempt.store(next_attempt, std::memory_order_release); drain_state_.pending_task.store(slot_state, std::memory_order_release); - drain_state_.drain_ack_mask.store(0, std::memory_order_relaxed); - drain_state_.drain_worker_elected.store(0, std::memory_order_relaxed); drain_state_.drain_stage_go.store(0, std::memory_order_relaxed); drain_state_.drain_stage_done_mask.store(0, std::memory_order_relaxed); drain_state_.drain_running_staged.store(0, std::memory_order_relaxed); @@ -668,21 +669,22 @@ SchedulerContext::drain_stage_cores(PTO2TaskSlotState *slot_state, int32_t block // Called by each scheduler thread when drain_state_.sync_start_pending != 0. // // Protocol: -// 1. Ack barrier: all threads signal they've stopped dispatch, spin until all acked. -// If this thread's ack bit gets cleared while waiting, a reset occurred -- return. -// 2. Election + availability: one thread wins the CAS. It checks global resources; if -// insufficient it resets ack/election so all threads resume completion polling to free -// cores, then retry. If sufficient it releases parallel staging (stage_go). +// 1. Ack barrier: all threads signal they've stopped dispatch through an O(log N) +// tree. Thread 0 publishes the completed root token before followers continue. +// Each ack carries the current attempt, so an attempt change invalidates the old cohort. +// 2. Availability: thread 0 checks global resources. If insufficient it advances the +// attempt so all threads resume completion polling and retry. If sufficient it +// releases parallel staging (stage_go). // 3. Parallel stage: EVERY thread stages its OWN cores concurrently (CAS-claimed block // indices), accumulates its running-slot cores, and marks its stage_done bit. -// 4. Finalize: the elected thread waits for all stage_done bits, seeds the rendezvous +// 4. Finalize: the coordinator waits for all stage_done bits, seeds the rendezvous // (running_slot_count) for a gated drain, and reopens the gate -// (a release-store the non-elected threads acquire, so the seed is visible before any -// completion promotes a pending block). Non-elected threads spin until the gate reopens. +// (a release-store the followers acquire, so the seed is visible before any +// completion promotes a pending block). Followers spin until the gate reopens. void SchedulerContext::handle_drain_mode(int32_t thread_idx, [[maybe_unused]] uint64_t *out_stage_wall_cycles) { #if SIMPLER_DFX bool drain_prof = (l2_swimlane_level_ >= L2SwimlaneLevel::SCHED_PHASES && out_stage_wall_cycles != nullptr); - uint64_t drain_acked_ts = 0; // set at ack-barrier end; used to measure the stage wall + uint64_t drain_acked_ts = 0; // set immediately before staging; used to measure the stage wall #endif // Every spin in this function honors is_completed(): once the run latches // completed_ (all tasks done, or a fatal error raised elsewhere), peers leave @@ -703,41 +705,37 @@ void SchedulerContext::handle_drain_mode(int32_t thread_idx, [[maybe_unused]] ui if (block_num == 0) return; uint32_t all_acked = (1u << active_sched_threads_) - 1; - - // Ack barrier -- signal this thread has stopped dispatch. - drain_state_.drain_ack_mask.fetch_or(1u << thread_idx, std::memory_order_release); - - // Spin until all threads have acked. - // If our bit is cleared while waiting, elected reset due to insufficient resources. - while (true) { - if (is_completed()) return; - uint32_t ack = drain_state_.drain_ack_mask.load(std::memory_order_acquire); - if ((ack & all_acked) == all_acked) break; - if ((ack & (1u << thread_idx)) == 0) return; - SPIN_WAIT_HINT(); + uint64_t drain_attempt = drain_state_.drain_attempt.load(std::memory_order_acquire); + uint64_t subtree_token = sync_start_drain_ack_subtree_token(drain_attempt); + + // Production uses one store per thread: an O(log N) reduction with O(1) + // fan-out per thread and no shared RMW. + for (int32_t child : {thread_idx * 2 + 1, thread_idx * 2 + 2}) { + if (child >= active_sched_threads_) continue; + while (drain_ack_tokens_[child].load(std::memory_order_acquire) != subtree_token) { + if (is_completed()) return; + if (drain_state_.drain_attempt.load(std::memory_order_acquire) != drain_attempt) return; + SPIN_WAIT_HINT(); + } + } + drain_ack_tokens_[thread_idx].store(subtree_token, std::memory_order_release); + if (thread_idx != 0) { + while (drain_ack_tokens_[0].load(std::memory_order_acquire) != subtree_token) { + if (is_completed()) return; + if (drain_state_.drain_attempt.load(std::memory_order_acquire) != drain_attempt) return; + SPIN_WAIT_HINT(); + } } - // Election -- exactly one thread wins the CAS. - int32_t expected = 0; - drain_state_.drain_worker_elected.compare_exchange_strong( - expected, thread_idx + 1, std::memory_order_acquire, std::memory_order_relaxed - ); - bool elected = drain_state_.drain_worker_elected.load(std::memory_order_relaxed) == thread_idx + 1; + + bool coordinator = thread_idx == 0; PTO2TaskSlotState *slot_state = drain_state_.pending_task.load(std::memory_order_acquire); // OWNER is acquired before the drain is published and persists through // completion, so every staging thread makes the same gate decision even if // producer release changes early_dispatch_state during the barrier. - bool gated = slot_state != nullptr && slot_state->payload != nullptr && - PTO2SchedulerState::owns_early_sync_drain(*slot_state->payload); - - if (elected) { - if (slot_state == nullptr) { - // pending_task observed null only when a concurrent drain completion already cleared - // it. Stale-elected: release the election lock and return. Do NOT clear drain_ack_mask - // / sync_start_pending -- a *new* drain run may already be accumulating acks. - drain_state_.drain_worker_elected.store(0, std::memory_order_release); - return; - } + bool gated = slot_state->payload != nullptr && PTO2SchedulerState::owns_early_sync_drain(*slot_state->payload); + + if (coordinator) { PTO2ResourceShape shape = slot_state->active_mask.to_shape(); // A gated drain may pre-stage onto pending slots too (idle+pending); the ready drain // needs block_num idle cores/clusters. @@ -745,8 +743,7 @@ void SchedulerContext::handle_drain_mode(int32_t thread_idx, [[maybe_unused]] ui count_global_available(shape, slot_state->active_mask.core_mask(), /*include_pending=*/gated); if (available < block_num) { // Insufficient -- reset so all threads resume completion polling to free cores, then retry. - drain_state_.drain_ack_mask.store(0, std::memory_order_release); - drain_state_.drain_worker_elected.store(0, std::memory_order_release); + drain_state_.drain_attempt.store(sync_start_drain_next_attempt(drain_attempt), std::memory_order_release); return; } // Release parallel staging: every thread (this one included) now stages its own cores. @@ -754,16 +751,13 @@ void SchedulerContext::handle_drain_mode(int32_t thread_idx, [[maybe_unused]] ui drain_state_.drain_stage_done_mask.store(0, std::memory_order_relaxed); drain_state_.drain_stage_go.store(1, std::memory_order_release); } else { - // Non-elected: wait for the go signal, or bail if the elected thread reset (stale / - // insufficient resources). + // Followers wait for the go signal, or bail if the coordinator + // advanced the attempt after finding insufficient resources. while (drain_state_.drain_stage_go.load(std::memory_order_acquire) == 0) { if (is_completed()) return; - if (drain_state_.drain_worker_elected.load(std::memory_order_acquire) == 0) return; + if (drain_state_.drain_attempt.load(std::memory_order_acquire) != drain_attempt) return; SPIN_WAIT_HINT(); } - slot_state = drain_state_.pending_task.load(std::memory_order_acquire); - if (slot_state == nullptr) return; - gated = slot_state->payload != nullptr && PTO2SchedulerState::owns_early_sync_drain(*slot_state->payload); } // Parallel stage this thread's own cores (CAS-claimed block indices), then mark done. @@ -779,20 +773,19 @@ void SchedulerContext::handle_drain_mode(int32_t thread_idx, [[maybe_unused]] ui drain_state_.drain_running_staged.fetch_add(my_running, std::memory_order_acq_rel); drain_state_.drain_stage_done_mask.fetch_or(1u << thread_idx, std::memory_order_release); - if (!elected) { - // Non-elected: staging done; wait for the elected thread to reopen the gate. Exiting via - // sync_start_pending==0 (release/acquire) or drain_worker_elected==0 both synchronize - // with the elected's finalize (its release fence sequences the seed before both stores), - // so the running_slot_count seed is visible before this thread resumes completions. + if (!coordinator) { + // Follower staging is done; wait for the coordinator to reopen the gate. + // sync_start_pending==0 synchronizes with finalize, so the running_slot_count seed is + // visible before this thread resumes completions. while (drain_state_.sync_start_pending.load(std::memory_order_acquire) != 0) { if (is_completed()) return; - if (drain_state_.drain_worker_elected.load(std::memory_order_acquire) == 0) return; + if (drain_state_.drain_attempt.load(std::memory_order_acquire) != drain_attempt) return; SPIN_WAIT_HINT(); } return; } - // Elected: wait for all threads to finish staging, then seed the rendezvous and reopen. + // Coordinator: wait for all threads to finish staging, then seed the rendezvous and reopen. while ((drain_state_.drain_stage_done_mask.load(std::memory_order_acquire) & all_acked) != all_acked) { if (is_completed()) return; SPIN_WAIT_HINT(); @@ -807,16 +800,15 @@ void SchedulerContext::handle_drain_mode(int32_t thread_idx, [[maybe_unused]] ui ); } // Clear drain state and reopen the gate FIRST, so the other threads resume immediately. - // Release fence sequences the seed + tracker mutations before every clear, so any thread - // that acquire-observes one of them (sync_start_pending==0 / drain_worker_elected==0) sees - // the seed. `slot_state` is a local holding the fa_fused slot (not drain_state_), so it stays - // valid for the propagate below even if a new drain reuses pending_task after reopen. + // Release fence sequences the seed + tracker mutations before sync_start_pending. The + // attempt remains published while the gate is open; the next drain owner advances it + // behind the -1 sentinel. Followers identify reopen by gate-open or attempt change. + // `slot_state` is a local holding the fa_fused slot (not drain_state_), so it stays valid for + // the propagate below even if a new drain reuses pending_task after reopen. std::atomic_thread_fence(std::memory_order_release); drain_state_.pending_task.store(nullptr, std::memory_order_release); drain_state_.drain_stage_go.store(0, std::memory_order_relaxed); drain_state_.drain_stage_done_mask.store(0, std::memory_order_relaxed); - drain_state_.drain_ack_mask.store(0, std::memory_order_relaxed); - drain_state_.drain_worker_elected.store(0, std::memory_order_relaxed); drain_state_.sync_start_pending.store(0, std::memory_order_release); // Recheck after publishing the drain seed. The producer-side rendezvous check can race 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 a6f316173f..cc1f27ae6e 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 @@ -164,6 +164,7 @@ class SchedulerContext { // sync_start drain coordination SyncStartDrainState drain_state_; + std::atomic drain_ack_tokens_[MAX_AICPU_THREADS]{}; #if SIMPLER_DFX SchedL2SwimlaneCounters sched_l2_swimlane_[MAX_AICPU_THREADS]; 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 00bf111d0c..ac3a840975 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 @@ -1146,7 +1146,7 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ handle_drain_mode(thread_idx, &drain_stage_wall); // Record a Drain bar only when this thread actually did drain work (reached // drain_stage_cores). The many no-op entries — ack + availability-insufficient - // reset, stale-elected, non-elected bail before stage_go — never stage, so they + // reset or follower bail before stage_go — never stage, so they // would otherwise clutter the lane with zero-work drain(0) bars. if (l2_swimlane_level_ >= L2SwimlaneLevel::SCHED_PHASES && drain_stage_wall != 0) { l2_swimlane_aicpu_record_sched_phase( diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_types.h b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_types.h index c1ccf4a022..a4ee4493ca 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_types.h +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_types.h @@ -12,6 +12,7 @@ #define SCHEDULER_TYPES_H #include +#include #include #include "common/core_type.h" @@ -545,22 +546,36 @@ struct alignas(64) SchedL2SwimlaneCounters { // ============================================================================= // When sync_start_pending != 0, all scheduler threads skip dispatch -// (only process completions) until the drain worker finishes launching all blocks. +// (only process completions) until the fixed coordinator finishes launching all blocks. struct alignas(64) SyncStartDrainState { - std::atomic sync_start_pending{0}; // 0=normal; -1=initializing; >0=active (value=block_num) - std::atomic drain_worker_elected{0}; // 0=none; >0: elected thread's (thread_idx+1) - std::atomic drain_ack_mask{0}; // bit per thread; all-set = all threads reached ack barrier + std::atomic sync_start_pending{0}; // 0=normal; -1=initializing; >0=active (value=block_num) std::atomic pending_task{nullptr}; // held task (not re-queued) - // Parallel staging: after the elected thread confirms global availability it sets + std::atomic drain_attempt{0}; // incremented whenever an ack round is reset + // Parallel staging: after the coordinator confirms global availability it sets // stage_go, releasing every thread to stage its OWN cores concurrently (vs the old // single-thread serial fill). Each thread ORs its bit into stage_done_mask when it - // finishes and accumulates its running-slot cores into running_staged; the elected + // finishes and accumulates its running-slot cores into running_staged; the coordinator // thread waits for all bits, seeds the rendezvous, and reopens the gate. - std::atomic drain_stage_go{0}; // 0=hold; 1=elected released parallel staging + std::atomic drain_stage_go{0}; // 0=hold; 1=coordinator released parallel staging std::atomic drain_stage_done_mask{0}; // bit per thread; all-set = all threads done staging std::atomic drain_running_staged{0}; // sum of running-slot cores staged (rendezvous seed) int32_t _pad[7]; }; static_assert(sizeof(SyncStartDrainState) == 64); +static_assert(offsetof(SyncStartDrainState, pending_task) == 8); +static_assert(offsetof(SyncStartDrainState, drain_attempt) == 16); +static_assert(offsetof(SyncStartDrainState, drain_stage_go) == 24); + +constexpr uint64_t SYNC_START_DRAIN_ACK_SUBTREE_READY = uint64_t{1} << 63; +constexpr uint64_t SYNC_START_DRAIN_ATTEMPT_MASK = ~SYNC_START_DRAIN_ACK_SUBTREE_READY; + +inline uint64_t sync_start_drain_next_attempt(uint64_t attempt) { + uint64_t next = (attempt + 1) & SYNC_START_DRAIN_ATTEMPT_MASK; + return next == 0 ? 1 : next; +} + +inline uint64_t sync_start_drain_ack_subtree_token(uint64_t attempt) { + return attempt | SYNC_START_DRAIN_ACK_SUBTREE_READY; +} #endif // SCHEDULER_TYPES_H diff --git a/src/a5/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md b/src/a5/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md index 2c74179df2..f03fe21f03 100644 --- a/src/a5/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md +++ b/src/a5/runtime/tensormap_and_ringbuffer/docs/RUNTIME_LOGIC.md @@ -52,7 +52,8 @@ The primary production runtime. Uses ring buffers for task slots and output memo - **Task storage**: `PTO2TaskDescriptor[]` in shared memory ring buffer - **Memory**: GM Heap ring for output buffer allocation - **Dependencies**: automatically derived from tensor read/write patterns via TensorMap -- **Thread model**: 3 scheduler threads + 1 orchestrator thread on AICPU +- **Thread model**: `aicpu_thread_num - 1` scheduler threads + 1 orchestrator + thread on AICPU; the common four-thread configuration uses 3 schedulers - **Multi-ring**: HeapRing, TaskRing, and DepPool are split into `PTO2_MAX_RING_DEPTH` (4) independent instances for nested scope isolation. See [MULTI_RING.md](MULTI_RING.md) for details. - **Use case**: production workloads; supports streaming, flow control, and large batch sizes @@ -62,7 +63,7 @@ The primary production runtime. Uses ring buffers for task slots and output memo Two platform implementations exist under `src/platform/`, sharing a common interface. -### 2.1 a2a3 (Real Ascend Hardware) +### 2.1 a5 (Real Ascend Hardware) | Component | Description | | --------- | ----------- | @@ -72,7 +73,7 @@ Two platform implementations exist under `src/platform/`, sharing a common inter | `aicpu/kernel.cpp` | `DynTileFwkBackendKernelServer` entry → `aicpu_execute` | | `spin_hint.h` | ARM `wfe`/`yield` instructions for efficient spinning | -### 2.2 a2a3sim (Thread Simulation) +### 2.2 a5sim (Thread Simulation) | Component | Description | | --------- | ----------- | @@ -85,11 +86,11 @@ Two platform implementations exist under `src/platform/`, sharing a common inter | Constant | Value | Description | | -------- | ----- | ----------- | -| `PLATFORM_MAX_BLOCKDIM` | 24 | Maximum blocks (each = 1 AIC + 2 AIV) | -| `PLATFORM_MAX_AICPU_THREADS` | 4 | AICPU thread count (3 schedulers + 1 orchestrator) | -| `PLATFORM_MAX_AIC_PER_THREAD` | 24 | Max AIC cores per scheduler thread | -| `PLATFORM_MAX_AIV_PER_THREAD` | 48 | Max AIV cores per scheduler thread | -| `PLATFORM_PROF_SYS_CNT_FREQ` | 50 MHz | System counter frequency for profiling | +| `PLATFORM_MAX_BLOCKDIM` | 36 | Maximum blocks (each = 1 AIC + 2 AIV) | +| `PLATFORM_MAX_AICPU_THREADS` | 7 | Maximum total AICPU threads (up to 6 schedulers + 1 orchestrator) | +| `PLATFORM_MAX_AIC_PER_THREAD` | 36 | Max AIC cores per scheduler thread | +| `PLATFORM_MAX_AIV_PER_THREAD` | 72 | Max AIV cores per scheduler thread | +| `PLATFORM_PROF_SYS_CNT_FREQ` | 1000 MHz | System counter frequency for profiling | ### 2.4 Host Temporary Buffer @@ -524,13 +525,16 @@ verified by review. ### 8.1 Thread Model -With `aicpu_thread_num=4`, the AICPU runs 4 threads: +With `aicpu_thread_num=4`, the AICPU runs 4 threads. The core counts below +describe the 36-cluster target in §2.3; the runtime partitions the detected +cores across the scheduler threads, so smaller SKUs use proportionally smaller +slices. | Thread | Role | Cores | | ------ | ---- | ----- | -| 0 | Scheduler | 6 AIC + ~13 AIV | -| 1 | Scheduler | 6 AIC + ~13 AIV | -| 2 | Scheduler | 6 AIC + ~13 AIV | +| 0 | Scheduler | 12 AIC + 24 AIV | +| 1 | Scheduler | 12 AIC + 24 AIV | +| 2 | Scheduler | 12 AIC + 24 AIV | | 3 | Orchestrator | none | Core assignment: AICs and AIVs are divided equally among the 3 scheduler threads. @@ -645,12 +649,14 @@ unconditional true for hidden alloc creators). There is no auto-chain inheritanc A sync_start cohort of `block_num` cores must occupy all its cores before any of them run. When it cannot fit inline, `enter_drain_mode` arms a stop-the-world drain: -1. **Single election** — a CAS on `sync_start_pending` makes drains mutually exclusive. -2. **All-or-nothing** — the elected thread checks global available capacity ≥ `block_num` - before staging; if short it aborts and retries after completions free cores. -3. **Parallel stage** — threads barrier, then each CAS-claims a block range and stages its - own cores with a non-zero `src_payload` gate (`drain_stage_cores`): idle → running, - busy → pending (`pending_gated` when still waiting for the doorbell). +1. **Single ownership** — a CAS on `sync_start_pending` makes drains mutually exclusive. +2. **All-or-nothing** — scheduler thread 0 coordinates the generation-tagged ack tree and + checks global available capacity ≥ `block_num` before staging; if short it advances the + attempt and retries after completions free cores. +3. **Parallel stage** — after thread 0 broadcasts the completed root token, each scheduler + thread CAS-claims a block range and stages its own cores with a non-zero `src_payload` + gate (`drain_stage_cores`): idle → running, busy → pending (`pending_gated` when still + waiting for the doorbell). 4. **Rendezvous launch** — `running_slot_count` counts staged running-slot cores; when it reaches `popcount(staged_core_mask)` **and** the producer has released, `maybe_rendezvous_ring` rings every gated core's doorbell together — the cohort starts as one. diff --git a/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_cold_path.cpp b/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_cold_path.cpp index 6d4ad2cc1a..6df22543e4 100644 --- a/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_cold_path.cpp +++ b/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_cold_path.cpp @@ -1281,10 +1281,12 @@ void SchedulerContext::deinit() { // the count-gated consumer never reads entries[] past the fresh count. // Reset sync-start drain coordination — a previous run that aborted mid-drain - // would otherwise leave dirty pending/elected/ack state for the next reuse. + // would otherwise leave dirty pending/ack state for the next reuse. drain_state_.sync_start_pending.store(0, std::memory_order_release); - drain_state_.drain_worker_elected.store(0, std::memory_order_release); - drain_state_.drain_ack_mask.store(0, std::memory_order_release); + drain_state_.drain_attempt.store(0, std::memory_order_release); + for (int32_t t = 0; t < MAX_AICPU_THREADS; t++) { + drain_ack_tokens_[t].store(0, std::memory_order_release); + } drain_state_.pending_task.store(nullptr, std::memory_order_release); drain_state_.drain_stage_go.store(0, std::memory_order_release); drain_state_.drain_stage_done_mask.store(0, std::memory_order_release); 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 402956023d..dbc88af65d 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 @@ -410,19 +410,20 @@ void SchedulerContext::check_running_cores_for_completion( // Returns false if another thread already holds drain; caller must re-push slot_state. // // Two-phase protocol: CAS 0 -> -1 (sentinel) to claim ownership, store task and -// reset election flag, then release-store block_num. Other threads acquire-load +// reset staging state, then release-store block_num. Other threads acquire-load // sync_start_pending; seeing block_num > 0 ensures all relaxed stores are visible. bool SchedulerContext::enter_drain_mode(PTO2TaskSlotState *slot_state, int32_t block_num) { int32_t expected = 0; if (!drain_state_.sync_start_pending.compare_exchange_strong( - expected, -1, std::memory_order_relaxed, std::memory_order_relaxed + expected, -1, std::memory_order_acquire, std::memory_order_relaxed )) { return false; // Another thread already holds the drain slot. } - // We own the drain slot. Store the task and reset election flag before making it visible. + // Advance the attempt before publishing the new task so delayed participants + // from the previous round cannot satisfy this round's ack tree. + uint64_t next_attempt = sync_start_drain_next_attempt(drain_state_.drain_attempt.load(std::memory_order_relaxed)); + drain_state_.drain_attempt.store(next_attempt, std::memory_order_release); drain_state_.pending_task.store(slot_state, std::memory_order_release); - drain_state_.drain_ack_mask.store(0, std::memory_order_relaxed); - drain_state_.drain_worker_elected.store(0, std::memory_order_relaxed); drain_state_.drain_stage_go.store(0, std::memory_order_relaxed); drain_state_.drain_stage_done_mask.store(0, std::memory_order_relaxed); drain_state_.drain_running_staged.store(0, std::memory_order_relaxed); @@ -532,38 +533,37 @@ void SchedulerContext::handle_drain_mode(int32_t thread_idx, [[maybe_unused]] ui if (block_num == 0) return; uint32_t all_acked = (1u << active_sched_threads_) - 1; + uint64_t drain_attempt = drain_state_.drain_attempt.load(std::memory_order_acquire); + uint64_t subtree_token = sync_start_drain_ack_subtree_token(drain_attempt); - drain_state_.drain_ack_mask.fetch_or(1u << thread_idx, std::memory_order_release); - - while (true) { - if (is_completed()) return; - uint32_t ack = drain_state_.drain_ack_mask.load(std::memory_order_acquire); - if ((ack & all_acked) == all_acked) break; - if ((ack & (1u << thread_idx)) == 0) return; - SPIN_WAIT_HINT(); + // O(log N) reduction with O(1) fan-out per thread and no shared RMW. + for (int32_t child : {thread_idx * 2 + 1, thread_idx * 2 + 2}) { + if (child >= active_sched_threads_) continue; + while (drain_ack_tokens_[child].load(std::memory_order_acquire) != subtree_token) { + if (is_completed()) return; + if (drain_state_.drain_attempt.load(std::memory_order_acquire) != drain_attempt) return; + SPIN_WAIT_HINT(); + } } - - int32_t expected = 0; - drain_state_.drain_worker_elected.compare_exchange_strong( - expected, thread_idx + 1, std::memory_order_acquire, std::memory_order_relaxed - ); - bool elected = drain_state_.drain_worker_elected.load(std::memory_order_relaxed) == thread_idx + 1; + drain_ack_tokens_[thread_idx].store(subtree_token, std::memory_order_release); + if (thread_idx != 0) { + while (drain_ack_tokens_[0].load(std::memory_order_acquire) != subtree_token) { + if (is_completed()) return; + if (drain_state_.drain_attempt.load(std::memory_order_acquire) != drain_attempt) return; + SPIN_WAIT_HINT(); + } + } + bool coordinator = thread_idx == 0; PTO2TaskSlotState *slot_state = drain_state_.pending_task.load(std::memory_order_acquire); - bool gated = slot_state != nullptr && slot_state->payload != nullptr && - PTO2SchedulerState::owns_early_sync_drain(*slot_state->payload); + bool gated = slot_state->payload != nullptr && PTO2SchedulerState::owns_early_sync_drain(*slot_state->payload); - if (elected) { - if (slot_state == nullptr) { - drain_state_.drain_worker_elected.store(0, std::memory_order_release); - return; - } + if (coordinator) { PTO2ResourceShape shape = slot_state->active_mask.to_shape(); int32_t available = count_global_available(shape, slot_state->active_mask.core_mask(), /*include_pending=*/gated); if (available < block_num) { - drain_state_.drain_ack_mask.store(0, std::memory_order_release); - drain_state_.drain_worker_elected.store(0, std::memory_order_release); + drain_state_.drain_attempt.store(sync_start_drain_next_attempt(drain_attempt), std::memory_order_release); return; } drain_state_.drain_running_staged.store(0, std::memory_order_relaxed); @@ -572,12 +572,9 @@ void SchedulerContext::handle_drain_mode(int32_t thread_idx, [[maybe_unused]] ui } else { while (drain_state_.drain_stage_go.load(std::memory_order_acquire) == 0) { if (is_completed()) return; - if (drain_state_.drain_worker_elected.load(std::memory_order_acquire) == 0) return; + if (drain_state_.drain_attempt.load(std::memory_order_acquire) != drain_attempt) return; SPIN_WAIT_HINT(); } - slot_state = drain_state_.pending_task.load(std::memory_order_acquire); - if (slot_state == nullptr) return; - gated = slot_state->payload != nullptr && PTO2SchedulerState::owns_early_sync_drain(*slot_state->payload); } #if SIMPLER_DFX @@ -590,10 +587,10 @@ void SchedulerContext::handle_drain_mode(int32_t thread_idx, [[maybe_unused]] ui drain_state_.drain_running_staged.fetch_add(my_running, std::memory_order_acq_rel); drain_state_.drain_stage_done_mask.fetch_or(1u << thread_idx, std::memory_order_release); - if (!elected) { + if (!coordinator) { while (drain_state_.sync_start_pending.load(std::memory_order_acquire) != 0) { if (is_completed()) return; - if (drain_state_.drain_worker_elected.load(std::memory_order_acquire) == 0) return; + if (drain_state_.drain_attempt.load(std::memory_order_acquire) != drain_attempt) return; SPIN_WAIT_HINT(); } return; @@ -613,8 +610,6 @@ void SchedulerContext::handle_drain_mode(int32_t thread_idx, [[maybe_unused]] ui drain_state_.pending_task.store(nullptr, std::memory_order_release); drain_state_.drain_stage_go.store(0, std::memory_order_relaxed); drain_state_.drain_stage_done_mask.store(0, std::memory_order_relaxed); - drain_state_.drain_ack_mask.store(0, std::memory_order_relaxed); - drain_state_.drain_worker_elected.store(0, std::memory_order_relaxed); drain_state_.sync_start_pending.store(0, std::memory_order_release); if (gated) { 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 1f13d02aa9..2fed49b305 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 @@ -163,6 +163,7 @@ class SchedulerContext { // sync_start drain coordination SyncStartDrainState drain_state_; + std::atomic drain_ack_tokens_[MAX_AICPU_THREADS]{}; #if SIMPLER_DFX SchedL2SwimlaneCounters sched_l2_swimlane_[MAX_AICPU_THREADS]; diff --git a/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_types.h b/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_types.h index 1bc154c65d..d634b32a4c 100644 --- a/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_types.h +++ b/src/a5/runtime/tensormap_and_ringbuffer/runtime/scheduler/scheduler_types.h @@ -12,6 +12,7 @@ #define SCHEDULER_TYPES_H #include +#include #include #include "common/core_type.h" @@ -518,17 +519,31 @@ struct alignas(64) SchedL2SwimlaneCounters { // ============================================================================= // When sync_start_pending != 0, all scheduler threads skip dispatch -// (only process completions) until the drain worker finishes launching all blocks. +// (only process completions) until the fixed coordinator finishes launching all blocks. struct alignas(64) SyncStartDrainState { - std::atomic sync_start_pending{0}; // 0=normal; -1=initializing; >0=active (value=block_num) - std::atomic drain_worker_elected{0}; // 0=none; >0: elected thread's (thread_idx+1) - std::atomic drain_ack_mask{0}; // bit per thread; all-set = all threads reached ack barrier + std::atomic sync_start_pending{0}; // 0=normal; -1=initializing; >0=active (value=block_num) std::atomic pending_task{nullptr}; // held task (not re-queued) + std::atomic drain_attempt{0}; // incremented whenever an ack round is reset std::atomic drain_stage_go{0}; std::atomic drain_stage_done_mask{0}; std::atomic drain_running_staged{0}; int32_t _pad[7]; }; static_assert(sizeof(SyncStartDrainState) == 64); +static_assert(offsetof(SyncStartDrainState, pending_task) == 8); +static_assert(offsetof(SyncStartDrainState, drain_attempt) == 16); +static_assert(offsetof(SyncStartDrainState, drain_stage_go) == 24); + +constexpr uint64_t SYNC_START_DRAIN_ACK_SUBTREE_READY = uint64_t{1} << 63; +constexpr uint64_t SYNC_START_DRAIN_ATTEMPT_MASK = ~SYNC_START_DRAIN_ACK_SUBTREE_READY; + +inline uint64_t sync_start_drain_next_attempt(uint64_t attempt) { + uint64_t next = (attempt + 1) & SYNC_START_DRAIN_ATTEMPT_MASK; + return next == 0 ? 1 : next; +} + +inline uint64_t sync_start_drain_ack_subtree_token(uint64_t attempt) { + return attempt | SYNC_START_DRAIN_ACK_SUBTREE_READY; +} #endif // SCHEDULER_TYPES_H diff --git a/tests/ut/cpp/a2a3/test_scheduler_state.cpp b/tests/ut/cpp/a2a3/test_scheduler_state.cpp index b629230e76..e00ce44d4e 100644 --- a/tests/ut/cpp/a2a3/test_scheduler_state.cpp +++ b/tests/ut/cpp/a2a3/test_scheduler_state.cpp @@ -23,6 +23,33 @@ #include "scheduler/scheduler_types.h" #include "scheduler/pto_scheduler.h" +TEST(SyncStartDrainAttemptTest, LateAckCannotSatisfyNextAttemptBarrier) { + std::atomic ack_tokens[3]{}; + uint64_t old_attempt = 41; + uint64_t next_attempt = sync_start_drain_next_attempt(old_attempt); + uint64_t old_subtree = sync_start_drain_ack_subtree_token(old_attempt); + uint64_t next_subtree = sync_start_drain_ack_subtree_token(next_attempt); + + ack_tokens[1].store(old_subtree, std::memory_order_relaxed); + ack_tokens[2].store(old_subtree, std::memory_order_relaxed); + EXPECT_EQ(ack_tokens[1].load(std::memory_order_relaxed), old_subtree); + EXPECT_EQ(ack_tokens[2].load(std::memory_order_relaxed), old_subtree); + + // A late child token from attempt 41 cannot satisfy attempt 42's tree barrier. + ack_tokens[1].store(next_subtree, std::memory_order_relaxed); + ack_tokens[2].store(old_subtree, std::memory_order_relaxed); + EXPECT_EQ(ack_tokens[1].load(std::memory_order_relaxed), next_subtree); + EXPECT_NE(ack_tokens[2].load(std::memory_order_relaxed), next_subtree); + + ack_tokens[2].store(next_subtree, std::memory_order_relaxed); + ack_tokens[0].store(old_subtree, std::memory_order_relaxed); + EXPECT_NE(ack_tokens[0].load(std::memory_order_relaxed), next_subtree); + ack_tokens[0].store(next_subtree, std::memory_order_relaxed); + EXPECT_EQ(ack_tokens[0].load(std::memory_order_relaxed), next_subtree); + + EXPECT_EQ(sync_start_drain_next_attempt(SYNC_START_DRAIN_ATTEMPT_MASK), 1u); +} + class SchedulerStateTest : public ::testing::Test { protected: PTO2SchedulerState sched; diff --git a/tests/ut/cpp/a5/test_scheduler_state.cpp b/tests/ut/cpp/a5/test_scheduler_state.cpp index 6db6def8ca..5a11e65fdf 100644 --- a/tests/ut/cpp/a5/test_scheduler_state.cpp +++ b/tests/ut/cpp/a5/test_scheduler_state.cpp @@ -23,6 +23,33 @@ #include "scheduler/scheduler_types.h" #include "scheduler/pto_scheduler.h" +TEST(SyncStartDrainAttemptTest, LateAckCannotSatisfyNextAttemptBarrier) { + std::atomic ack_tokens[3]{}; + uint64_t old_attempt = 41; + uint64_t next_attempt = sync_start_drain_next_attempt(old_attempt); + uint64_t old_subtree = sync_start_drain_ack_subtree_token(old_attempt); + uint64_t next_subtree = sync_start_drain_ack_subtree_token(next_attempt); + + ack_tokens[1].store(old_subtree, std::memory_order_relaxed); + ack_tokens[2].store(old_subtree, std::memory_order_relaxed); + EXPECT_EQ(ack_tokens[1].load(std::memory_order_relaxed), old_subtree); + EXPECT_EQ(ack_tokens[2].load(std::memory_order_relaxed), old_subtree); + + // A late child token from attempt 41 cannot satisfy attempt 42's tree barrier. + ack_tokens[1].store(next_subtree, std::memory_order_relaxed); + ack_tokens[2].store(old_subtree, std::memory_order_relaxed); + EXPECT_EQ(ack_tokens[1].load(std::memory_order_relaxed), next_subtree); + EXPECT_NE(ack_tokens[2].load(std::memory_order_relaxed), next_subtree); + + ack_tokens[2].store(next_subtree, std::memory_order_relaxed); + ack_tokens[0].store(old_subtree, std::memory_order_relaxed); + EXPECT_NE(ack_tokens[0].load(std::memory_order_relaxed), next_subtree); + ack_tokens[0].store(next_subtree, std::memory_order_relaxed); + EXPECT_EQ(ack_tokens[0].load(std::memory_order_relaxed), next_subtree); + + EXPECT_EQ(sync_start_drain_next_attempt(SYNC_START_DRAIN_ATTEMPT_MASK), 1u); +} + class SchedulerStateTest : public ::testing::Test { protected: PTO2SchedulerState sched; diff --git a/tests/ut/cpp/common/test_error_code_names.cpp b/tests/ut/cpp/common/test_error_code_names.cpp index c27f9e55c9..e0e412693c 100644 --- a/tests/ut/cpp/common/test_error_code_names.cpp +++ b/tests/ut/cpp/common/test_error_code_names.cpp @@ -69,7 +69,7 @@ TEST(RuntimeErrorNames, NoErrorIsNotAnError) { // An unrecognised code must be reported as such. Annotating it with a neighbouring code's // text would be worse than printing the bare number. TEST(RuntimeErrorNames, UnknownCodeFallsBack) { - for (int32_t code : {6, 12, 42, 99, 104, 9999, -1}) { + for (int32_t code : {6, 12, 42, 99, 105, 9999, -1}) { EXPECT_STREQ(error_name(code), "unknown") << "code " << code; EXPECT_TRUE(is_blank(error_desc(code))) << "code " << code; EXPECT_TRUE(is_blank(error_hint(code))) << "code " << code;