From b69be27e72c18a1b268006c05b6654a15f1f600b Mon Sep 17 00:00:00 2001 From: Chao Wang <26245345+ChaoWao@users.noreply.github.com> Date: Fri, 14 Aug 2026 06:26:35 -0700 Subject: [PATCH] a2a3: submit every run on one AICore/AICPU stream pair MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A stream is an ordered queue and the execution claim is exclusive, so runs reach the device one at a time. A pipeline slot indexes the resources preparation mutates — arena banks, host Runtime buffers, retained temporaries — and preparing a run writes nothing to a stream. The per-slot stream sets were required only while every run created its own AICore stream, because a prepared successor then held a second live stream alongside its predecessor's. With reuse keyed on code publication that requirement is gone, so RunStreamSlots collapses to RunStreamPair: one pair, no slot index. The pair is readied by launch rather than by preparation. Launch is the first point the caller holds the execution claim, and a prepared successor overlaps its predecessor's execution, so replacing a stale AICore stream during preparation would destroy a stream the predecessor is still running on. For the same reason only the run that submitted the pair may retire it. A prepared successor that fails after preparation reaches cleanup_execution with an unproven retirement, which under a shared pair would otherwise destroy the live stream; the pair records its submitter and ignores anyone else. Readiness is gated on "a run still owns the pair" rather than on the previous run having completed, so a handle that survives a failed teardown stays stale and is replaced by the next launch instead of locking a reused runner out. The two streams stay distinct: the AICPU Run kernel spins in the handshake waiting for the AICore workers, so one queue would leave the AICore submission behind a spin that can never end. Co-Authored-By: Claude Opus 5 (1M context) --- docs/task-flow.md | 32 +- python/bindings/task_interface.cpp | 9 +- python/simpler/worker.py | 11 +- .../platform/onboard/host/device_runner.cpp | 74 ++- .../platform/onboard/host/device_runner.h | 42 +- .../platform/include/host/run_stream_pair.h | 219 ++++++++ .../platform/include/host/run_stream_slots.h | 204 ------- src/common/worker/chip_worker.h | 9 +- src/common/worker/pto_runtime_c_api.h | 11 +- .../test_native_run_lifecycle.py | 19 +- .../run_stream_reuse/test_run_stream_reuse.py | 73 ++- tests/ut/cpp/CMakeLists.txt | 16 +- .../cpp/hierarchical/test_run_stream_pair.cpp | 510 ++++++++++++++++++ .../hierarchical/test_run_stream_slots.cpp | 507 ----------------- 14 files changed, 878 insertions(+), 858 deletions(-) create mode 100644 src/common/platform/include/host/run_stream_pair.h delete mode 100644 src/common/platform/include/host/run_stream_slots.h create mode 100644 tests/ut/cpp/hierarchical/test_run_stream_pair.cpp delete mode 100644 tests/ut/cpp/hierarchical/test_run_stream_slots.cpp diff --git a/docs/task-flow.md b/docs/task-flow.md index ce6f445a00..eb8200c1f0 100644 --- a/docs/task-flow.md +++ b/docs/task-flow.md @@ -290,16 +290,28 @@ that window needs the admission layer to gate dispatch on `pool.owns(lease)` before handing work down, which is whole-run admission's job, not this layer's. -AICore streams are outside that lease. Registered callables keep content-hash -deduplicated GM allocations: simultaneously resident code images occupy -different allocations, while unregister frees an allocation that a later -registration may reuse. A dedup miss is the only path that publishes new -AICore instruction bytes. After that H2D copy succeeds, every retained slot -stream is marked stale; its next acquire destroys and recreates the AICore -stream before launch. Without a new publication, completed slot streams remain -reusable even when two resident callables alternate. Unproven completion still -destroys the stream conservatively. The AICPU stream carries no instruction -cache state and stays with its slot. +Run streams are outside that lease, and there is one pair of them per runner +rather than one per slot. A slot indexes the resources *preparation* mutates, +and preparing a run writes nothing to a stream: only launch submits, and launch +holds the exclusive execution claim, so runs reach the device one at a time and +the stream orders them. The two streams stay distinct because the AICPU Run +kernel spins in the handshake waiting for the AICore workers — one queue would +leave the AICore submission behind a spin that never ends. + +The AICore stream carries instruction-cache state. Registered callables keep +content-hash deduplicated GM allocations: simultaneously resident code images +occupy different allocations, while unregister frees an allocation that a later +registration may reuse. A dedup miss is the only repeatable path that publishes +new AICore instruction bytes; after that H2D copy succeeds the pair is marked +stale, and the next launch destroys the AICore stream and creates a replacement. +(The `kernel_entry` ELF that `rtRegisterAllKernel` publishes needs no such mark: +CANN offers no unregister, so it is registered once per runner and released with +the streams at finalize.) Without a new publication the pair stays warm even +when two resident callables alternate. Unproven completion still destroys the +AICore stream conservatively, and only the run that submitted the pair may +retire it — a prepared successor overlaps its predecessor's execution and must +leave the live pair alone. The AICPU stream carries no instruction cache state +and lives for the runner. #### Whole-run FIFO admission diff --git a/python/bindings/task_interface.cpp b/python/bindings/task_interface.cpp index 451145a230..e26c5cea63 100644 --- a/python/bindings/task_interface.cpp +++ b/python/bindings/task_interface.cpp @@ -2304,10 +2304,11 @@ NB_MODULE(_task_interface, m) { ) .def_prop_ro( "run_stream_set_create_count", &ChipWorker::run_stream_set_create_count, - "Number of AICore run streams the bound runner has created. The AICPU " - "stream belongs to a pipeline slot for the worker's lifetime. A completed " - "AICore stream is reused until a new code upload makes it stale. Platforms " - "using the persistent bootstrap pair report 0." + "Number of AICore run streams the bound runner has created. One AICPU " + "+ AICore pair serves every run for the runner's lifetime. The AICPU " + "stream persists; the AICore stream is recreated when a new code upload " + "makes it stale, and destroyed when an unproven completion retires it. " + "Platforms using the persistent bootstrap pair report 0." ) .def_prop_ro( "committed_device_memory", &ChipWorker::committed_device_memory, diff --git a/python/simpler/worker.py b/python/simpler/worker.py index c34884b2d6..623bfbdc8f 100644 --- a/python/simpler/worker.py +++ b/python/simpler/worker.py @@ -11131,10 +11131,13 @@ def host_dlopen_count(self) -> int: def run_stream_set_create_count(self) -> int: """L2 only: number of AICore run streams the runner has created. - AICPU streams belong to pipeline slots for the worker's lifetime. A - completed AICore stream is reused until a new code upload makes it - stale. Returns 0 on non-L2 workers and on platforms whose runs use the - persistent bootstrap stream pair (simulation, a5). + One AICPU + AICore pair serves every run for the runner's lifetime. The + AICPU stream persists; the AICore stream is recreated when a new code + upload makes it stale, and destroyed when an unproven completion retires + it, so this advances per publication or unproven retirement rather than + once per run or per pipeline slot. Returns 0 on non-L2 workers and on + platforms whose runs use the persistent bootstrap stream pair + (simulation, a5). """ if self.level != 2 or self._chip_worker is None: return 0 diff --git a/src/a2a3/platform/onboard/host/device_runner.cpp b/src/a2a3/platform/onboard/host/device_runner.cpp index 4d5bb55c2a..4d240932ed 100644 --- a/src/a2a3/platform/onboard/host/device_runner.cpp +++ b/src/a2a3/platform/onboard/host/device_runner.cpp @@ -217,29 +217,16 @@ void DeviceRunner::set_dep_gen_enabled(bool enable) { dep_gen_host_graph_set_enabled(enable); } -int DeviceRunner::provision_native_run_resources(uint32_t pipeline_slot) { - return ensure_run_stream_set(pipeline_slot); -} - -int DeviceRunner::abandon_native_run_resources(uint32_t pipeline_slot) { - return retire_run_aicore_stream(pipeline_slot, RunStreamSlots::CompletionStatus::Unproven); -} - int DeviceRunner::prepare_execution( Runtime &runtime, const CallConfig &config, uint32_t pipeline_slot, const NativeRunIdentity &identity, std::unique_ptr *prepared ) { if (prepared == nullptr || *prepared != nullptr) return -1; - const unsigned selected_pipeline_slot = pipeline_slot; auto execution = std::make_unique(identity, runtime, config, pipeline_slot); execution->resources_owned = true; auto prepare_rollback = RAIIScopeGuard([this, &execution]() { cleanup_execution(*execution, /*retire_aicore=*/false); }); - if (!run_stream_slots_.ready(selected_pipeline_slot)) { - LOG_ERROR("run stream set %u was not provisioned during native prepare", selected_pipeline_slot); - return -1; - } const int block_dim = runtime.get_worker_count() / cores_per_blockdim_; int launch_aicpu_num = config.aicpu_thread_num; // A prior AICore launch/sync error poisoned the device context and the @@ -444,8 +431,7 @@ int DeviceRunner::prepare_execution( int DeviceRunner::poll_execution(const ActiveExecution &active) { if (active.prepared == nullptr) return SIMPLER_NATIVE_RUN_POLL_ERROR; - const uint32_t pipeline_slot = active.prepared->pipeline_slot; - return run_stream_slots_.poll(pipeline_slot, [](void *aicpu, void *aicore) { + return run_streams_.poll([](void *aicpu, void *aicore) { return query_stream_pair_nonblocking(static_cast(aicpu), static_cast(aicore)); }); } @@ -453,12 +439,11 @@ int DeviceRunner::poll_execution(const ActiveExecution &active) { int DeviceRunner::drain_execution(ActiveExecution &active) { if (active.prepared == nullptr || !active.prepared->resources_owned) return -1; PreparedExecution &prepared = *active.prepared; - const uint32_t pipeline_slot = prepared.pipeline_slot; auto drain_cleanup = RAIIScopeGuard([this, &prepared]() { cleanup_execution(prepared, /*retire_aicore=*/true); }); - int rc = reap_run(pipeline_slot); + int rc = reap_run(); if (rc != 0) { // The device/sync error remains authoritative over teardown errors. return rc; @@ -468,7 +453,7 @@ int DeviceRunner::drain_execution(ActiveExecution &active) { // stale. Publish retirement so cleanup does not replace it with an // unproven state after the device result has already been established. prepared.aicore_retirement_attempted = true; - rc = retire_run_aicore_stream(pipeline_slot, RunStreamSlots::CompletionStatus::Complete); + rc = retire_run_aicore_stream(&prepared, RunStreamPair::CompletionStatus::Complete); if (rc != 0) return rc; // Reads device memory, so it must precede KernelArgs/runtime cleanup. @@ -507,7 +492,7 @@ void DeviceRunner::cleanup_execution(PreparedExecution &prepared, bool retire_ai } if (retire_aicore && !abandon && !prepared.aicore_retirement_attempted) { prepared.aicore_retirement_attempted = true; - (void)retire_run_aicore_stream(prepared.pipeline_slot, RunStreamSlots::CompletionStatus::Unproven); + (void)retire_run_aicore_stream(&prepared, RunStreamPair::CompletionStatus::Unproven); } prepared.resources_owned = false; } @@ -547,30 +532,30 @@ int DeviceRunner::create_run_stream(void **out) { return 0; } -int DeviceRunner::ensure_run_stream_set(unsigned slot) { - int rc = run_stream_slots_.acquire(slot); +int DeviceRunner::ensure_run_streams() { + int rc = run_streams_.ensure(); if (rc != 0) { - LOG_ERROR("ensure_run_stream_set(%u) failed: %d", slot, rc); + LOG_ERROR("ensure_run_streams failed: %d", rc); ACL_LOG_ERROR_DETAIL(rc); } return rc; } -int DeviceRunner::retire_run_aicore_stream(unsigned slot, RunStreamSlots::CompletionStatus completion_status) { - int rc = run_stream_slots_.retire_aicore(slot, completion_status); +int DeviceRunner::retire_run_aicore_stream(const void *owner, RunStreamPair::CompletionStatus completion_status) { + int rc = run_streams_.retire(completion_status, owner); if (rc != 0) { - LOG_ERROR("rtStreamDestroy (run AICore slot %u) failed: %d, slot is now unusable", slot, rc); + LOG_ERROR("rtStreamDestroy (run AICore stream) failed: %d, the handle is retained for teardown", rc); } return rc; } -int DeviceRunner::destroy_run_stream_sets() { +int DeviceRunner::destroy_run_streams() { // No pre-destroy sync, for the reason finalize_common() documents for the // bootstrap pair: rtStreamDestroy is the supported teardown for a stream // left in the error state by an op-timeout. - int rc = run_stream_slots_.destroy_all(); + int rc = run_streams_.destroy(); if (rc != 0) { - LOG_ERROR("destroy_run_stream_sets: a stream survived teardown: %d", rc); + LOG_ERROR("destroy_run_streams: a stream survived teardown: %d", rc); } return rc; } @@ -597,15 +582,18 @@ LaunchTransactionResult DeviceRunner::launch_run(PreparedExecution &prepared, La Runtime &runtime = *prepared.runtime; const int num_aicore = prepared.num_aicore; const int launch_aicpu_num = prepared.launch_aicpu_num; - const unsigned slot = prepared.pipeline_slot; // KernelLaunch is the pipeline boundary: this method clears the handshake // consumed by the launch and submits exactly the AICore and AICPU kernels. // It intentionally performs no stream synchronization or per-run cleanup. - if (!run_stream_slots_.ready(slot)) { - LOG_ERROR("launch_run: stream set %u is not ready", slot); + // + // The pair is readied here rather than at prepare because this is the first + // point the caller holds the execution claim: a prepared successor overlaps + // its predecessor's execution, so replacing a stale AICore stream during + // preparation would destroy a stream the predecessor is still running on. + if (ensure_run_streams() != 0) { return LaunchTransactionResult{}; } - RunStreamSet streams{run_stream_slots_.aicpu(slot), run_stream_slots_.aicore(slot)}; + RunStreamSet streams{static_cast(run_streams_.aicpu()), static_cast(run_streams_.aicore())}; LaunchTransactionResult result = exact_launch_transaction( prepared.identity, std::move(permit), [&]() { @@ -655,9 +643,9 @@ LaunchTransactionResult DeviceRunner::launch_run(PreparedExecution &prepared, La // Publishing query/drain ownership is a state change, so it sits past // the arming block: everything the block covers is rollback-free. - int rc = run_stream_slots_.mark_submitted(slot); + int rc = run_streams_.mark_submitted(&prepared); if (rc != 0) { - LOG_ERROR("launch_run: failed to publish stream set %u: %d", slot, rc); + LOG_ERROR("launch_run: failed to publish the run stream pair: %d", rc); return rc; } @@ -685,12 +673,12 @@ LaunchTransactionResult DeviceRunner::launch_run(PreparedExecution &prepared, La return result; } -int DeviceRunner::reap_run(unsigned slot) { - if (!run_stream_slots_.ready(slot)) { - LOG_ERROR("reap_run: invalid stream set %u", slot); +int DeviceRunner::reap_run() { + if (!run_streams_.ready()) { + LOG_ERROR("reap_run: the run stream pair is not ready"); return -1; } - int rc = sync_stream_pair(run_stream_slots_.aicpu(slot), run_stream_slots_.aicore(slot)); + int rc = sync_stream_pair(run_streams_.aicpu(), run_streams_.aicore()); if (rc != 0) { // The pair wait surfaces the AICore op-timeout (STARS-reaped op -> // 507000/507018/507046 at AICPU/AICore stream sync). The op-timeout @@ -1044,7 +1032,7 @@ int DeviceRunner::finalize() { ); } - run_stream_slots_.abandon_all(); + run_streams_.abandon(); int abandon_rc = abandon_common_after_device_failure(); // Only finalize the ACL owner after force reset established a clean @@ -1082,10 +1070,10 @@ int DeviceRunner::finalize() { // for the no-run-since-init case. finalize_collectors(); - // The per-run stream sets are this subclass's own RTS-owning member, so - // they are released here, while RTS is live and before the device reset - // below — the same window finalize_common() uses for the bootstrap pair. - int stream_rc = destroy_run_stream_sets(); + // The run stream pair is this subclass's own RTS-owning member, so it is + // released here, while RTS is live and before the device reset below — the + // same window finalize_common() uses for the bootstrap pair. + int stream_rc = destroy_run_streams(); // Shared cleanup body — streams, kernel_args, callable/orch maps, // chip-callable buffer pool, the three arenas, device_wall, diff --git a/src/a2a3/platform/onboard/host/device_runner.h b/src/a2a3/platform/onboard/host/device_runner.h index 0eb976299a..5bd030fbe2 100644 --- a/src/a2a3/platform/onboard/host/device_runner.h +++ b/src/a2a3/platform/onboard/host/device_runner.h @@ -38,7 +38,7 @@ #include "callable.h" #include "prepare_callable_common.h" #include "pto_runtime_c_api.h" // PTO_PIPELINE_MAX_DEPTH -#include "host/run_stream_slots.h" +#include "host/run_stream_pair.h" #include "common/kernel_args.h" #include "common/memory_barrier.h" #include "common/chip_swimlane_profiling.h" @@ -103,9 +103,10 @@ class DeviceRunner : public DeviceRunnerBase { int poll_execution(const ActiveExecution &active) override; int drain_execution(ActiveExecution &active) override; bool can_accept_run() const override { return !device_unusable_.load(std::memory_order_acquire); } - int provision_native_run_resources(uint32_t pipeline_slot) override; - int abandon_native_run_resources(uint32_t pipeline_slot) override; - void mark_run_streams_stale() override { run_stream_slots_.mark_all_stale(); } + // provision/abandon_native_run_resources keep the base no-op: preparation + // owns no stream, so there is nothing for a prepared run to provision or + // hand back. launch_run() readies the pair under the execution claim. + void mark_run_streams_stale() override { run_streams_.mark_stale(); } // Map/unmap a device buffer into host address space via // halHostRegister(DEV_SVM_MAP_HOST) / halHostUnregister. The returned host @@ -182,7 +183,7 @@ class DeviceRunner : public DeviceRunnerBase { // `aicpu_dlopen_count`, and `host_dlopen_count` are inherited from // `DeviceRunnerBase`. - size_t run_stream_set_create_count() const override { return run_stream_slots_.created_count(); } + size_t run_stream_set_create_count() const override { return run_streams_.created_count(); } private: // Most lifecycle state (device_id_, block_dim_, cores_per_blockdim_, @@ -223,14 +224,12 @@ class DeviceRunner : public DeviceRunnerBase { rtStream_t aicore{nullptr}; }; - // One slot per in-flight run the pipeline contract can declare. - static constexpr unsigned kRunStreamSetCount = PTO_PIPELINE_MAX_DEPTH; - - // AICPU streams belong to slots for the worker's lifetime. A completed - // AICore stream remains reusable until a new code upload marks every slot - // stale. Stream lifetimes live in RunStreamSlots so publication and - // failed-destroy states are testable without a device. - RunStreamSlots run_stream_slots_{ + // One pair carries every run: the execution claim is exclusive, so runs + // reach the device one at a time and the stream orders them. Pipeline slots + // index resources that preparation mutates, and preparing a run writes + // nothing to a stream. Stream lifetimes live in RunStreamPair so + // publication and failed-destroy states are testable without a device. + RunStreamPair run_streams_{ [this](void **out) { return create_run_stream(out); }, @@ -239,13 +238,14 @@ class DeviceRunner : public DeviceRunnerBase { } }; int create_run_stream(void **out); - int ensure_run_stream_set(unsigned slot); - // Destroys this run's AICore stream. Returns the driver's error and KEEPS - // the handle when the destroy fails: the stream may still hold the previous - // image's instructions, so the slot must refuse the next run until finalize - // reclaims it. - int retire_run_aicore_stream(unsigned slot, RunStreamSlots::CompletionStatus completion_status); - int destroy_run_stream_sets(); + int ensure_run_streams(); + // Retires the run pair on behalf of `owner`, the PreparedExecution that + // submitted it. Returns the driver's error and KEEPS the handle when an + // unproven destroy fails, so finalize can reclaim it. A caller that never + // submitted retires nothing: a prepared successor overlaps its + // predecessor's execution and must not destroy the live pair. + int retire_run_aicore_stream(const void *owner, RunStreamPair::CompletionStatus completion_status); + int destroy_run_streams(); // Release execution-owned resources in collector, runtime-argument, // register-buffer, then stream order. The collectors this releases were @@ -257,7 +257,7 @@ class DeviceRunner : public DeviceRunnerBase { // The kernel submission boundary is separate from the stream wait and // post-run teardown: launch_run() submits and drain_execution() reaps. LaunchTransactionResult launch_run(PreparedExecution &prepared, LaunchPermit permit); - int reap_run(unsigned slot); + int reap_run(); // On an AICore launch/sync error, best-effort drain the device so a later // enqueue on the same DeviceRunner can recover in place; if the drain itself diff --git a/src/common/platform/include/host/run_stream_pair.h b/src/common/platform/include/host/run_stream_pair.h new file mode 100644 index 0000000000..a4bb7ad45c --- /dev/null +++ b/src/common/platform/include/host/run_stream_pair.h @@ -0,0 +1,219 @@ +/* + * Copyright (c) PyPTO Contributors. + * This program is free software, you can redistribute it and/or modify it under the terms and conditions of + * CANN Open Software License Agreement Version 2.0 (the "License"). + * Please refer to the License for details. You may not use this file except in compliance with the License. + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, + * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. + * See LICENSE in the root of the software repository for the full text of the License. + * ----------------------------------------------------------------------------------------------------------- + */ + +#pragma once + +#include +#include +#include +#include +#include + +#include "pto_runtime_c_api.h" + +/** + * The one AICPU + AICore stream pair every run submits on. + * + * A stream is an ordered queue and the execution claim is exclusive, so runs + * reach the device one at a time and a single pair carries all of them. The + * pair is not indexed by pipeline slot: a slot exists for resources that + * *preparation* mutates, and preparing a run writes nothing to a stream — only + * launch submits, and launch holds the claim. + * + * The two streams must stay distinct. The AICPU Run kernel spins in the + * handshake waiting for the AICore workers, so serializing both onto one queue + * would leave the AICore submission behind a spin that can never end. + * + * The AICore stream carries instruction-cache state: cores may retain code + * fetched from a GM address whose contents a later registration replaced. + * Publishing new AICore code therefore marks the stream stale, and the next + * launch destroys it and creates a replacement. Creating a stream is the only + * operation known to leave a core free of the previous image's instructions. + * + * Only the run that submitted the pair may retire it. A prepared successor + * overlaps its predecessor's execution, so an unproven retirement from a run + * that never submitted must leave the live pair alone. + * + * Threading: launch and drain are the owning operations, but poll may query the + * pair from a progress thread while the executor retires it. The pair therefore + * serializes query with handle mutation. Poll uses try-lock and reports + * NOT_READY rather than waiting behind retirement. `created_count_` is + * additionally readable from unrelated threads and remains atomic. Stream + * creation and destruction are injected so this state machine is exercisable + * without a device. + */ +class RunStreamPair { +public: + using CreateFn = std::function; + using DestroyFn = std::function; + enum class CompletionStatus { Unproven, Complete }; + + RunStreamPair(CreateFn create, DestroyFn destroy) : + create_(std::move(create)), + destroy_(std::move(destroy)) {} + + /** + * Ready the pair for a launch: both streams on first use, and a + * replacement AICore stream when a code publication marked it stale. + * Callers hold the execution claim, so the pair is idle here. + */ + int ensure() { + std::lock_guard lock(mutex_); + if (aicpu_ == nullptr) { + int rc = create_(&aicpu_); + if (rc != 0) { + aicpu_ = nullptr; + return rc; + } + } + if (aicore_ != nullptr) { + // A run that has not retired still owns the pair: a device-complete + // poll is not a finalized run, and replacing the stream under it + // would strand a live submission. + if (owner_ != nullptr) return -1; + if (!stale_) { + submitted_ = false; + complete_ = false; + return 0; + } + int rc = destroy_(aicore_); + if (rc != 0) return rc; + aicore_ = nullptr; + } + int rc = create_(&aicore_); + if (rc != 0) { + aicore_ = nullptr; + return rc; + } + created_count_.fetch_add(1, std::memory_order_relaxed); + stale_ = false; + submitted_ = false; + complete_ = false; + return 0; + } + + /** Mark the AICore stream stale after new AICore code is published. */ + void mark_stale() { + std::lock_guard lock(mutex_); + stale_ = true; + } + + /** Make the pair visible to non-blocking poll and record its submitter. */ + int mark_submitted(const void *owner) { + if (owner == nullptr) return -1; + std::lock_guard lock(mutex_); + if (aicpu_ == nullptr || aicore_ == nullptr) return -1; + owner_ = owner; + submitted_ = true; + complete_ = false; + return 0; + } + + /** + * Query both streams without waiting behind retirement. + * + * A completed result is sticky until the pair is readied for its next run, + * so a poll racing with successful stream destruction never observes a + * missing handle as an error. + */ + template + int poll(QueryPairFn &&query) { + std::unique_lock lock(mutex_, std::try_to_lock); + if (!lock.owns_lock()) return SIMPLER_NATIVE_RUN_POLL_NOT_READY; + if (complete_) return SIMPLER_NATIVE_RUN_POLL_COMPLETE; + if (!submitted_ || aicpu_ == nullptr || aicore_ == nullptr) { + return SIMPLER_NATIVE_RUN_POLL_ERROR; + } + const int rc = std::forward(query)(aicpu_, aicore_); + if (rc == SIMPLER_NATIVE_RUN_POLL_COMPLETE) complete_ = true; + return rc; + } + + /** + * Retire the pair on behalf of the run that submitted it. Complete keeps + * the AICore stream for the next launch. Unproven destroys it: a failed + * launch or an abandoned drain may leave the stream in the error state + * rtStreamDestroy is the supported teardown for, and the handle survives a + * failed destroy so teardown can retry it. A caller that never submitted + * retires nothing. + */ + int retire(CompletionStatus completion_status, const void *owner) { + std::lock_guard lock(mutex_); + if (owner == nullptr || owner_ != owner) return 0; + // Publish the proven terminal state before destroying the handle. Poll + // either finishes its in-flight query first or observes this result. + // An error-path retirement clears a completion that raced ahead of a + // later failing sync, so the sync error remains authoritative. + submitted_ = false; + complete_ = completion_status == CompletionStatus::Complete; + owner_ = nullptr; + if (completion_status == CompletionStatus::Complete) return 0; + if (aicore_ == nullptr) return 0; + int rc = destroy_(aicore_); + if (rc != 0) return rc; + aicore_ = nullptr; + return 0; + } + + /** Destroy both streams, keeping a handle whose destroy failed. */ + int destroy() { + std::lock_guard lock(mutex_); + int first_error = 0; + submitted_ = false; + complete_ = false; + owner_ = nullptr; + for (void **stream : {&aicpu_, &aicore_}) { + if (*stream == nullptr) continue; + int rc = destroy_(*stream); + if (rc != 0) { + if (first_error == 0) first_error = rc; + continue; + } + *stream = nullptr; + } + // A handle that survived teardown may still hold the previous image's + // instructions, so it stays stale for whoever retries it. + stale_ = aicore_ != nullptr; + return first_error; + } + + /** Forget both handles after a device reset without invoking destroy_. */ + void abandon() { + std::lock_guard lock(mutex_); + aicpu_ = nullptr; + aicore_ = nullptr; + stale_ = false; + submitted_ = false; + complete_ = false; + owner_ = nullptr; + } + + // Handle reads are unsynchronized: the claim holder is the only writer once + // the pair is readied, and poll must never block behind a retirement. + void *aicpu() const { return aicpu_; } + void *aicore() const { return aicore_; } + bool ready() const { return aicpu_ != nullptr && aicore_ != nullptr; } + size_t created_count() const { return created_count_.load(std::memory_order_relaxed); } + +private: + mutable std::mutex mutex_; + void *aicpu_{nullptr}; + void *aicore_{nullptr}; + // The run that submitted the pair, or null while no run owns it. + const void *owner_{nullptr}; + bool stale_{false}; + bool submitted_{false}; + bool complete_{false}; + + CreateFn create_; + DestroyFn destroy_; + std::atomic created_count_{0}; +}; diff --git a/src/common/platform/include/host/run_stream_slots.h b/src/common/platform/include/host/run_stream_slots.h deleted file mode 100644 index 6cf0544d6c..0000000000 --- a/src/common/platform/include/host/run_stream_slots.h +++ /dev/null @@ -1,204 +0,0 @@ -/* - * Copyright (c) PyPTO Contributors. - * This program is free software, you can redistribute it and/or modify it under the terms and conditions of - * CANN Open Software License Agreement Version 2.0 (the "License"). - * Please refer to the License for details. You may not use this file except in compliance with the License. - * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, - * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. - * See LICENSE in the root of the software repository for the full text of the License. - * ----------------------------------------------------------------------------------------------------------- - */ - -#pragma once - -#include -#include -#include -#include -#include -#include - -#include "pto_runtime_c_api.h" - -/** - * Per-slot ownership of the two streams a run submits on. - * - * The AICPU stream carries no instruction-cache state, so it belongs to its - * slot for the runner's lifetime. A completed AICore stream remains reusable - * until new AICore code is published. Publication marks every slot stale; its - * next acquire destroys the old AICore stream before creating a replacement. - * - * A destroy that fails keeps its handle. Such a stream may still hold the - * previous image's instructions, so the slot must refuse the next run rather - * than hand it a fresh stream beside a live one, and teardown needs the handle - * to retry. Stream creation and destruction are injected so this state machine - * is exercisable without a device. - * - * Threading: prepare and drain remain the owning operations, but poll may query - * the active slot from a progress thread while the executor retires its AICore - * stream. Each slot therefore serializes query with handle mutation. Poll uses - * try-lock and reports NOT_READY rather than waiting behind retirement. Two - * different slots remain independent, so native prepare can acquire a - * successor while the executor drains its predecessor. `created_count_` is - * additionally readable from unrelated threads and remains atomic. - */ -class RunStreamSlots { -public: - using CreateFn = std::function; - using DestroyFn = std::function; - enum class CompletionStatus { Unproven, Complete }; - - RunStreamSlots(CreateFn create, DestroyFn destroy) : - create_(std::move(create)), - destroy_(std::move(destroy)) {} - - /** Ready `slot` for a run, reusing its AICore stream unless stale. */ - int acquire(unsigned slot) { - if (slot >= slots_.size()) return -1; - Slot &s = slots_[slot]; - std::lock_guard lock(s.mutex); - if (s.aicpu == nullptr) { - int rc = create_(&s.aicpu); - if (rc != 0) { - s.aicpu = nullptr; - return rc; - } - } - if (s.aicore != nullptr) { - if (s.submitted || !s.complete) return -1; - if (!s.stale) { - s.submitted = false; - s.complete = false; - return 0; - } - int rc = destroy_(s.aicore); - if (rc != 0) return rc; - s.aicore = nullptr; - } - int rc = create_(&s.aicore); - if (rc != 0) { - s.aicore = nullptr; - return rc; - } - created_count_.fetch_add(1, std::memory_order_relaxed); - s.stale = false; - s.submitted = false; - s.complete = false; - return 0; - } - - /** Mark every retained AICore stream stale after code publication. */ - void mark_all_stale() { - for (Slot &s : slots_) { - std::lock_guard lock(s.mutex); - s.stale = true; - } - } - - /** Make the prepared stream pair visible to non-blocking poll. */ - int mark_submitted(unsigned slot) { - if (slot >= slots_.size()) return -1; - Slot &s = slots_[slot]; - std::lock_guard lock(s.mutex); - if (s.aicpu == nullptr || s.aicore == nullptr) return -1; - s.submitted = true; - s.complete = false; - return 0; - } - - /** - * Query both streams without waiting behind retirement. - * - * A completed result is sticky until the slot is acquired for its next - * generation, so a poll racing with successful stream destruction never - * observes a missing handle as an error. - */ - template - int poll(unsigned slot, QueryPairFn &&query) { - if (slot >= slots_.size()) return SIMPLER_NATIVE_RUN_POLL_ERROR; - Slot &s = slots_[slot]; - std::unique_lock lock(s.mutex, std::try_to_lock); - if (!lock.owns_lock()) return SIMPLER_NATIVE_RUN_POLL_NOT_READY; - if (s.complete) return SIMPLER_NATIVE_RUN_POLL_COMPLETE; - if (!s.submitted || s.aicpu == nullptr || s.aicore == nullptr) { - return SIMPLER_NATIVE_RUN_POLL_ERROR; - } - const int rc = std::forward(query)(s.aicpu, s.aicore); - if (rc == SIMPLER_NATIVE_RUN_POLL_COMPLETE) s.complete = true; - return rc; - } - - /** - * Complete retains the stream for reuse. Unproven destroys it. - */ - int retire_aicore(unsigned slot, CompletionStatus completion_status) { - if (slot >= slots_.size()) return -1; - Slot &s = slots_[slot]; - std::lock_guard lock(s.mutex); - // Publish the proven terminal state before destroying the handle. Poll - // either finishes its in-flight query first or observes this result. - // An error-path retirement clears a completion that raced ahead of a - // later failing sync, so the sync error remains authoritative. - s.submitted = false; - s.complete = completion_status == CompletionStatus::Complete; - if (completion_status == CompletionStatus::Complete) return 0; - if (s.aicore == nullptr) return 0; - int rc = destroy_(s.aicore); - if (rc != 0) return rc; - s.aicore = nullptr; - return 0; - } - - /** Destroy every stream, keeping handles whose destroy failed. */ - int destroy_all() { - int first_error = 0; - for (Slot &s : slots_) { - std::lock_guard lock(s.mutex); - s.submitted = false; - s.complete = false; - s.stale = false; - for (void **stream : {&s.aicpu, &s.aicore}) { - if (*stream == nullptr) continue; - int rc = destroy_(*stream); - if (rc != 0) { - if (first_error == 0) first_error = rc; - continue; - } - *stream = nullptr; - } - } - return first_error; - } - - /** Forget every handle after device reset without invoking destroy_. */ - void abandon_all() { - for (Slot &s : slots_) { - s.aicpu = nullptr; - s.aicore = nullptr; - s.stale = false; - s.submitted = false; - s.complete = false; - } - } - - void *aicpu(unsigned slot) const { return slot < slots_.size() ? slots_[slot].aicpu : nullptr; } - void *aicore(unsigned slot) const { return slot < slots_.size() ? slots_[slot].aicore : nullptr; } - bool ready(unsigned slot) const { return aicpu(slot) != nullptr && aicore(slot) != nullptr; } - size_t created_count() const { return created_count_.load(std::memory_order_relaxed); } - static constexpr size_t capacity() { return PTO_PIPELINE_MAX_DEPTH; } - -private: - struct Slot { - mutable std::mutex mutex; - void *aicpu{nullptr}; - void *aicore{nullptr}; - bool stale{false}; - bool submitted{false}; - bool complete{false}; - }; - - CreateFn create_; - DestroyFn destroy_; - std::array slots_{}; - std::atomic created_count_{0}; -}; diff --git a/src/common/worker/chip_worker.h b/src/common/worker/chip_worker.h index eb84b8fb1a..d678f96508 100644 --- a/src/common/worker/chip_worker.h +++ b/src/common/worker/chip_worker.h @@ -161,10 +161,11 @@ class ChipWorker { /// `aicpu_dlopen_count` for the trb path; returns 0 on device-orch variants. size_t host_dlopen_count() const; - /// Number of AICore run streams the bound runner has created. AICPU streams - /// belong to slots for the worker's lifetime. Completed AICore streams are - /// reused until a new code upload makes them stale. Platforms using the - /// persistent bootstrap pair report 0. + /// Number of AICore run streams the bound runner has created. One AICPU + + /// AICore pair serves every run for the runner's lifetime. The AICPU stream + /// persists; the AICore stream is recreated when a new code upload makes it + /// stale, and destroyed when an unproven completion retires it. Platforms + /// using the persistent bootstrap pair report 0. size_t run_stream_set_create_count() const; uint64_t malloc(size_t size); diff --git a/src/common/worker/pto_runtime_c_api.h b/src/common/worker/pto_runtime_c_api.h index f3c3b782f8..79c4ec9cb1 100644 --- a/src/common/worker/pto_runtime_c_api.h +++ b/src/common/worker/pto_runtime_c_api.h @@ -447,10 +447,13 @@ size_t get_aicpu_dlopen_count(DeviceContextHandle ctx); size_t get_host_dlopen_count(DeviceContextHandle ctx); /** - * Number of AICore run streams the runner bound to `ctx` has created. AICPU - * streams belong to pipeline slots for the worker's lifetime. A completed - * AICore stream is reused until a new code upload makes it stale. Returns 0 on - * platforms whose runs use the persistent bootstrap pair. + * Number of AICore run streams the runner bound to `ctx` has created. One + * AICPU + AICore pair serves every run for the runner's lifetime. The AICPU + * stream persists; the AICore stream is recreated when a new code upload makes + * it stale, and destroyed when an unproven completion retires it, so this + * advances per publication or unproven retirement rather than once per run or + * per pipeline slot. Returns 0 on platforms whose runs use the persistent + * bootstrap pair. */ size_t get_run_stream_set_create_count(DeviceContextHandle ctx); diff --git a/tests/st/a2a3/host_build_graph/native_run_lifecycle/test_native_run_lifecycle.py b/tests/st/a2a3/host_build_graph/native_run_lifecycle/test_native_run_lifecycle.py index 385cab91a4..7a8a5d6fa4 100644 --- a/tests/st/a2a3/host_build_graph/native_run_lifecycle/test_native_run_lifecycle.py +++ b/tests/st/a2a3/host_build_graph/native_run_lifecycle/test_native_run_lifecycle.py @@ -150,8 +150,11 @@ def _run_and_validate_l2( # noqa: PLR0913, PLR0915 -- lifecycle contract is int _SLOT, chip_args, _SLOT, _GENERATION, config=config ) first_run = native_run - expected_stream_count = stream_count_before_prepare + int(supports_concurrent_prepare) - assert chip_worker.run_stream_set_create_count == expected_stream_count + # Preparation stops short of the device launch fence, and that + # includes the run streams: the pair is readied by launch, under the + # execution claim, because a prepared successor overlaps its + # predecessor's execution. + assert chip_worker.run_stream_set_create_count == stream_count_before_prepare assert torch.count_nonzero(test_args.out) == 0, "prepare crossed the device launch fence" with pytest.raises(RuntimeError, match="unfinished native run|owns the runner|active predecessor"): chip_worker._prepare_native_run_with_pipeline_lease(_SLOT, chip_args, 1, _GENERATION, config=config) @@ -227,10 +230,12 @@ def build_run_args(): self.compute_golden(run_golden, case["params"]) return run_args, run_chip_args, run_output_names, run_golden - # The successor owns a distinct bank and slot stream while A - # still owns the execution claim. A failed early launch - # must leave B prepared so the same token can launch after A's - # complete fence and finalization. + # The successor owns a distinct bank while A still owns the + # execution claim. Preparation touches no stream — the run pair + # is readied at launch, under the claim — so neither prepare + # advances the creation count. A failed early launch must leave B + # prepared so the same token can launch after A's complete fence + # and finalization. active_args, active_chip_args, active_outputs, active_golden = build_run_args() successor_args, successor_chip_args, successor_outputs, successor_golden = build_run_args() stream_count = chip_worker.run_stream_set_create_count @@ -242,7 +247,7 @@ def build_run_args(): successor_run = chip_worker._prepare_native_run_with_pipeline_lease( _SLOT, successor_chip_args, 1, _GENERATION, config=config ) - assert chip_worker.run_stream_set_create_count == stream_count + 1 + assert chip_worker.run_stream_set_create_count == stream_count bank0 = chip_worker.arena_bank_gm_heap_base(0) bank1 = chip_worker.arena_bank_gm_heap_base(1) assert bank0 != 0 diff --git a/tests/st/a2a3/host_build_graph/run_stream_reuse/test_run_stream_reuse.py b/tests/st/a2a3/host_build_graph/run_stream_reuse/test_run_stream_reuse.py index 30b650d9a3..27f1df1901 100644 --- a/tests/st/a2a3/host_build_graph/run_stream_reuse/test_run_stream_reuse.py +++ b/tests/st/a2a3/host_build_graph/run_stream_reuse/test_run_stream_reuse.py @@ -7,16 +7,20 @@ # INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. # See LICENSE in the root of the software repository for the full text of the License. # ----------------------------------------------------------------------------------------------------------- -"""A2A3 reuses completed AICore streams between code publications. +"""A2A3 reuses one AICore stream between code publications. -A2A3 submits each run's AICore and AICPU kernels on the stream set of the -selected pipeline slot rather than on the persistent bootstrap pair. The AICPU -stream carries no instruction-cache state and belongs to the slot for the -worker's lifetime. A completed AICore stream remains reusable until a new -AICore code buffer is uploaded, which marks every retained slot stream stale. +A2A3 submits every run's AICore and AICPU kernels on its own run stream pair +rather than on the persistent bootstrap pair. One pair carries all runs: the +execution claim is exclusive, so runs reach the device one at a time and the +stream orders them. A pipeline slot indexes the resources preparation mutates, +and preparing a run writes nothing to a stream. + +The AICPU stream carries no instruction-cache state. The AICore stream stays +reusable until a new AICore code buffer is uploaded, which marks it stale; the +next launch then replaces it. `Worker.run_stream_set_create_count` counts AICore stream creations, so runs -plateau after each slot is warm until another code publication. +plateau after the first launch until another code publication. """ import itertools @@ -66,7 +70,7 @@ class _SubtractCallable(SceneTestCase): @scene_test(level=2, runtime="host_build_graph") class TestRunStreamReuseHbg(SceneTestCase): - """Each pipeline slot retains a completed AICore stream until publication.""" + """One run stream pair serves every slot until a code publication.""" RTOL = 1e-5 ATOL = 1e-5 @@ -242,30 +246,28 @@ def test_resident_code_images_reuse_one_stream(self, st_platform, st_worker): st_worker.unregister(sub_handle) st_worker.unregister(add_handle) - def test_depth_two_slots_reuse_across_resident_images(self, st_platform, st_worker): - """Both slots reuse their streams while resident images alternate.""" + def test_depth_two_slots_reuse_one_pair_across_resident_images(self, st_platform, st_worker): + """Both slots submit on the one pair while resident images alternate.""" if st_platform != "a2a3": pytest.skip("AICore code images are an a2a3 onboard resource") add_handle = st_worker.register(self.build_callable(st_platform)) sub_handle = st_worker.register(_SubtractCallable.compile_chip_callable(st_platform)) try: - for slot_id, handle, subtract in ( - (0, add_handle, False), - (1, sub_handle, True), - ): - self._run_registered_with_lease( - st_worker, - handle, - slot_id=slot_id, - generation=self._next_generation(), - subtract=subtract, - ) - + self._run_registered_with_lease( + st_worker, + add_handle, + slot_id=0, + generation=self._next_generation(), + ) + # The pair is warm after the first launch. A slot is an index into + # the resources preparation mutates, not into the stream set. warmed = st_worker.run_stream_set_create_count for slot_id, handle, subtract in ( + (1, sub_handle, True), (0, sub_handle, True), (1, add_handle, False), + (0, add_handle, False), ): self._run_registered_with_lease( st_worker, @@ -275,27 +277,12 @@ def test_depth_two_slots_reuse_across_resident_images(self, st_platform, st_work subtract=subtract, ) assert st_worker.run_stream_set_create_count == warmed - - self._run_registered_with_lease( - st_worker, - sub_handle, - slot_id=0, - generation=self._next_generation(), - subtract=True, - ) - self._run_registered_with_lease( - st_worker, - add_handle, - slot_id=1, - generation=self._next_generation(), - ) - assert st_worker.run_stream_set_create_count == warmed finally: st_worker.unregister(sub_handle) st_worker.unregister(add_handle) - def test_code_publication_invalidates_both_slots_across_reregistration(self, st_platform, st_worker): - """A->B->A uploads invalidate streams retained by either slot.""" + def test_code_publication_invalidates_the_pair_across_reregistration(self, st_platform, st_worker): + """A->B->A uploads invalidate the pair a run on either slot warmed.""" if st_platform != "a2a3": pytest.skip("AICore code publication is an a2a3 onboard resource") @@ -368,11 +355,13 @@ def test_depth_two_slots_own_separate_resources(self, st_platform, st_worker): generation=self._next_generation(), ) after_slot0 = st_worker.run_stream_set_create_count + # Registering the callable published its code, so the first launch + # replaces the pair. Slot 1 then submits on that same pair: the + # stream set is not part of what a slot indexes. expected_increment = 1 if st_platform == "a2a3" else 0 assert after_slot0 == stream_sets + expected_increment self._run_registered_with_lease(st_worker, add_handle, slot_id=1, generation=self._next_generation()) - after_slot1 = st_worker.run_stream_set_create_count - assert after_slot1 == after_slot0 + expected_increment + assert st_worker.run_stream_set_create_count == after_slot0 self._run_registered_with_lease( st_worker, @@ -381,7 +370,7 @@ def test_depth_two_slots_own_separate_resources(self, st_platform, st_worker): generation=self._next_generation(), ) self._run_registered_with_lease(st_worker, add_handle, slot_id=1, generation=self._next_generation()) - assert st_worker.run_stream_set_create_count == after_slot1 + assert st_worker.run_stream_set_create_count == after_slot0 # hbg declares its GM heap HOST_PER_RUN, so a run on slot 1 must # commit a second device allocation rather than reuse slot 0's. diff --git a/tests/ut/cpp/CMakeLists.txt b/tests/ut/cpp/CMakeLists.txt index 84b0f0dba6..d96809d953 100644 --- a/tests/ut/cpp/CMakeLists.txt +++ b/tests/ut/cpp/CMakeLists.txt @@ -399,18 +399,18 @@ add_hierarchical_test(test_remote_wire hierarchical/test_remote_wire.cpp) add_hierarchical_test(test_remote_endpoint hierarchical/test_remote_endpoint.cpp) add_hierarchical_test(test_pipeline_contract hierarchical/test_pipeline_contract.cpp) -# Run-stream slot state machine: publication-aware AICore stream reuse and -# failed-destroy ownership. Header-only with injected create/destroy, so it -# needs no runtime objects and no device. -add_executable(test_run_stream_slots hierarchical/test_run_stream_slots.cpp) -target_include_directories(test_run_stream_slots PRIVATE +# Run-stream pair state machine: publication-aware AICore stream reuse, +# submitter-gated retirement, and failed-destroy ownership. Header-only with +# injected create/destroy, so it needs no runtime objects and no device. +add_executable(test_run_stream_pair hierarchical/test_run_stream_pair.cpp) +target_include_directories(test_run_stream_pair PRIVATE ${GTEST_INCLUDE_DIRS} ${CMAKE_SOURCE_DIR}/../../../src/common/platform/include ${CMAKE_SOURCE_DIR}/../../../src/common/worker ) -target_link_libraries(test_run_stream_slots PRIVATE ${GTEST_MAIN_LIB} ${GTEST_LIB} pthread) -add_test(NAME test_run_stream_slots COMMAND test_run_stream_slots) -set_tests_properties(test_run_stream_slots PROPERTIES LABELS "no_hardware") +target_link_libraries(test_run_stream_pair PRIVATE ${GTEST_MAIN_LIB} ${GTEST_LIB} pthread) +add_test(NAME test_run_stream_pair COMMAND test_run_stream_pair) +set_tests_properties(test_run_stream_pair PROPERTIES LABELS "no_hardware") add_executable(test_native_run_execution common/test_native_run_execution.cpp) target_include_directories(test_native_run_execution PRIVATE diff --git a/tests/ut/cpp/hierarchical/test_run_stream_pair.cpp b/tests/ut/cpp/hierarchical/test_run_stream_pair.cpp new file mode 100644 index 0000000000..b305678164 --- /dev/null +++ b/tests/ut/cpp/hierarchical/test_run_stream_pair.cpp @@ -0,0 +1,510 @@ +/* + * Copyright (c) PyPTO Contributors. + * This program is free software, you can redistribute it and/or modify it under the terms and conditions of + * CANN Open Software License Agreement Version 2.0 (the "License"). + * Please refer to the License for details. You may not use this file except in compliance with the License. + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, + * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. + * See LICENSE in the root of the software repository for the full text of the License. + * ----------------------------------------------------------------------------------------------------------- + */ + +#include + +#include +#include +#include +#include +#include + +#include "host/run_stream_pair.h" + +namespace { + +using CompletionStatus = RunStreamPair::CompletionStatus; + +// Stand-ins for the PreparedExecution addresses the runner passes as owners. +// Distinct values so no linker's identical-data folding can give the two +// constants one address — the tests compare owners by address. +const int kRunA = 1; +const int kRunB = 2; +const void *const kOwnerA = &kRunA; +const void *const kOwnerB = &kRunB; + +// Hands out distinct fake handles and can be told to fail the next N destroys, +// which is what makes the destroy-failure path reachable without a device. +class FakeStreams { +public: + int create(void **out) { + if (create_failures_ > 0) { + --create_failures_; + return -7; + } + *out = reinterpret_cast(++next_handle_); + live_.push_back(*out); + return 0; + } + + int destroy(void *stream) { + if (destroy_failures_ > 0) { + --destroy_failures_; + return -13; + } + for (auto it = live_.begin(); it != live_.end(); ++it) { + if (*it == stream) { + live_.erase(it); + return 0; + } + } + ADD_FAILURE() << "destroyed a handle that was not live"; + return -1; + } + + void fail_next_destroys(int n) { destroy_failures_ = n; } + void fail_next_creates(int n) { create_failures_ = n; } + size_t live_count() const { return live_.size(); } + +private: + std::vector live_; + uintptr_t next_handle_{0}; + int destroy_failures_{0}; + int create_failures_{0}; +}; + +RunStreamPair make_pair(FakeStreams &fake) { + return RunStreamPair( + [&fake](void **out) { + return fake.create(out); + }, + [&fake](void *s) { + return fake.destroy(s); + } + ); +} + +// Ready, submit and complete one run, the shape every reuse test starts from. +void run_once(RunStreamPair &pair, const void *owner) { + ASSERT_EQ(pair.ensure(), 0); + ASSERT_EQ(pair.mark_submitted(owner), 0); + ASSERT_EQ(pair.retire(CompletionStatus::Complete, owner), 0); +} + +TEST(RunStreamPair, RetiredPairIsReusedWithoutCodePublication) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + + ASSERT_EQ(pair.ensure(), 0); + void *aicpu = pair.aicpu(); + void *first_aicore = pair.aicore(); + ASSERT_NE(aicpu, nullptr); + ASSERT_NE(first_aicore, nullptr); + EXPECT_EQ(pair.created_count(), 1u); + + ASSERT_EQ(pair.mark_submitted(kOwnerA), 0); + ASSERT_EQ(pair.retire(CompletionStatus::Complete, kOwnerA), 0); + EXPECT_EQ(pair.aicore(), first_aicore); + + ASSERT_EQ(pair.ensure(), 0); + EXPECT_EQ(pair.aicpu(), aicpu) << "the AICPU stream must not be recreated"; + EXPECT_EQ(pair.aicore(), first_aicore) << "a retired stream stays warm until code publication"; + EXPECT_EQ(pair.created_count(), 1u); +} + +// Successive runs share one pair: the count is per publication, not per slot. +TEST(RunStreamPair, ConsecutiveRunsShareOneAicoreStream) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + + run_once(pair, kOwnerA); + void *aicore = pair.aicore(); + run_once(pair, kOwnerB); + run_once(pair, kOwnerA); + + EXPECT_EQ(pair.aicore(), aicore); + EXPECT_EQ(pair.created_count(), 1u); + EXPECT_EQ(fake.live_count(), 2u) << "one AICPU and one AICore stream, no per-run churn"; +} + +TEST(RunStreamPair, CodePublicationInvalidatesTheAicoreStreamOnly) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + + run_once(pair, kOwnerA); + void *aicpu = pair.aicpu(); + void *first_aicore = pair.aicore(); + + pair.mark_stale(); + ASSERT_EQ(pair.ensure(), 0); + EXPECT_EQ(pair.aicpu(), aicpu) << "the AICPU stream carries no instruction state"; + EXPECT_NE(pair.aicore(), first_aicore); + EXPECT_EQ(pair.created_count(), 2u); + EXPECT_EQ(fake.live_count(), 2u) << "the replaced stream is destroyed, not leaked"; + + // One publication invalidates once, not every run after it. + ASSERT_EQ(pair.mark_submitted(kOwnerA), 0); + ASSERT_EQ(pair.retire(CompletionStatus::Complete, kOwnerA), 0); + ASSERT_EQ(pair.ensure(), 0); + EXPECT_EQ(pair.created_count(), 2u); +} + +TEST(RunStreamPair, CodePublicationDuringARunRebuildsAfterRetirement) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + + ASSERT_EQ(pair.ensure(), 0); + void *active_aicore = pair.aicore(); + ASSERT_EQ(pair.mark_submitted(kOwnerA), 0); + + // A registration lands while the run is on the device. Its own code was + // never overwritten, so it finishes on the stream it launched with. + pair.mark_stale(); + EXPECT_EQ(pair.aicore(), active_aicore); + ASSERT_EQ(pair.retire(CompletionStatus::Complete, kOwnerA), 0); + + ASSERT_EQ(pair.ensure(), 0); + EXPECT_NE(pair.aicore(), active_aicore); + EXPECT_EQ(pair.created_count(), 2u); +} + +// The invariant a shared pair needs that per-slot streams did not: a prepared +// successor overlaps its predecessor's execution, so its cleanup must not +// destroy the stream the predecessor is still running on. +TEST(RunStreamPair, ANonSubmittingOwnerRetiresNothing) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + + ASSERT_EQ(pair.ensure(), 0); + void *active_aicore = pair.aicore(); + ASSERT_EQ(pair.mark_submitted(kOwnerA), 0); + + // The successor prepared, failed after preparation, and cleans up. + EXPECT_EQ(pair.retire(CompletionStatus::Unproven, kOwnerB), 0); + EXPECT_EQ(pair.aicore(), active_aicore) << "the predecessor's live stream must survive"; + EXPECT_EQ(fake.live_count(), 2u); + + // The predecessor's own poll and retirement still work. + EXPECT_EQ( + pair.poll([](void *, void *) { + return SIMPLER_NATIVE_RUN_POLL_COMPLETE; + }), + SIMPLER_NATIVE_RUN_POLL_COMPLETE + ); + ASSERT_EQ(pair.retire(CompletionStatus::Complete, kOwnerA), 0); + EXPECT_EQ(pair.aicore(), active_aicore); +} + +TEST(RunStreamPair, RetirementAfterCompletionIsIgnored) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + + run_once(pair, kOwnerA); + void *aicore = pair.aicore(); + + // Cleanup runs after drain already published the terminal state. + EXPECT_EQ(pair.retire(CompletionStatus::Unproven, kOwnerA), 0); + EXPECT_EQ(pair.aicore(), aicore) << "a second retirement must not destroy the warm stream"; + EXPECT_EQ(fake.live_count(), 2u); +} + +TEST(RunStreamPair, CompleteRetirementWithoutAHandleIsBenign) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + + EXPECT_EQ(pair.retire(CompletionStatus::Complete, kOwnerA), 0); + EXPECT_EQ(pair.retire(CompletionStatus::Unproven, kOwnerA), 0); + EXPECT_EQ(pair.created_count(), 0u); +} + +TEST(RunStreamPair, PollCompletionDoesNotPermitReuseBeforeRetirement) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + + ASSERT_EQ(pair.ensure(), 0); + void *aicore = pair.aicore(); + ASSERT_EQ(pair.mark_submitted(kOwnerA), 0); + ASSERT_EQ( + pair.poll([](void *, void *) { + return SIMPLER_NATIVE_RUN_POLL_COMPLETE; + }), + SIMPLER_NATIVE_RUN_POLL_COMPLETE + ); + + EXPECT_NE(pair.ensure(), 0) << "a device-complete run is not a finalized one"; + ASSERT_EQ(pair.retire(CompletionStatus::Complete, kOwnerA), 0); + ASSERT_EQ(pair.ensure(), 0); + EXPECT_EQ(pair.aicore(), aicore); +} + +TEST(RunStreamPair, UnprovenRunDestroysTheAicoreStreamAndKeepsTheAicpuOne) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + + ASSERT_EQ(pair.ensure(), 0); + void *aicpu = pair.aicpu(); + ASSERT_EQ(pair.mark_submitted(kOwnerA), 0); + ASSERT_EQ(pair.retire(CompletionStatus::Unproven, kOwnerA), 0); + EXPECT_EQ(pair.aicore(), nullptr); + EXPECT_EQ(pair.aicpu(), aicpu); + + ASSERT_EQ(pair.ensure(), 0); + EXPECT_EQ(pair.aicpu(), aicpu); + EXPECT_EQ(pair.created_count(), 2u); +} + +// A launch that never reached submission owns nothing, so its cleanup is a +// no-op and the next launch reuses what is already there. +TEST(RunStreamPair, AnUnsubmittedReadyingIsReusedByTheNextLaunch) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + + ASSERT_EQ(pair.ensure(), 0); + void *aicore = pair.aicore(); + EXPECT_EQ(pair.retire(CompletionStatus::Unproven, kOwnerA), 0); + EXPECT_EQ(pair.aicore(), aicore); + + ASSERT_EQ(pair.ensure(), 0); + EXPECT_EQ(pair.aicore(), aicore); + EXPECT_EQ(pair.created_count(), 1u); +} + +TEST(RunStreamPair, AFailedStaleDestroyReportsAndKeepsTheHandle) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + + run_once(pair, kOwnerA); + void *stranded = pair.aicore(); + + // 1. the image transition reports the destroy error + pair.mark_stale(); + fake.fail_next_destroys(1); + EXPECT_EQ(pair.ensure(), -13); + + // 2. the handle survives, so teardown still has something to reclaim + EXPECT_EQ(pair.aicore(), stranded); + EXPECT_EQ(pair.created_count(), 1u) << "a failed readying must not create anything"; + + // 3. the stream stays stale, so the next launch retries the destroy rather + // than running on a stream that may hold the previous image + ASSERT_EQ(pair.ensure(), 0); + EXPECT_NE(pair.aicore(), stranded); + EXPECT_EQ(pair.created_count(), 2u); + EXPECT_EQ(fake.live_count(), 2u); + + EXPECT_EQ(pair.destroy(), 0); + EXPECT_EQ(pair.aicore(), nullptr); + EXPECT_EQ(fake.live_count(), 0u); +} + +TEST(RunStreamPair, AHandleThatSurvivedTeardownStaysStale) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + + run_once(pair, kOwnerA); + void *stranded_aicpu = pair.aicpu(); + void *stranded_aicore = pair.aicore(); + + // Teardown fails on both handles and keeps them for a later retry. + fake.fail_next_destroys(2); + EXPECT_EQ(pair.destroy(), -13); + EXPECT_EQ(pair.aicpu(), stranded_aicpu); + EXPECT_EQ(pair.aicore(), stranded_aicore); + EXPECT_EQ(fake.live_count(), 2u); + + // A reused runner must not launch on the surviving AICore stream: readying + // replaces it first. The AICPU stream carries no instruction state, so it + // is kept. + ASSERT_EQ(pair.ensure(), 0); + EXPECT_EQ(pair.aicpu(), stranded_aicpu); + EXPECT_NE(pair.aicore(), stranded_aicore); + EXPECT_EQ(pair.destroy(), 0); + EXPECT_EQ(fake.live_count(), 0u); +} + +TEST(RunStreamPair, UnprovenRetirementClearsAnEarlyCompletion) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + ASSERT_EQ(pair.ensure(), 0); + ASSERT_EQ(pair.mark_submitted(kOwnerA), 0); + + EXPECT_EQ( + pair.poll([](void *, void *) { + return SIMPLER_NATIVE_RUN_POLL_COMPLETE; + }), + SIMPLER_NATIVE_RUN_POLL_COMPLETE + ); + + // A later stream-sync error is authoritative even if a concurrent query + // reached COMPLETE first. Error-path cleanup must not preserve that fence. + ASSERT_EQ(pair.retire(CompletionStatus::Unproven, kOwnerA), 0); + EXPECT_EQ( + pair.poll([](void *, void *) { + ADD_FAILURE() << "an unproven retired stream must not be queried"; + return SIMPLER_NATIVE_RUN_POLL_COMPLETE; + }), + SIMPLER_NATIVE_RUN_POLL_ERROR + ); +} + +TEST(RunStreamPair, PollRequiresSubmissionAndPropagatesQueryErrors) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + ASSERT_EQ(pair.ensure(), 0); + + EXPECT_EQ( + pair.poll([](void *, void *) { + return SIMPLER_NATIVE_RUN_POLL_COMPLETE; + }), + SIMPLER_NATIVE_RUN_POLL_ERROR + ); + ASSERT_EQ(pair.mark_submitted(kOwnerA), 0); + EXPECT_EQ( + pair.poll([](void *, void *) { + return SIMPLER_NATIVE_RUN_POLL_ERROR; + }), + SIMPLER_NATIVE_RUN_POLL_ERROR + ); +} + +TEST(RunStreamPair, SubmissionRequiresBothStreamsAndAnOwner) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + + EXPECT_NE(pair.mark_submitted(kOwnerA), 0) << "nothing is created yet"; + ASSERT_EQ(pair.ensure(), 0); + EXPECT_NE(pair.mark_submitted(nullptr), 0) << "an anonymous run could never retire"; + EXPECT_EQ(pair.mark_submitted(kOwnerA), 0); +} + +TEST(RunStreamPair, RetireWaitsForAnInFlightPoll) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + ASSERT_EQ(pair.ensure(), 0); + ASSERT_EQ(pair.mark_submitted(kOwnerA), 0); + + std::promise query_entered; + std::promise release_query; + std::shared_future release = release_query.get_future().share(); + auto poll = std::async(std::launch::async, [&]() { + return pair.poll([&](void *, void *) { + query_entered.set_value(); + release.wait(); + return SIMPLER_NATIVE_RUN_POLL_NOT_READY; + }); + }); + query_entered.get_future().wait(); + + std::promise retire_started; + auto retire = std::async(std::launch::async, [&]() { + retire_started.set_value(); + return pair.retire(CompletionStatus::Unproven, kOwnerA); + }); + retire_started.get_future().wait(); + EXPECT_EQ(retire.wait_for(std::chrono::seconds(0)), std::future_status::timeout); + + release_query.set_value(); + EXPECT_EQ(poll.get(), SIMPLER_NATIVE_RUN_POLL_NOT_READY); + EXPECT_EQ(retire.get(), 0); + EXPECT_EQ(fake.live_count(), 1u); + EXPECT_EQ( + pair.poll([](void *, void *) { + return SIMPLER_NATIVE_RUN_POLL_ERROR; + }), + SIMPLER_NATIVE_RUN_POLL_ERROR + ); +} + +TEST(RunStreamPair, PollDoesNotWaitBehindRetirement) { + FakeStreams fake; + std::promise destroy_entered; + std::promise release_destroy; + std::shared_future release = release_destroy.get_future().share(); + RunStreamPair pair( + [&fake](void **out) { + return fake.create(out); + }, + [&](void *stream) { + destroy_entered.set_value(); + release.wait(); + return fake.destroy(stream); + } + ); + ASSERT_EQ(pair.ensure(), 0); + ASSERT_EQ(pair.mark_submitted(kOwnerA), 0); + + auto retire = std::async(std::launch::async, [&]() { + return pair.retire(CompletionStatus::Unproven, kOwnerA); + }); + destroy_entered.get_future().wait(); + + // The destroy holds the pair's mutex. Poll reports NOT_READY rather than + // blocking a progress thread behind a driver call. + EXPECT_EQ( + pair.poll([](void *, void *) { + ADD_FAILURE() << "poll must not run its query while retirement holds the pair"; + return SIMPLER_NATIVE_RUN_POLL_COMPLETE; + }), + SIMPLER_NATIVE_RUN_POLL_NOT_READY + ); + + release_destroy.set_value(); + EXPECT_EQ(retire.get(), 0); +} + +TEST(RunStreamPair, DestroyReportsFailureAndRetriesWhatSurvived) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + ASSERT_EQ(pair.ensure(), 0); + + fake.fail_next_destroys(1); + EXPECT_EQ(pair.destroy(), -13); + EXPECT_EQ(fake.live_count(), 1u); + + EXPECT_EQ(pair.destroy(), 0); + EXPECT_EQ(fake.live_count(), 0u); + EXPECT_FALSE(pair.ready()); +} + +TEST(RunStreamPair, AbandonClearsHandlesWithoutDestroyingThem) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + ASSERT_EQ(pair.ensure(), 0); + ASSERT_EQ(pair.mark_submitted(kOwnerA), 0); + + pair.abandon(); + + EXPECT_EQ(pair.aicpu(), nullptr); + EXPECT_EQ(pair.aicore(), nullptr); + EXPECT_FALSE(pair.ready()); + EXPECT_EQ(fake.live_count(), 2u) << "a reset device invalidated them; destroy must not be called"; + EXPECT_EQ( + pair.poll([](void *, void *) { + ADD_FAILURE() << "an abandoned pair has nothing to query"; + return SIMPLER_NATIVE_RUN_POLL_COMPLETE; + }), + SIMPLER_NATIVE_RUN_POLL_ERROR + ); + EXPECT_EQ(pair.destroy(), 0); +} + +TEST(RunStreamPair, AFailedCreateLeavesThePairEmpty) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + + fake.fail_next_creates(1); + EXPECT_NE(pair.ensure(), 0); + EXPECT_EQ(pair.aicpu(), nullptr); + EXPECT_EQ(pair.aicore(), nullptr); + EXPECT_EQ(pair.created_count(), 0u); + + // The AICPU stream lands, the AICore one fails: the pair stays unusable + // rather than half-ready. + fake.fail_next_creates(0); + ASSERT_EQ(pair.ensure(), 0); + ASSERT_EQ(pair.mark_submitted(kOwnerA), 0); + ASSERT_EQ(pair.retire(CompletionStatus::Unproven, kOwnerA), 0); + fake.fail_next_creates(1); + EXPECT_NE(pair.ensure(), 0); + EXPECT_FALSE(pair.ready()); +} + +} // namespace diff --git a/tests/ut/cpp/hierarchical/test_run_stream_slots.cpp b/tests/ut/cpp/hierarchical/test_run_stream_slots.cpp deleted file mode 100644 index f21119e213..0000000000 --- a/tests/ut/cpp/hierarchical/test_run_stream_slots.cpp +++ /dev/null @@ -1,507 +0,0 @@ -/* - * Copyright (c) PyPTO Contributors. - * This program is free software, you can redistribute it and/or modify it under the terms and conditions of - * CANN Open Software License Agreement Version 2.0 (the "License"). - * Please refer to the License for details. You may not use this file except in compliance with the License. - * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, - * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. - * See LICENSE in the root of the software repository for the full text of the License. - * ----------------------------------------------------------------------------------------------------------- - */ - -#include - -#include -#include -#include -#include -#include - -#include "host/run_stream_slots.h" - -namespace { - -using CompletionStatus = RunStreamSlots::CompletionStatus; - -// Hands out distinct fake handles and can be told to fail the next N destroys, -// which is what makes the destroy-failure path reachable without a device. -class FakeStreams { -public: - int create(void **out) { - if (create_failures_ > 0) { - --create_failures_; - return -7; - } - *out = reinterpret_cast(++next_handle_); - live_.push_back(*out); - return 0; - } - - int destroy(void *stream) { - if (destroy_failures_ > 0) { - --destroy_failures_; - return -13; - } - for (auto it = live_.begin(); it != live_.end(); ++it) { - if (*it == stream) { - live_.erase(it); - return 0; - } - } - ADD_FAILURE() << "destroyed a handle that was not live"; - return -1; - } - - void fail_next_destroys(int n) { destroy_failures_ = n; } - void fail_next_creates(int n) { create_failures_ = n; } - size_t live_count() const { return live_.size(); } - -private: - std::vector live_; - uintptr_t next_handle_{0}; - int destroy_failures_{0}; - int create_failures_{0}; -}; - -RunStreamSlots make_slots(FakeStreams &fake) { - return RunStreamSlots( - [&fake](void **out) { - return fake.create(out); - }, - [&fake](void *s) { - return fake.destroy(s); - } - ); -} - -TEST(RunStreamSlots, RetiredStreamIsReusedWithoutCodePublication) { - FakeStreams fake; - RunStreamSlots slots = make_slots(fake); - - ASSERT_EQ(slots.acquire(0), 0); - void *aicpu = slots.aicpu(0); - void *first_aicore = slots.aicore(0); - ASSERT_NE(aicpu, nullptr); - ASSERT_NE(first_aicore, nullptr); - EXPECT_EQ(slots.created_count(), 1u); - - ASSERT_EQ(slots.mark_submitted(0), 0); - ASSERT_EQ(slots.retire_aicore(0, CompletionStatus::Complete), 0); - EXPECT_EQ(slots.aicore(0), first_aicore); - - ASSERT_EQ(slots.acquire(0), 0); - EXPECT_EQ(slots.aicpu(0), aicpu) << "the AICPU stream must not be recreated"; - EXPECT_EQ(slots.aicore(0), first_aicore) << "a retired stream stays warm until code publication"; - EXPECT_EQ(slots.created_count(), 1u); -} - -TEST(RunStreamSlots, CodePublicationInvalidatesEveryRetiredSlot) { - FakeStreams fake; - RunStreamSlots slots = make_slots(fake); - - ASSERT_EQ(slots.acquire(0), 0); - ASSERT_EQ(slots.acquire(1), 0); - void *slot0_aicore = slots.aicore(0); - void *slot1_aicore = slots.aicore(1); - ASSERT_EQ(slots.mark_submitted(0), 0); - ASSERT_EQ(slots.mark_submitted(1), 0); - ASSERT_EQ(slots.retire_aicore(0, CompletionStatus::Complete), 0); - ASSERT_EQ(slots.retire_aicore(1, CompletionStatus::Complete), 0); - - slots.mark_all_stale(); - ASSERT_EQ(slots.acquire(0), 0); - ASSERT_EQ(slots.acquire(1), 0); - EXPECT_NE(slots.aicore(0), slot0_aicore); - EXPECT_NE(slots.aicore(1), slot1_aicore); - EXPECT_EQ(slots.created_count(), 4u); -} - -TEST(RunStreamSlots, CodePublicationDuringARunRebuildsAfterRetirement) { - FakeStreams fake; - RunStreamSlots slots = make_slots(fake); - - ASSERT_EQ(slots.acquire(0), 0); - void *active_aicore = slots.aicore(0); - ASSERT_EQ(slots.mark_submitted(0), 0); - - slots.mark_all_stale(); - ASSERT_EQ(slots.retire_aicore(0, CompletionStatus::Complete), 0); - ASSERT_EQ(slots.acquire(0), 0); - EXPECT_NE(slots.aicore(0), active_aicore); - EXPECT_EQ(slots.created_count(), 2u); -} - -TEST(RunStreamSlots, CompleteRetirementWithoutAHandleIsBenign) { - FakeStreams fake; - RunStreamSlots slots = make_slots(fake); - - EXPECT_EQ(slots.retire_aicore(0, CompletionStatus::Complete), 0); -} - -TEST(RunStreamSlots, PollCompletionDoesNotPermitReuseBeforeRetirement) { - FakeStreams fake; - RunStreamSlots slots = make_slots(fake); - - ASSERT_EQ(slots.acquire(0), 0); - void *aicore = slots.aicore(0); - ASSERT_EQ(slots.mark_submitted(0), 0); - ASSERT_EQ( - slots.poll( - 0, - [](void *, void *) { - return SIMPLER_NATIVE_RUN_POLL_COMPLETE; - } - ), - SIMPLER_NATIVE_RUN_POLL_COMPLETE - ); - - EXPECT_NE(slots.acquire(0), 0); - ASSERT_EQ(slots.retire_aicore(0, CompletionStatus::Complete), 0); - ASSERT_EQ(slots.acquire(0), 0); - EXPECT_EQ(slots.aicore(0), aicore); -} - -TEST(RunStreamSlots, UnprovenRunDestroysTheAicoreStream) { - FakeStreams fake; - RunStreamSlots slots = make_slots(fake); - - ASSERT_EQ(slots.acquire(0), 0); - ASSERT_EQ(slots.retire_aicore(0, CompletionStatus::Unproven), 0); - EXPECT_EQ(slots.aicore(0), nullptr); - - ASSERT_EQ(slots.acquire(0), 0); - EXPECT_EQ(slots.created_count(), 2u); -} - -TEST(RunStreamSlots, PipelineSlotsKeepIndependentWarmStreams) { - FakeStreams fake; - RunStreamSlots slots = make_slots(fake); - - ASSERT_EQ(slots.acquire(0), 0); - ASSERT_EQ(slots.acquire(1), 0); - void *slot0_aicore = slots.aicore(0); - void *slot1_aicore = slots.aicore(1); - ASSERT_NE(slot0_aicore, slot1_aicore); - - ASSERT_EQ(slots.mark_submitted(0), 0); - ASSERT_EQ(slots.mark_submitted(1), 0); - ASSERT_EQ(slots.retire_aicore(0, CompletionStatus::Complete), 0); - ASSERT_EQ(slots.retire_aicore(1, CompletionStatus::Complete), 0); - - ASSERT_EQ(slots.acquire(0), 0); - ASSERT_EQ(slots.acquire(1), 0); - EXPECT_EQ(slots.aicore(0), slot0_aicore); - EXPECT_EQ(slots.aicore(1), slot1_aicore); - EXPECT_EQ(slots.created_count(), 2u); -} - -TEST(RunStreamSlots, AFailedStaleDestroyReportsAndKeepsTheHandle) { - FakeStreams fake; - RunStreamSlots slots = make_slots(fake); - - ASSERT_EQ(slots.acquire(0), 0); - ASSERT_EQ(slots.mark_submitted(0), 0); - void *stranded = slots.aicore(0); - - ASSERT_EQ(slots.retire_aicore(0, CompletionStatus::Complete), 0); - - slots.mark_all_stale(); - fake.fail_next_destroys(1); - EXPECT_EQ(slots.acquire(0), -13); - - EXPECT_EQ(slots.aicore(0), stranded); - - // Device execution was already drained before retirement. The retained - // handle is teardown ownership, not permission to query that stream again. - EXPECT_EQ( - slots.poll( - 0, - [](void *, void *) { - return SIMPLER_NATIVE_RUN_POLL_ERROR; - } - ), - SIMPLER_NATIVE_RUN_POLL_COMPLETE - ); - - ASSERT_EQ(slots.acquire(0), 0); - EXPECT_NE(slots.aicore(0), stranded); - EXPECT_EQ(slots.created_count(), 2u); - EXPECT_EQ(slots.destroy_all(), 0); - EXPECT_EQ(slots.aicore(0), nullptr); - EXPECT_EQ(fake.live_count(), 0u); -} - -// A slot locked by a failed destroy must not take the whole runner with it. -TEST(RunStreamSlots, AStrandedSlotDoesNotBlockItsPeer) { - FakeStreams fake; - RunStreamSlots slots = make_slots(fake); - - ASSERT_EQ(slots.acquire(0), 0); - fake.fail_next_destroys(1); - ASSERT_NE(slots.retire_aicore(0, CompletionStatus::Unproven), 0); - - ASSERT_EQ(slots.acquire(1), 0); - EXPECT_NE(slots.aicore(1), nullptr); - EXPECT_EQ(slots.retire_aicore(1, CompletionStatus::Unproven), 0); -} - -TEST(RunStreamSlots, SuccessorProvisionAndAbandonDoNotTouchTheActiveSlot) { - FakeStreams fake; - RunStreamSlots slots = make_slots(fake); - - ASSERT_EQ(slots.acquire(0), 0); - void *active_aicore = slots.aicore(0); - - // Native prepare provisions the successor before the predecessor retires. - ASSERT_EQ(slots.acquire(1), 0); - EXPECT_TRUE(slots.ready(0)); - EXPECT_TRUE(slots.ready(1)); - EXPECT_NE(slots.aicore(1), active_aicore); - EXPECT_EQ(slots.created_count(), 2u); - - // Abandoning the unpublished successor retires only its AICore - // stream. Its slot-persistent AICPU stream remains reusable. - ASSERT_EQ(slots.retire_aicore(1, CompletionStatus::Unproven), 0); - EXPECT_TRUE(slots.ready(0)); - EXPECT_EQ(slots.aicore(0), active_aicore); - EXPECT_EQ(slots.aicore(1), nullptr); - EXPECT_NE(slots.aicpu(1), nullptr); - - EXPECT_EQ(slots.retire_aicore(0, CompletionStatus::Unproven), 0); - EXPECT_EQ(slots.destroy_all(), 0); -} - -TEST(RunStreamSlots, PollReportsPendingThenStickyCompleteWithoutRetiringStreams) { - FakeStreams fake; - RunStreamSlots slots = make_slots(fake); - ASSERT_EQ(slots.acquire(0), 0); - ASSERT_EQ(slots.mark_submitted(0), 0); - - void *expected_aicpu = slots.aicpu(0); - void *expected_aicore = slots.aicore(0); - int query_count = 0; - auto query = [&](void *aicpu, void *aicore) { - EXPECT_EQ(aicpu, expected_aicpu); - EXPECT_EQ(aicore, expected_aicore); - ++query_count; - return query_count == 1 ? SIMPLER_NATIVE_RUN_POLL_NOT_READY : SIMPLER_NATIVE_RUN_POLL_COMPLETE; - }; - - EXPECT_EQ(slots.poll(0, query), SIMPLER_NATIVE_RUN_POLL_NOT_READY); - EXPECT_EQ(fake.live_count(), 2u); - EXPECT_EQ(slots.poll(0, query), SIMPLER_NATIVE_RUN_POLL_COMPLETE); - EXPECT_EQ(slots.poll(0, query), SIMPLER_NATIVE_RUN_POLL_COMPLETE); - EXPECT_EQ(query_count, 2) << "a completed stream pair must not be queried again"; - EXPECT_EQ(fake.live_count(), 2u) << "poll observes ownership but never retires it"; -} - -TEST(RunStreamSlots, UnprovenRetirementClearsAnEarlyCompletion) { - FakeStreams fake; - RunStreamSlots slots = make_slots(fake); - ASSERT_EQ(slots.acquire(0), 0); - ASSERT_EQ(slots.mark_submitted(0), 0); - - EXPECT_EQ( - slots.poll( - 0, - [](void *, void *) { - return SIMPLER_NATIVE_RUN_POLL_COMPLETE; - } - ), - SIMPLER_NATIVE_RUN_POLL_COMPLETE - ); - - // A later stream-sync error is authoritative even if a concurrent query - // reached COMPLETE first. Error-path cleanup must not preserve that fence. - ASSERT_EQ(slots.retire_aicore(0, CompletionStatus::Unproven), 0); - EXPECT_EQ( - slots.poll( - 0, - [](void *, void *) { - ADD_FAILURE() << "an unproven retired stream must not be queried"; - return SIMPLER_NATIVE_RUN_POLL_COMPLETE; - } - ), - SIMPLER_NATIVE_RUN_POLL_ERROR - ); -} - -TEST(RunStreamSlots, PollRequiresSubmissionAndPropagatesQueryErrors) { - FakeStreams fake; - RunStreamSlots slots = make_slots(fake); - ASSERT_EQ(slots.acquire(0), 0); - - EXPECT_EQ( - slots.poll( - 0, - [](void *, void *) { - return SIMPLER_NATIVE_RUN_POLL_COMPLETE; - } - ), - SIMPLER_NATIVE_RUN_POLL_ERROR - ); - ASSERT_EQ(slots.mark_submitted(0), 0); - EXPECT_EQ( - slots.poll( - 0, - [](void *, void *) { - return SIMPLER_NATIVE_RUN_POLL_ERROR; - } - ), - SIMPLER_NATIVE_RUN_POLL_ERROR - ); - EXPECT_EQ( - slots.poll( - static_cast(RunStreamSlots::capacity()), - [](void *, void *) { - return SIMPLER_NATIVE_RUN_POLL_COMPLETE; - } - ), - SIMPLER_NATIVE_RUN_POLL_ERROR - ); -} - -TEST(RunStreamSlots, RetireWaitsForAnInFlightPoll) { - FakeStreams fake; - RunStreamSlots slots = make_slots(fake); - ASSERT_EQ(slots.acquire(0), 0); - ASSERT_EQ(slots.mark_submitted(0), 0); - - std::promise query_entered; - std::promise release_query; - std::shared_future release = release_query.get_future().share(); - auto poll = std::async(std::launch::async, [&]() { - return slots.poll(0, [&](void *, void *) { - query_entered.set_value(); - release.wait(); - return SIMPLER_NATIVE_RUN_POLL_NOT_READY; - }); - }); - query_entered.get_future().wait(); - - std::promise retire_started; - auto retire = std::async(std::launch::async, [&]() { - retire_started.set_value(); - return slots.retire_aicore(0, CompletionStatus::Unproven); - }); - retire_started.get_future().wait(); - EXPECT_EQ(retire.wait_for(std::chrono::seconds(0)), std::future_status::timeout); - - release_query.set_value(); - EXPECT_EQ(poll.get(), SIMPLER_NATIVE_RUN_POLL_NOT_READY); - EXPECT_EQ(retire.get(), 0); - EXPECT_EQ(fake.live_count(), 1u); - EXPECT_EQ( - slots.poll( - 0, - [](void *, void *) { - return SIMPLER_NATIVE_RUN_POLL_ERROR; - } - ), - SIMPLER_NATIVE_RUN_POLL_ERROR - ); -} - -TEST(RunStreamSlots, PollDoesNotWaitBehindRetirement) { - FakeStreams fake; - std::promise destroy_entered; - std::promise release_destroy; - std::shared_future release = release_destroy.get_future().share(); - RunStreamSlots slots( - [&fake](void **out) { - return fake.create(out); - }, - [&](void *stream) { - destroy_entered.set_value(); - release.wait(); - return fake.destroy(stream); - } - ); - ASSERT_EQ(slots.acquire(0), 0); - ASSERT_EQ(slots.mark_submitted(0), 0); - - auto retire = std::async(std::launch::async, [&]() { - return slots.retire_aicore(0, CompletionStatus::Unproven); - }); - destroy_entered.get_future().wait(); - - std::promise poll_started; - bool query_called = false; - auto poll = std::async(std::launch::async, [&]() { - poll_started.set_value(); - return slots.poll(0, [&](void *, void *) { - query_called = true; - return SIMPLER_NATIVE_RUN_POLL_COMPLETE; - }); - }); - poll_started.get_future().wait(); - const std::future_status poll_status = poll.wait_for(std::chrono::seconds(1)); - - release_destroy.set_value(); - EXPECT_EQ(poll_status, std::future_status::ready); - EXPECT_EQ(poll.get(), SIMPLER_NATIVE_RUN_POLL_NOT_READY); - EXPECT_FALSE(query_called); - EXPECT_EQ(retire.get(), 0); -} - -// destroy_all reports the first failure and keeps only what it could not free, -// so a second teardown attempt is still meaningful. -TEST(RunStreamSlots, DestroyAllReportsFailureAndRetriesWhatSurvived) { - FakeStreams fake; - RunStreamSlots slots = make_slots(fake); - ASSERT_EQ(slots.acquire(0), 0); - ASSERT_EQ(slots.acquire(1), 0); - - fake.fail_next_destroys(1); - EXPECT_EQ(slots.destroy_all(), -13); - EXPECT_EQ(fake.live_count(), 1u) << "only the stream whose destroy failed may survive"; - - EXPECT_EQ(slots.destroy_all(), 0); - EXPECT_EQ(fake.live_count(), 0u); -} - -TEST(RunStreamSlots, AbandonAllClearsHandlesWithoutDestroyingThem) { - FakeStreams fake; - RunStreamSlots slots = make_slots(fake); - ASSERT_EQ(slots.acquire(0), 0); - - slots.abandon_all(); - - EXPECT_EQ(slots.aicpu(0), nullptr); - EXPECT_EQ(slots.aicore(0), nullptr); - EXPECT_EQ(fake.live_count(), 2u); - EXPECT_EQ(slots.destroy_all(), 0); - EXPECT_EQ(fake.live_count(), 2u); -} - -// A failed create leaves nothing half-owned behind. -TEST(RunStreamSlots, AFailedCreateLeavesTheSlotEmpty) { - FakeStreams fake; - RunStreamSlots slots = make_slots(fake); - - fake.fail_next_creates(1); - EXPECT_NE(slots.acquire(0), 0); - EXPECT_EQ(slots.aicpu(0), nullptr); - EXPECT_EQ(slots.aicore(0), nullptr); - EXPECT_EQ(slots.created_count(), 0u); - - // The AICPU stream lands, the AICore one fails: the slot stays unusable - // rather than half-ready. - fake.fail_next_creates(0); - ASSERT_EQ(slots.acquire(0), 0); - ASSERT_EQ(slots.retire_aicore(0, CompletionStatus::Unproven), 0); - fake.fail_next_creates(1); - EXPECT_NE(slots.acquire(0), 0); - EXPECT_FALSE(slots.ready(0)); -} - -TEST(RunStreamSlots, RejectsASlotOutsideTheContractDepth) { - FakeStreams fake; - RunStreamSlots slots = make_slots(fake); - const unsigned past_end = static_cast(RunStreamSlots::capacity()); - EXPECT_NE(slots.acquire(past_end), 0); - EXPECT_NE(slots.retire_aicore(past_end, CompletionStatus::Unproven), 0); - EXPECT_EQ(slots.aicore(past_end), nullptr); -} - -} // namespace