diff --git a/docs/dfx/device-phases.md b/docs/dfx/device-phases.md index a9848406e0..6dc40686d4 100644 --- a/docs/dfx/device-phases.md +++ b/docs/dfx/device-phases.md @@ -188,7 +188,8 @@ threads, no per-task AICore records, works in `SIMPLER_DFX=0`. See `ChipTaskSlotState` in the `TaskAttrs` byte (bit 3 `is_timed` + bits 4-7 the 0..15 tag), co-located with the other per-task scheduling flags. The 16 slots are a fixed `TaskTimingRecord[16]` **tail** appended after the `AicpuPhaseRecord` - region in the same device buffer — same base pointer and per-run H2D reset. + region in the same per-slot device buffer — one base per in-flight slot and a + per-run H2D reset. A 16-byte header at the front lets the host skip the tail D2H entirely when no task was tagged (saving 1536 bytes on a2a3 and 3584 bytes on a5 per run). The tail is a distinct record type (dispatch/finish, not start/end) reduced by diff --git a/docs/dfx/host-trace.md b/docs/dfx/host-trace.md index 26e0df14bf..ca1de6d32a 100644 --- a/docs/dfx/host-trace.md +++ b/docs/dfx/host-trace.md @@ -152,7 +152,7 @@ children are TMR-only; HBG orchestration runs on the host and stamps none of those phases. All emitted device spans are tagged `clk=dev`. They are not host `steady_clock` spans: the AICPU stamps raw sys-counter cycles into a host-allocated buffer (whose address rides on `KernelArgs::device_wall_data_base`), the host reads it -back after stream-sync, converts cycles → ns, and emits the marker. `orch`/ +back after that run's completion event, converts cycles → ns, and emits the marker. `orch`/ `sched` are the orchestrator/scheduler windows that formerly only appeared as device-log lines. A phase that was never stamped (0 ns) is skipped — e.g. `so_load` is ~0 on a cached-callable run. See @@ -161,7 +161,7 @@ device-log lines. A phase that was never stamped The phased native-run interface preserves this same marker contract. Prepare allocates one `inv` and records the host-wall start; prepare, the child progress path's launch/drain lifecycle, and finalize bind that `(inv, hid)` while -emitting their spans. Finalize releases the runner claim, destroys the per-run +emitting their spans. Finalize releases the run's submitted-owner slot, destroys the per-run state, and then emits the stored `chip.run` wall, so the root includes that cleanup tail. No trace scope or synthetic nesting remains active between C API calls. For @@ -343,13 +343,13 @@ tree, and the root span already carries the identity that tells two runs apart. | -------- | --------- | | successor's preparation | `chip.run.bind` — its arena build + host orchestration | | predecessor's device work | `chip.run.runner_run` | -| when a successor may launch | `chip.run.claim_release` | +| successor queued launch | overlapping `chip.run.runner_run` spans | | which run each belongs to | root `chip.run` attrs, joined by `(pid, inv)` | -Only `claim_release` was added for this: it wraps `release_native_run` inside -finalize, the point a successor's launch becomes admissible, and no other span -marks that boundary. `node.post_fence_retirement` covers the L3 orchestrator's -`release_run` tail for the same reason. +`claim_release` wraps `release_native_run` inside finalize and records when one +submitted-owner slot is returned. It is not a launch barrier on a backend that +queues two runs. `node.post_fence_retirement` covers the L3 orchestrator's +`release_run` tail. The identity is `run_id / dispatch_id / run_epoch / slot_id / generation`. Each field means one thing: `run_id` and `dispatch_id` are zero on the direct-chip @@ -363,9 +363,10 @@ python -m simpler_setup.tools.strace_timing path/to/log --assert-native-overlap ``` Per adjacent run on one child process, the command requires `bind(N+1)` to -**overlap** `runner_run(N)` — the intervals intersect — and `runner_run(N+1)` not -to start before `claim_release(N)`. It exits nonzero on a missing identity, a -missing span, or an ordering violation. +**overlap** `runner_run(N)` and `runner_run(N+1)` to begin before +`runner_run(N)` ends. The latter proves the successor was submitted before the +predecessor completion fence. It exits nonzero on a missing identity, a missing +span, or an ordering violation. Reading `bind` rather than the whole prepare is deliberate: `bind` sits inside prepare, so an overlap it reports is one the prepare certainly had. diff --git a/docs/dynamic-linking.md b/docs/dynamic-linking.md index 0e6f4f6377..3593a38bc7 100644 --- a/docs/dynamic-linking.md +++ b/docs/dynamic-linking.md @@ -302,7 +302,8 @@ ChipWorker.init(device_id, bins) # Python wrapper simpler_prepare_run, simpler_launch_run, simpler_poll_run, simpler_wait_run, simpler_finalize_run, simpler_run, simpler_unregister_callable, get_pipeline_contract, - supports_concurrent_native_prepare_ctx, + supports_concurrent_native_prepare_ctx, supports_queued_native_launch_ctx, + native_run_error_poisons_ctx, get_arena_bank_gm_heap_base_ctx, get_retained_temp_addr_ctx, finalize_device create_device_context() → DeviceContextHandle @@ -388,9 +389,9 @@ device_worker_main(device_id) launch_aicpu_kernel(Run) rtsLaunchCpuKernel, cached rtFuncHandle publish acceptance from the completed launch receipt simpler_poll_run(...) nonblocking child progress query - DeviceRunner::poll_execution(active) nonblocking stream query + DeviceRunner::poll_execution(active) query this run's event pair simpler_wait_run(...) - DeviceRunner::drain_execution(active) wait on both streams + DeviceRunner::drain_execution(active) wait on this run's event pair simpler_finalize_run(...) rtMemcpy results back; destroy state ChipWorker.finalize() diff --git a/docs/task-flow.md b/docs/task-flow.md index abfb7a0068..ab39a2cbf3 100644 --- a/docs/task-flow.md +++ b/docs/task-flow.md @@ -296,9 +296,10 @@ layer's. 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 +and preparing a run writes nothing to a stream. On a2a3 onboard, two successful +launches may enqueue consecutive runs on that pair; per-slot completion events +identify each run's tail while the stream orders their device work. 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. @@ -307,15 +308,15 @@ 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. +stale, and the next launch after all outstanding runs retire 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. +when two resident callables alternate. Proven owners retire in submission order. +An unproven completion prevents reuse and defers AICore stream destruction until +the submitted queue is empty. The AICPU stream carries no instruction cache +state and lives for the runner. #### Whole-run FIFO admission @@ -414,10 +415,13 @@ Backend and per-run capabilities, rather than the mailbox protocol, select between these meanings. An HBG successor's prepared token remains unlaunched and unaccepted until -`ACTIVATE`, and activation still cannot launch it until the predecessor has -polled complete and finalized. The sticky acceptance word therefore remains -zero throughout preparation. Shutdown, stale activation, and pre-launch -failure finalize the token exactly once before the frame becomes terminal. +`ACTIVATE`. On a queued-launch-capable a2a3 backend, activation may submit it +behind the predecessor before that predecessor finalizes; per-slot events keep +completion and finalization FIFO. Diagnostic, incompatible, and unsupported +paths still wait for predecessor finalization. The sticky acceptance word +therefore remains zero throughout preparation and changes only after launch. +Shutdown, stale activation, and pre-launch failure finalize the token exactly +once before the frame becomes terminal. The scheduler stages only the first eligible single NEXT_LEVEL task from the prepared FIFO successor. Tasks from the active run use only the active lane, so @@ -757,7 +761,7 @@ Step-by-step (one chip worker): | 4 | Scheduler thread | pop `slot` from worker 0's FIFO; resolve stable worker ID 0 to WT_chip_0; dispatch | | 5 | WT_chip_0 parent side | encode one leased task frame: write `config`, digest prefix, and the args blob; publish `TASK_READY` for the active lane or `PREPARE_READY` for a staged successor | | 6 | chip_0 child process | validate the frame and resolve its digest; ordinary HBG with an active predecessor also prepares the leased inactive arena bank before publishing `FRAME_STAGED`, while a frame with no active predecessor, diagnostic HBG, and TMR publish after validation and defer native prepare | -| 7 | chip_0 native-run path | after activation and the predecessor's finalization fence, launch an already-prepared HBG run or finish deferred native preparation and then launch; poll it to completion and finalize it before another staged frame may launch. Compatibility endpoints perform the equivalent operation through blocking `ChipWorker::run` | +| 7 | chip_0 native-run path | after activation, a capable a2a3 backend launches the prepared successor onto the predecessor's stream pair; per-slot events preserve FIFO completion and finalization. Diagnostic and incompatible paths wait for predecessor finalization before preparing or launching. Compatibility endpoints perform the equivalent operation through blocking `ChipWorker::run` | | 8 | runtime.so | translate host ptrs → device ptrs; dispatch AICPU / AICore; write output into `c`'s shm | | 9 | chip_0 child | native finalization returns; write `TASK_DONE` | | 10 | WT_chip_0 parent | observe `TASK_DONE`; push success completion | diff --git a/docs/worker-manager.md b/docs/worker-manager.md index 0bec5109fd..90d2311248 100644 --- a/docs/worker-manager.md +++ b/docs/worker-manager.md @@ -251,10 +251,11 @@ the same two-frame protocol but defer native prepare because their shared diagnostic or device-scratch state cannot be rewritten while another run is active. -An HBG token remains unlaunched and unaccepted until activation, and no backend -launches a successor until the predecessor is polled and finalized. Shutdown, -stale activation, and pre-launch failure finalize any unlaunched token exactly -once. +An HBG token remains unlaunched and unaccepted until activation. A backend with +queued-launch support may then submit it behind its predecessor; completion and +finalization remain FIFO. Other backends and diagnostic runs wait for the +predecessor to finalize. Shutdown, stale activation, and pre-launch failure +finalize every token exactly once. Activation is sticky on the parent side: FIFO promotion may be observed before the child reaches `FRAME_STAGED`. The endpoint records that permission and diff --git a/python/bindings/task_interface.cpp b/python/bindings/task_interface.cpp index 0da92d4996..7e330be9a8 100644 --- a/python/bindings/task_interface.cpp +++ b/python/bindings/task_interface.cpp @@ -3088,6 +3088,10 @@ NB_MODULE(_task_interface, m) { "supports_concurrent_native_prepare", &ChipWorker::supports_concurrent_native_prepare, "Whether non-diagnostic native preparation may overlap one active run in another slot." ) + .def_prop_ro( + "supports_queued_native_launch", &ChipWorker::supports_queued_native_launch, + "Whether two non-diagnostic native runs may be queued on the device streams." + ) .def_prop_ro( "runtime_buffer_addrs", &ChipWorker::runtime_buffer_addrs, "Host Runtime staging buffer address of every copy the runtime's " diff --git a/simpler_setup/tools/strace_timing.py b/simpler_setup/tools/strace_timing.py index ca1a63a2d8..898bc0e873 100644 --- a/simpler_setup/tools/strace_timing.py +++ b/simpler_setup/tools/strace_timing.py @@ -384,12 +384,10 @@ def bucket_by_hid(invocations): return buckets -# The spans one native run contributes to the overlap proof. All three already -# exist in the `chip.run` tree; only `claim_release` was added for it. +# The spans one native run contributes to the overlap proof. _PREPARE_SPAN = "chip.run.bind" _DEVICE_SPAN = "chip.run.runner_run" -_RELEASE_SPAN = "chip.run.claim_release" -_NATIVE_REQUIRED_SPANS = (_PREPARE_SPAN, _DEVICE_SPAN, _RELEASE_SPAN) +_NATIVE_REQUIRED_SPANS = (_PREPARE_SPAN, _DEVICE_SPAN) _PIPELINE_IDENTITY_FIELDS = ("run_id", "dispatch_id", "run_epoch", "slot_id", "generation") @@ -441,7 +439,8 @@ def assert_native_overlap(spans, *, require_hidden=False): * ``bind(N+1)`` overlaps ``runner_run(N)`` — the intervals intersect, which is what makes the successor's preparation concurrent with the predecessor's device work. - * ``runner_run(N+1)`` does not start before ``claim_release(N)``. + * ``runner_run(N+1)`` starts before ``runner_run(N)`` ends, proving the + successor was submitted before the predecessor's completion fence. ``bind`` is the successor's own arena build and host orchestration and sits inside its prepare, so reading it is conservative: an overlap it reports is @@ -470,7 +469,6 @@ def assert_native_overlap(spans, *, require_hidden=False): f"launch order is not monotonic: predecessor={predecessor.sequence} successor={successor.sequence}" ) pred_device = pred_spans[_DEVICE_SPAN] - pred_release = pred_spans[_RELEASE_SPAN] succ_prepare = succ_spans[_PREPARE_SPAN] succ_device = succ_spans[_DEVICE_SPAN] pred_device_end = pred_device.ts + pred_device.dur @@ -486,10 +484,10 @@ def assert_native_overlap(spans, *, require_hidden=False): f"preparation was not fully hidden: sequence={successor.sequence} prepare_end=" f"{succ_prepare_end} predecessor_device_end={pred_device_end}" ) - if succ_device.ts < pred_release.ts: + if succ_device.ts >= pred_device_end: raise NativeOverlapError( - f"device execution reordered before the claim release: sequence={successor.sequence} " - f"device_start={succ_device.ts} predecessor_release={pred_release.ts}" + f"successor was not queued before predecessor completion: sequence={successor.sequence} " + f"successor_start={succ_device.ts} predecessor_end={pred_device_end}" ) checks.append(NativeOverlapCheck(predecessor=predecessor, successor=successor)) if not checks: diff --git a/src/a2a3/platform/onboard/host/device_runner.cpp b/src/a2a3/platform/onboard/host/device_runner.cpp index 9a14a5790f..1ce9c7f616 100644 --- a/src/a2a3/platform/onboard/host/device_runner.cpp +++ b/src/a2a3/platform/onboard/host/device_runner.cpp @@ -254,7 +254,7 @@ int DeviceRunner::prepare_execution( return rc; } - ensure_device_wall_buffer(execution->kernel_args); + ensure_device_wall_buffer(execution->kernel_args, pipeline_slot); if (block_dim < 1) { LOG_ERROR("prepare_execution computed block_dim < 1 from worker_count=%d", runtime.get_worker_count()); @@ -431,8 +431,8 @@ int DeviceRunner::prepare_execution( int DeviceRunner::poll_execution(const ActiveExecution &active) { if (active.prepared == nullptr) return SIMPLER_NATIVE_RUN_POLL_ERROR; - return run_streams_.poll([](void *aicpu, void *aicore) { - return query_stream_pair_nonblocking(static_cast(aicpu), static_cast(aicore)); + return run_streams_.poll(active.prepared.get(), [this, &active](void *, void *) { + return query_run_completion_events(*active.prepared); }); } @@ -443,11 +443,14 @@ int DeviceRunner::drain_execution(ActiveExecution &active) { cleanup_execution(prepared, /*retire_aicore=*/true); }); - int rc = reap_run(); + int rc = reap_run(prepared); if (rc != 0) { - // The device/sync error remains authoritative over teardown errors. + // reap_run() marks the runner unusable before returning an error. The + // recorded event remains owned until fatal finalize abandons this + // device generation; no later launch can reach the occupied slot. return rc; } + reset_run_completion_events(prepared.pipeline_slot); // A proven-complete stream is reusable until a code publication marks it // stale. Publish retirement so cleanup does not replace it with an @@ -560,6 +563,108 @@ int DeviceRunner::destroy_run_streams() { return rc; } +int DeviceRunner::ensure_run_completion_events(uint32_t pipeline_slot) { + if (pipeline_slot >= run_completion_events_.size()) return PTO_RUNTIME_ERR_INTERNAL; + RunCompletionEvents &events = run_completion_events_[pipeline_slot]; + if (events.recorded) return PTO_RUNTIME_ERR_INTERNAL; + for (void **event : {&events.aicpu, &events.aicore}) { + if (*event != nullptr) continue; + aclrtEvent created = nullptr; + const aclError rc = aclrtCreateEventExWithFlag(&created, ACL_EVENT_CAPTURE_STREAM_PROGRESS); + if (rc != ACL_SUCCESS) { + LOG_ERROR("aclrtCreateEventExWithFlag (run completion) failed: %d", static_cast(rc)); + ACL_LOG_ERROR_DETAIL(rc); + return static_cast(rc); + } + *event = created; + } + return 0; +} + +int DeviceRunner::record_run_completion_events(const PreparedExecution &prepared) { + if (prepared.pipeline_slot >= run_completion_events_.size()) return PTO_RUNTIME_ERR_INTERNAL; + RunCompletionEvents &events = run_completion_events_[prepared.pipeline_slot]; + if (events.aicpu == nullptr || events.aicore == nullptr || events.recorded) return PTO_RUNTIME_ERR_INTERNAL; + aclError rc = aclrtRecordEvent(static_cast(events.aicore), run_streams_.aicore()); + if (rc == ACL_SUCCESS) { + rc = aclrtRecordEvent(static_cast(events.aicpu), run_streams_.aicpu()); + } + if (rc != ACL_SUCCESS) { + LOG_ERROR("aclrtRecordEvent (run completion) failed: %d", static_cast(rc)); + ACL_LOG_ERROR_DETAIL(rc); + return static_cast(rc); + } + events.recorded = true; + return 0; +} + +int DeviceRunner::query_run_completion_events(const PreparedExecution &prepared) { + if (prepared.pipeline_slot >= run_completion_events_.size()) return SIMPLER_NATIVE_RUN_POLL_ERROR; + const RunCompletionEvents &events = run_completion_events_[prepared.pipeline_slot]; + if (!events.recorded || events.aicpu == nullptr || events.aicore == nullptr) { + return SIMPLER_NATIVE_RUN_POLL_ERROR; + } + for (void *event : {events.aicpu, events.aicore}) { + aclrtEventRecordedStatus status = ACL_EVENT_RECORDED_STATUS_NOT_READY; + const aclError rc = aclrtQueryEventStatus(static_cast(event), &status); + if (rc != ACL_SUCCESS) { + LOG_ERROR("aclrtQueryEventStatus (run completion) failed: %d", static_cast(rc)); + ACL_LOG_ERROR_DETAIL(rc); + return SIMPLER_NATIVE_RUN_POLL_ERROR; + } + if (status != ACL_EVENT_RECORDED_STATUS_COMPLETE) return SIMPLER_NATIVE_RUN_POLL_NOT_READY; + } + return SIMPLER_NATIVE_RUN_POLL_COMPLETE; +} + +int DeviceRunner::wait_run_completion_events(const PreparedExecution &prepared) { + if (prepared.pipeline_slot >= run_completion_events_.size()) return PTO_RUNTIME_ERR_INTERNAL; + const RunCompletionEvents &events = run_completion_events_[prepared.pipeline_slot]; + if (!events.recorded || events.aicpu == nullptr || events.aicore == nullptr) return PTO_RUNTIME_ERR_INTERNAL; + for (const auto &[name, event] : + {std::pair{"AICPU", events.aicpu}, {"AICore", events.aicore}}) { + const aclError rc = + aclrtSynchronizeEventWithTimeout(static_cast(event), timeout_config_.stream_sync_timeout_ms); + if (rc != ACL_SUCCESS) { + LOG_ERROR( + "Event sync failed: event=%s timeout_ms=%d device_id=%d block_dim=%d rc=%d", name, + timeout_config_.stream_sync_timeout_ms, device_id_, block_dim_, static_cast(rc) + ); + ACL_LOG_ERROR_DETAIL(rc); + return static_cast(rc); + } + } + return 0; +} + +void DeviceRunner::reset_run_completion_events(uint32_t pipeline_slot) noexcept { + if (pipeline_slot >= run_completion_events_.size()) return; + RunCompletionEvents &events = run_completion_events_[pipeline_slot]; + events.recorded = false; +} + +int DeviceRunner::destroy_run_completion_events() { + int first_error = 0; + for (RunCompletionEvents &events : run_completion_events_) { + for (void **event : {&events.aicpu, &events.aicore}) { + if (*event == nullptr) continue; + const aclError rc = aclrtDestroyEvent(static_cast(*event)); + if (rc != ACL_SUCCESS) { + if (first_error == 0) first_error = static_cast(rc); + continue; + } + *event = nullptr; + } + events.recorded = false; + } + return first_error; +} + +void DeviceRunner::abandon_run_completion_events() noexcept { + for (RunCompletionEvents &events : run_completion_events_) + events = RunCompletionEvents{}; +} + DeviceRunnerBase::LaunchOutcome DeviceRunner::launch_execution(std::unique_ptr prepared, LaunchPermit permit) { LaunchOutcome outcome; @@ -587,10 +692,10 @@ LaunchTransactionResult DeviceRunner::launch_run(PreparedExecution &prepared, La // It intentionally performs no stream synchronization or per-run cleanup. // // 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) { + // point the caller owns a submitted-run slot. A prepared successor overlaps + // its predecessor's execution, and a queued successor must reuse the same + // warm pair rather than replace a stream with outstanding work. + if (ensure_run_streams() != 0 || ensure_run_completion_events(prepared.pipeline_slot) != 0) { return LaunchTransactionResult{}; } RunStreamSet streams{static_cast(run_streams_.aicpu()), static_cast(run_streams_.aicore())}; @@ -627,12 +732,11 @@ LaunchTransactionResult DeviceRunner::launch_run(PreparedExecution &prepared, La // slow-launch / 207001 wedge was measured on a5; this mirror is UNVERIFIED on // a2a3 silicon (the dev box is a5-only), relying on CI. See // docs/investigations/2026-06-pa-unroll-207001-optimeout-window.md. - // The AICore publishes aicore_done on launch (gated by nothing), and the - // workers region persists across runs in the pooled arena. Clearing each - // worker's aicore_done before the AICore kernel launches keeps the AICPU's - // handshake sweep from reading a prior run's report — which would open a - // window on that run's physical_core_id. Only aicore_done needs clearing; the - // AICore overwrites physical_core_id/core_type in the same report. + // Each prepared Runtime owns its inline workers array and its + // device runtime_args allocation, so a queued successor clears + // only its own slot. Clearing aicore_done before launch keeps + // this run's AICPU sweep from reading that slot's prior report. + // The AICore overwrites physical_core_id/core_type alongside it. Handshake *workers = runtime.get_workers(); for (int i = 0; i < num_aicore; i++) workers[i].aicore_done = 0; @@ -667,21 +771,22 @@ LaunchTransactionResult DeviceRunner::launch_run(PreparedExecution &prepared, La if (launch_rc != 0) { LOG_ERROR("launch_aicpu_kernel (main) failed: %d", launch_rc); } - return launch_rc; + if (launch_rc != 0) return launch_rc; + return record_run_completion_events(prepared); } ); return result; } -int DeviceRunner::reap_run() { +int DeviceRunner::reap_run(PreparedExecution &prepared) { if (!run_streams_.ready()) { LOG_ERROR("reap_run: the run stream pair is not ready"); return PTO_RUNTIME_ERR_INTERNAL; } - int rc = sync_stream_pair(run_streams_.aicpu(), run_streams_.aicore()); + int rc = wait_run_completion_events(prepared); 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 + // The event wait surfaces the AICore op-timeout (STARS-reaped op -> + // 507000/507018/507046 at the recorded stream completion point). The op-timeout // leaves the device context poisoned for the SAME DeviceRunner's next // run, so attempt recovery / mark-unusable here too, not only on the // launch-error path above. @@ -698,7 +803,7 @@ int DeviceRunner::reap_run() { return rc; } - read_device_wall_ns(); + read_device_wall_ns(prepared.kernel_args); // Tear down collectors. stop() joins mgmt then collector in the only safe // order (mgmt's final-drain pass into L2 has poll as its consumer). @@ -1033,6 +1138,7 @@ int DeviceRunner::finalize() { } run_streams_.abandon(); + abandon_run_completion_events(); int abandon_rc = abandon_common_after_device_failure(); // Only finalize the ACL owner after force reset established a clean @@ -1073,12 +1179,14 @@ int DeviceRunner::finalize() { // 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 event_rc = destroy_run_completion_events(); int stream_rc = destroy_run_streams(); // Shared cleanup body — streams, kernel_args, callable/orch maps, // chip-callable buffer pool, the three arenas, device_wall, // mem_alloc_.finalize(), and cached arena sizes. rc = finalize_common(); + if (rc == 0) rc = event_rc; if (rc == 0) rc = stream_rc; // Reset device AFTER all device memory is freed. Two paths: diff --git a/src/a2a3/platform/onboard/host/device_runner.h b/src/a2a3/platform/onboard/host/device_runner.h index 8f7c33babf..dc6f230a54 100644 --- a/src/a2a3/platform/onboard/host/device_runner.h +++ b/src/a2a3/platform/onboard/host/device_runner.h @@ -24,6 +24,7 @@ #include #include +#include #include #include #include @@ -103,6 +104,7 @@ 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); } + size_t native_launch_depth() const override { return PTO_PIPELINE_MAX_DEPTH; } // 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. @@ -224,11 +226,10 @@ class DeviceRunner : public DeviceRunnerBase { rtStream_t aicore{nullptr}; }; - // 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. + // One pair carries every run. Up to two owners may submit consecutive work; + // the stream order serializes it while per-slot events expose each run's + // completion point. 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); @@ -247,6 +248,20 @@ class DeviceRunner : public DeviceRunnerBase { int retire_run_aicore_stream(const void *owner, RunStreamPair::CompletionStatus completion_status); int destroy_run_streams(); + struct RunCompletionEvents { + void *aicpu{nullptr}; + void *aicore{nullptr}; + bool recorded{false}; + }; + std::array run_completion_events_{}; + int ensure_run_completion_events(uint32_t pipeline_slot); + int record_run_completion_events(const PreparedExecution &prepared); + int query_run_completion_events(const PreparedExecution &prepared); + int wait_run_completion_events(const PreparedExecution &prepared); + void reset_run_completion_events(uint32_t pipeline_slot) noexcept; + int destroy_run_completion_events(); + void abandon_run_completion_events() noexcept; + // Release execution-owned resources in collector, runtime-argument, // register-buffer, then stream order. The collectors this releases were // initialized by prepare_execution() for this run alone; an overlapping @@ -257,7 +272,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(); + int reap_run(PreparedExecution &prepared); // 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/a2a3/runtime/host_build_graph/host/runtime_maker.cpp b/src/a2a3/runtime/host_build_graph/host/runtime_maker.cpp index 46f9700860..601cb1e728 100644 --- a/src/a2a3/runtime/host_build_graph/host/runtime_maker.cpp +++ b/src/a2a3/runtime/host_build_graph/host/runtime_maker.cpp @@ -350,6 +350,15 @@ static int32_t read_runtime_status(Runtime *runtime, const HostApi *api, SharedM return runtime_status_from_error_codes(orch_error_code, sched_error_code); } +extern "C" int completed_runtime_status_impl(Runtime *runtime, const HostApi *api) { + SharedMemoryHeader host_header{}; + return read_runtime_status(runtime, api, &host_header); +} + +extern "C" int runtime_status_poisons_device_impl(int runtime_status) { + return latched_error_may_poison_device(-runtime_status) ? 1 : 0; +} + namespace { // host_build_graph is host-orchestration-first: the HOST dlopens the diff --git a/src/a2a3/runtime/tensormap_and_ringbuffer/host/runtime_maker.cpp b/src/a2a3/runtime/tensormap_and_ringbuffer/host/runtime_maker.cpp index 74987b1f84..cb364ed17f 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/host/runtime_maker.cpp +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/host/runtime_maker.cpp @@ -221,6 +221,15 @@ static int32_t read_runtime_status(Runtime *runtime, const HostApi *api, SharedM return runtime_status_from_error_codes(orch_error_code, sched_error_code); } +extern "C" int completed_runtime_status_impl(Runtime *runtime, const HostApi *api) { + SharedMemoryHeader host_header{}; + return read_runtime_status(runtime, api, &host_header); +} + +extern "C" int runtime_status_poisons_device_impl(int runtime_status) { + return latched_error_may_poison_device(-runtime_status) ? 1 : 0; +} + static void release_tensor_leases(Runtime *runtime, const HostApi *api) { int freed = 0; int buffer_noop = 0; diff --git a/src/a5/platform/onboard/host/device_runner.cpp b/src/a5/platform/onboard/host/device_runner.cpp index 6e178ac213..a203e956ee 100644 --- a/src/a5/platform/onboard/host/device_runner.cpp +++ b/src/a5/platform/onboard/host/device_runner.cpp @@ -267,7 +267,7 @@ int DeviceRunner::prepare_execution( return rc; } - ensure_device_wall_buffer(execution->kernel_args); + ensure_device_wall_buffer(execution->kernel_args, pipeline_slot); if (block_dim < 1) { LOG_ERROR("prepare_execution computed block_dim < 1 from worker_count=%d", runtime.get_worker_count()); @@ -590,7 +590,7 @@ int DeviceRunner::drain_execution(ActiveExecution &active) { return rc; } - read_device_wall_ns(); + read_device_wall_ns(prepared.kernel_args); teardown_shared_collectors_after_run(true); // a5-specific dep_gen teardown: stop + reconcile + replay emit. diff --git a/src/common/platform/include/common/device_phase.h b/src/common/platform/include/common/device_phase.h index 292bfdcf6c..1bf14c3ddf 100644 --- a/src/common/platform/include/common/device_phase.h +++ b/src/common/platform/include/common/device_phase.h @@ -130,7 +130,7 @@ reduce_aicpu_phase_windows(const AicpuPhaseRecord *buf, int threads, uint64_t *o // ============================================================================= // // A fixed tail appended after the AicpuPhaseRecord region in the SAME device -// buffer (same base pointer and per-run H2D reset). The header controls whether +// per-slot buffer (one base per in-flight slot and a per-run H2D reset). The header controls whether // the host performs the tail's separate post-sync D2H copy. // Orchestration tags selected tasks with a slot id 0..NUM_TASK_TIMING_SLOTS-1; // the scheduler folds each tagged task's AICPU dispatch/finish cycles into that diff --git a/src/common/platform/include/host/run_stream_pair.h b/src/common/platform/include/host/run_stream_pair.h index 31a4cf8466..8ad5588776 100644 --- a/src/common/platform/include/host/run_stream_pair.h +++ b/src/common/platform/include/host/run_stream_pair.h @@ -11,6 +11,8 @@ #pragma once +#include +#include #include #include #include @@ -22,11 +24,10 @@ /** * 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. + * A stream is an ordered queue, so consecutive runs share one pair and may be + * submitted before their predecessors complete. The pair is not indexed by + * pipeline slot: slots select run-owned preparation state and completion + * events, while the stream order serializes device execution. * * 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 @@ -38,9 +39,9 @@ * 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. + * Submitted owners retire in FIFO order after proven completion. An unproven + * retirement poisons reuse and defers stream destruction until no queued owner + * remains. A run that never submitted leaves 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 @@ -51,6 +52,19 @@ * without a device. */ class RunStreamPair { +private: + struct Submission { + const void *owner{nullptr}; + bool complete{false}; + }; + Submission *find_submission(const void *owner) { + auto end = submissions_.begin() + static_cast(submission_count_); + auto match = std::find_if(submissions_.begin(), end, [owner](const Submission &submission) { + return submission.owner == owner; + }); + return match == end ? nullptr : &*match; + } + public: using CreateFn = std::function; using DestroyFn = std::function; @@ -63,7 +77,8 @@ class RunStreamPair { /** * 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. + * A warm pair accepts another submission while earlier runs remain queued. + * A stale pair can only be replaced after every submitted run retires. */ int ensure() { std::lock_guard lock(mutex_); @@ -75,13 +90,8 @@ class RunStreamPair { } } 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 PTO_RUNTIME_ERR_INTERNAL; + if (submission_count_ != 0) return stale_ || unproven_ ? PTO_RUNTIME_ERR_INTERNAL : 0; if (!stale_) { - submitted_ = false; - complete_ = false; return 0; } int rc = destroy_(aicore_); @@ -95,8 +105,7 @@ class RunStreamPair { } created_count_.fetch_add(1, std::memory_order_relaxed); stale_ = false; - submitted_ = false; - complete_ = false; + unproven_ = false; return 0; } @@ -106,14 +115,16 @@ class RunStreamPair { stale_ = true; } - /** Make the pair visible to non-blocking poll and record its submitter. */ + /** Append one submitter to the stream-ordered in-flight FIFO. */ int mark_submitted(const void *owner) { if (owner == nullptr) return PTO_RUNTIME_ERR_INTERNAL; std::lock_guard lock(mutex_); if (aicpu_ == nullptr || aicore_ == nullptr) return PTO_RUNTIME_ERR_INTERNAL; - owner_ = owner; - submitted_ = true; - complete_ = false; + if (stale_ || unproven_ || submission_count_ >= submissions_.size()) { + return PTO_RUNTIME_ERR_INTERNAL; + } + if (find_submission(owner) != nullptr) return PTO_RUNTIME_ERR_INTERNAL; + submissions_[submission_count_++] = Submission{owner, false}; return 0; } @@ -125,15 +136,16 @@ class RunStreamPair { * missing handle as an error. */ template - int poll(QueryPairFn &&query) { + int poll(const void *owner, 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) { + Submission *submission = find_submission(owner); + if (submission == nullptr || aicpu_ == nullptr || aicore_ == nullptr) { return SIMPLER_NATIVE_RUN_POLL_ERROR; } + if (submission->complete) return SIMPLER_NATIVE_RUN_POLL_COMPLETE; const int rc = std::forward(query)(aicpu_, aicore_); - if (rc == SIMPLER_NATIVE_RUN_POLL_COMPLETE) complete_ = true; + if (rc == SIMPLER_NATIVE_RUN_POLL_COMPLETE) submission->complete = true; return rc; } @@ -147,29 +159,36 @@ class RunStreamPair { */ int retire(CompletionStatus completion_status, const void *owner) { std::lock_guard lock(mutex_); - if (owner == nullptr || owner_ != owner) return 0; + Submission *submission = find_submission(owner); + if (submission == nullptr) return 0; + const size_t index = static_cast(submission - submissions_.data()); + const bool order_error = completion_status == CompletionStatus::Complete && index != 0; + if (order_error) completion_status = CompletionStatus::Unproven; // 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; + if (completion_status == CompletionStatus::Unproven) unproven_ = true; + for (size_t i = index + 1; i < submission_count_; ++i) + submissions_[i - 1] = submissions_[i]; + submissions_[--submission_count_] = Submission{}; + if (submission_count_ != 0 || !unproven_ || aicore_ == nullptr) { + return order_error ? PTO_RUNTIME_ERR_INTERNAL : 0; + } int rc = destroy_(aicore_); if (rc != 0) return rc; aicore_ = nullptr; - return 0; + unproven_ = false; + return order_error ? PTO_RUNTIME_ERR_INTERNAL : 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; + submissions_.fill(Submission{}); + submission_count_ = 0; + unproven_ = false; for (void **stream : {&aicpu_, &aicore_}) { if (*stream == nullptr) continue; int rc = destroy_(*stream); @@ -191,9 +210,9 @@ class RunStreamPair { aicpu_ = nullptr; aicore_ = nullptr; stale_ = false; - submitted_ = false; - complete_ = false; - owner_ = nullptr; + submissions_.fill(Submission{}); + submission_count_ = 0; + unproven_ = false; } // Handle reads are unsynchronized: the claim holder is the only writer once @@ -207,11 +226,10 @@ class RunStreamPair { 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}; + bool unproven_{false}; + std::array submissions_{}; + size_t submission_count_{0}; CreateFn create_; DestroyFn destroy_; diff --git a/src/common/platform/onboard/host/c_api_shared.cpp b/src/common/platform/onboard/host/c_api_shared.cpp index af658e6ab2..4c89b8083e 100644 --- a/src/common/platform/onboard/host/c_api_shared.cpp +++ b/src/common/platform/onboard/host/c_api_shared.cpp @@ -66,6 +66,11 @@ extern "C" { * =========================================================================== */ int register_callable_impl(const ChipCallable *callable, const HostApi *api, CallableArtifacts *out); int validate_runtime_impl(Runtime *runtime, const HostApi *api, int execution_rc); +__attribute__((weak)) int completed_runtime_status_impl(Runtime * /*runtime*/, const HostApi * /*api*/) { return 0; } +// Backends that do not report a per-run status keep the conservative policy. +// a5 still completes through stream sync and never reaches this fallback for a +// successful drain; a future queued-launch backend must provide both hooks. +__attribute__((weak)) int runtime_status_poisons_device_impl(int /*runtime_status*/) { return 1; } __attribute__((weak)) int concurrent_native_prepare_supported_impl(void) { return 0; } __attribute__((weak)) int prepared_run_config_compatible_impl( const HostApi * /*api*/, const uint64_t * /*ring_task_window*/, const uint64_t * /*ring_heap*/, @@ -643,10 +648,36 @@ static void emit_native_run_runner_wall(OnboardNativeRunContext *state) { state->runner_trace_start_ns = 0; } +static int completed_execution_rc(OnboardNativeRunContext *state, int drain_rc) { + if (drain_rc != 0) return drain_rc; + // Per-run events delimit the completed stream prefix but do not carry the + // runtime's device-side error latches. Read this run's small status header + // before validation decides whether outputs are safe to copy back. + const int runtime_status = completed_runtime_status_impl(&state->runtime, &state->host_api); + if (runtime_status == 0) return 0; + if (runtime_status_poisons_device_impl(runtime_status) != 0) { + state->runner->recover_device_or_mark_unusable(runtime_status); + } + return runtime_status; +} + +int native_run_error_poisons_ctx(DeviceContextHandle ctx, int execution_rc) { + if (ctx == nullptr || execution_rc == 0) return 0; + if (execution_rc <= -1 && execution_rc >= -PTO_RUNTIME_LATCHED_CODE_MAX) { + return runtime_status_poisons_device_impl(execution_rc) != 0 ? 1 : 0; + } + return 1; +} + int supports_concurrent_native_prepare_ctx(DeviceContextHandle ctx) { return ctx != nullptr && concurrent_native_prepare_supported_impl() != 0 ? 1 : 0; } +int supports_queued_native_launch_ctx(DeviceContextHandle ctx) { + if (ctx == nullptr) return 0; + return static_cast(ctx)->native_launch_depth() > 1 ? 1 : 0; +} + static int cleanup_failed_prepare(OnboardNativeRunContext *state, int execution_rc, bool clear_gm_sm) { const uint64_t trace_inv = state->trace_inv; const uint64_t trace_hid = state->trace_hid; @@ -932,6 +963,7 @@ int simpler_wait_run(DeviceContextHandle ctx, RuntimeHandle runtime) { drain_rc = PTO_RUNTIME_ERR_INTERNAL; LOG_ERROR("simpler_wait_run: drain threw (%s)", state->trace_attrs); } + drain_rc = completed_execution_rc(state, drain_rc); if (state->completion_rc == 0) state->completion_rc = drain_rc; state->phase.store(NativeRunPhase::Complete, std::memory_order_release); emit_native_run_runner_wall(state); @@ -980,6 +1012,7 @@ int simpler_finalize_run(DeviceContextHandle ctx, RuntimeHandle runtime) { LOG_ERROR("simpler_finalize_run: drain_execution threw (%s)", state->trace_attrs); } } + drain_rc = completed_execution_rc(state, drain_rc); if (execution_rc == 0) execution_rc = drain_rc; state->completion_rc = execution_rc; state->phase.store(NativeRunPhase::Complete, std::memory_order_release); @@ -1027,9 +1060,7 @@ int simpler_finalize_run(DeviceContextHandle ctx, RuntimeHandle runtime) { // the provider/session. state->runner->finish_clock_correlation_session(false, !state->runner->can_accept_run()); if (state->runner_claimed) { - // The point a successor's launch becomes admissible. Ordering a - // successor's device work against this boundary is what separates a - // pipelined launch from a reordered one, and no other span marks it. + // This run no longer occupies one slot in the submitted-run owner set. STRACE("chip.run.claim_release"); state->runner->release_native_run(state); state->runner_claimed = false; diff --git a/src/common/platform/onboard/host/device_runner_base.cpp b/src/common/platform/onboard/host/device_runner_base.cpp index e9710ae1d5..4ea830c143 100644 --- a/src/common/platform/onboard/host/device_runner_base.cpp +++ b/src/common/platform/onboard/host/device_runner_base.cpp @@ -1506,11 +1506,10 @@ int DeviceRunnerBase::finalize_common_impl(bool abandon_device_resources) { // mem_alloc_ and the device context are still live. free_tensor() routes // through mem_alloc_.free(), so it must run before mem_alloc_.finalize() // and before the subclass's `rtDeviceReset()` tears down the device runtime. - if (device_wall_dev_ptr_ != nullptr) { - if (!abandon_device_resources) { - free_tensor(device_wall_dev_ptr_); - } - device_wall_dev_ptr_ = nullptr; + for (void *&device_wall_dev_ptr : device_wall_dev_ptrs_) { + if (device_wall_dev_ptr == nullptr) continue; + if (!abandon_device_resources) free_tensor(device_wall_dev_ptr); + device_wall_dev_ptr = nullptr; } // Free all remaining allocations (including handshake buffer and binGmAddr) @@ -1615,7 +1614,7 @@ int DeviceRunnerBase::resolve_aicpu_thread_num(int requested, int usable, int ar return total; } -void DeviceRunnerBase::ensure_device_wall_buffer(KernelArgsHelper &kernel_args) { +void DeviceRunnerBase::ensure_device_wall_buffer(KernelArgsHelper &kernel_args, uint32_t pipeline_slot) { if (!device_phase_capture_enabled()) { // A null base makes the AICPU stamping helpers no-op. kernel_args.args.device_wall_data_base = 0; @@ -1627,22 +1626,27 @@ void DeviceRunnerBase::ensure_device_wall_buffer(KernelArgsHelper &kernel_args) // the on-NPU portion. Each surviving AICPU thread writes its own records // (plain stores, no atomics); read_device_phases() reduces RunWall as // max(end) - min(start) and surfaces the other phases as trace markers. The - // buffer is allocated once (lazy) but RESET every run so a stale prior run + // buffer is allocated once per slot (lazy) but RESET every run so a stale prior run // cannot leak into the reduction. constexpr int kThreads = PLATFORM_MAX_AICPU_THREADS_JUST_FOR_LAUNCH; using BufferImage = DevicePhaseBufferStorage; constexpr size_t kBytes = device_phase_buffer_bytes(kThreads); static_assert(sizeof(BufferImage) == kBytes, "device-phase buffer layout drift"); - if (device_wall_dev_ptr_ == nullptr) { - device_wall_dev_ptr_ = allocate_tensor(kBytes); + if (pipeline_slot >= device_wall_dev_ptrs_.size()) { + kernel_args.args.device_wall_data_base = 0; + return; + } + void *&device_wall_dev_ptr = device_wall_dev_ptrs_[pipeline_slot]; + if (device_wall_dev_ptr == nullptr) { + device_wall_dev_ptr = allocate_tensor(kBytes); } - if (device_wall_dev_ptr_ != nullptr) { - kernel_args.args.device_wall_data_base = reinterpret_cast(device_wall_dev_ptr_); + if (device_wall_dev_ptr != nullptr) { + kernel_args.args.device_wall_data_base = reinterpret_cast(device_wall_dev_ptr); } } int DeviceRunnerBase::arm_device_wall_buffer(KernelArgsHelper &kernel_args) { - if (device_wall_dev_ptr_ == nullptr || kernel_args.args.device_wall_data_base == 0) return 0; + if (kernel_args.args.device_wall_data_base == 0) return 0; constexpr int kThreads = PLATFORM_MAX_AICPU_THREADS_JUST_FOR_LAUNCH; using BufferImage = DevicePhaseBufferStorage; static const BufferImage init = [] { @@ -1650,10 +1654,11 @@ int DeviceRunnerBase::arm_device_wall_buffer(KernelArgsHelper &kernel_args) { reset_device_phase_buffer(&image, kThreads); return image; }(); - if (copy_to_device(device_wall_dev_ptr_, &init, sizeof(init)) != 0) { + void *device_wall_dev_ptr = reinterpret_cast(kernel_args.args.device_wall_data_base); + if (copy_to_device(device_wall_dev_ptr, &init, sizeof(init)) != 0) { // Reset failed — disable capture for this run so stale slot data - // can't leak into the reduction. Keep the shared allocation alive: - // an earlier run may still reference it, and the next run retries reset. + // can't leak into the reduction. Keep the slot allocation alive so its + // next owner can retry the reset. LOG_WARN("device_phase reset H2D failed; disabling phase capture this run"); kernel_args.args.device_wall_data_base = 0; return 0; @@ -1759,7 +1764,7 @@ int DeviceRunnerBase::sync_stream_pair(rtStream_t aicpu_stream, rtStream_t aicor return 0; } -void DeviceRunnerBase::read_device_wall_ns() { +void DeviceRunnerBase::read_device_wall_ns(const KernelArgsHelper &kernel_args) { // Pull the per-thread AICPU phase records back from the device buffer that // AICPU writes through via KernelArgs::device_wall_data_base. (We can't use // the device_k_args_ shadow here — CANN's rtAicpuKernelLaunchExWithArgs @@ -1776,13 +1781,14 @@ void DeviceRunnerBase::read_device_wall_ns() { task_slot_finish_ns_[s] = 0; } if (!device_phase_capture_enabled()) return; - if (device_wall_dev_ptr_ == nullptr) return; + if (kernel_args.args.device_wall_data_base == 0) return; + const void *device_wall_dev_ptr = reinterpret_cast(kernel_args.args.device_wall_data_base); constexpr int kThreads = PLATFORM_MAX_AICPU_THREADS_JUST_FOR_LAUNCH; using BufferPrefix = DevicePhaseBufferPrefixStorage; static_assert(sizeof(BufferPrefix) == task_timing_tail_offset(kThreads), "device-phase prefix layout drift"); BufferPrefix buf{}; - int wall_rc = rtMemcpy(&buf, sizeof(buf), device_wall_dev_ptr_, sizeof(buf), RT_MEMCPY_DEVICE_TO_HOST); + int wall_rc = rtMemcpy(&buf, sizeof(buf), device_wall_dev_ptr, sizeof(buf), RT_MEMCPY_DEVICE_TO_HOST); if (wall_rc != 0) { LOG_WARN("rtMemcpy(device_phase) D2H failed: %d", wall_rc); return; @@ -1820,7 +1826,7 @@ void DeviceRunnerBase::read_device_wall_ns() { int tail_rc = read_task_timing_tail_if_used(buf.header, [&]() { TaskTimingRecord tail[kTailRecords] = {}; const void *tail_src = - reinterpret_cast(device_wall_dev_ptr_) + task_timing_tail_offset(kThreads); + reinterpret_cast(device_wall_dev_ptr) + task_timing_tail_offset(kThreads); int rc = rtMemcpy(tail, sizeof(tail), tail_src, sizeof(tail), RT_MEMCPY_DEVICE_TO_HOST); if (rc != 0) return rc; resolve_task_timing_slots_ns( @@ -1931,31 +1937,37 @@ bool DeviceRunnerBase::try_acquire_native_run( } } if (!reserved) return false; - const void *expected = nullptr; - if (!active_native_run_.compare_exchange_strong( - expected, owner, std::memory_order_acq_rel, std::memory_order_acquire - )) { - return false; + + size_t active_count = 0; + for (const void *active : active_native_runs_) { + if (active == owner) return false; + if (active != nullptr) ++active_count; } + if (active_count >= native_launch_depth()) return false; + auto slot = std::find(active_native_runs_.begin(), active_native_runs_.end(), nullptr); + if (slot == active_native_runs_.end()) return false; + *slot = owner; *permit = LaunchPermit(identity); return true; } void DeviceRunnerBase::release_native_run(const void *owner) { std::lock_guard lk(native_run_mu_); - if (active_native_run_.load(std::memory_order_acquire) != owner) return; - const void *expected = owner; - (void)active_native_run_.compare_exchange_strong( - expected, nullptr, std::memory_order_release, std::memory_order_relaxed - ); + auto slot = std::find(active_native_runs_.begin(), active_native_runs_.end(), owner); + if (slot != active_native_runs_.end()) *slot = nullptr; } bool DeviceRunnerBase::native_run_active() const { - return active_native_run_.load(std::memory_order_acquire) != nullptr; + std::lock_guard lk(native_run_mu_); + return std::any_of(active_native_runs_.begin(), active_native_runs_.end(), [](const void *owner) { + return owner != nullptr; + }); } bool DeviceRunnerBase::native_run_owned_by(const void *owner) const { - return owner != nullptr && active_native_run_.load(std::memory_order_acquire) == owner; + if (owner == nullptr) return false; + std::lock_guard lk(native_run_mu_); + return std::find(active_native_runs_.begin(), active_native_runs_.end(), owner) != active_native_runs_.end(); } bool DeviceRunnerBase::try_reserve_native_run( @@ -1977,9 +1989,12 @@ bool DeviceRunnerBase::try_reserve_native_run( existing = &reservation; } if (occupied != 0) { - const void *active = active_native_run_.load(std::memory_order_acquire); + const bool existing_active = + std::any_of(active_native_runs_.begin(), active_native_runs_.end(), [existing](const void *active_owner) { + return active_owner == existing->owner; + }); if (!allow_prepared_successor || occupied != 1 || existing == nullptr || - !existing->permits_prepared_successor || active != existing->owner) { + !existing->permits_prepared_successor || !existing_active) { return false; } } diff --git a/src/common/platform/onboard/host/device_runner_base.h b/src/common/platform/onboard/host/device_runner_base.h index 4a96d8d74a..738bc5f2f8 100644 --- a/src/common/platform/onboard/host/device_runner_base.h +++ b/src/common/platform/onboard/host/device_runner_base.h @@ -97,9 +97,8 @@ class DeviceRunnerBase { DeviceRunnerBase &operator=(DeviceRunnerBase &&) = delete; /** - * Claim the runner for one native execution. The opaque owner and - * runner-owned timing and diagnostic state remain exclusive through - * validation/finalize. + * Claim one submitted-run slot and mint its identity-bound launch permit. + * The backend's native_launch_depth() controls how many owners coexist. */ bool try_acquire_native_run(const void *owner, const NativeRunIdentity &identity, LaunchPermit *permit); void release_native_run(const void *owner); @@ -109,7 +108,7 @@ class DeviceRunnerBase { /** * Reserve caller-owned native-run storage before binding starts. A * concurrent reservation is admitted only while the first reservation - * owns the execution claim and selects a distinct pipeline slot. A backend + * owns a submitted-run slot and selects a distinct pipeline slot. A backend * that shares an arena bank must reject or defer incompatible preparation * before mutating that bank. */ @@ -526,6 +525,9 @@ class DeviceRunnerBase { */ virtual bool can_accept_run() const = 0; + /** Maximum number of submitted native runs retained by this runner. */ + virtual size_t native_launch_depth() const { return 1; } + /** * An AICore launch or stream sync failed outside the per-run path. The arch * runner drains what it can and flips its device-unusable flag, so the next @@ -861,7 +863,7 @@ class DeviceRunnerBase { * reset every record, and publish the base for AICPU stamping. Allocation or * reset failure is non-fatal; the base stays null and timing reads as 0. */ - void ensure_device_wall_buffer(KernelArgsHelper &kernel_args); + void ensure_device_wall_buffer(KernelArgsHelper &kernel_args, uint32_t pipeline_slot); int arm_device_wall_buffer(KernelArgsHelper &kernel_args); /** @@ -900,11 +902,11 @@ class DeviceRunnerBase { int sync_stream_pair(rtStream_t aicpu_stream, rtStream_t aicore_stream); /** - * Read and reduce the device-phase/task-timing records after stream sync. + * Read and reduce one run's device-phase/task-timing records after its fence. * Capture-disabled runs and missing buffers leave all cached timings at 0. * A D2H failure is a soft warning and also leaves timing at 0. */ - void read_device_wall_ns(); + void read_device_wall_ns(const KernelArgsHelper &kernel_args); /** * H2D the Runtime struct via the supplied per-execution kernel arguments. Log config @@ -958,7 +960,7 @@ class DeviceRunnerBase { * - chip_callable_buffers_ free + clear * - callables_ dlclose-on-hbg + clear + aicpu counter reset * - 3 arenas release + cached size reset - * - device_wall_dev_ptr_ free (before mem_alloc_.finalize) + * - per-slot device phase buffer frees (before mem_alloc_.finalize) * - mem_alloc_.finalize * - block_dim_, worker_count_, aicore_kernel_binary_ reset * @@ -1074,7 +1076,7 @@ class DeviceRunnerBase { }; mutable std::mutex native_run_mu_; std::array native_run_reservations_{}; - std::atomic active_native_run_{nullptr}; + std::array active_native_runs_{}; // ---- State shared by both a2a3 and a5 --------------------------------- // @@ -1218,10 +1220,10 @@ class DeviceRunnerBase { // Platform-level device phase buffer: a header, thread-major phase records, // and the optional task-timing tail. Its address rides on // `KernelArgs.device_wall_data_base`. AICPU stamps raw sys-counter cycles; - // subclass drain always pulls back the header + phases after stream sync, + // subclass drain pulls back the completed run's header + phases after its fence, // and only pulls the tail when the header marks it used. Allocated lazily - // on the first capture-enabled run and freed in subclass `finalize()`. - void *device_wall_dev_ptr_{nullptr}; + // once per pipeline slot and freed in subclass `finalize()`. + std::array device_wall_dev_ptrs_{}; uint64_t device_wall_ns_{0}; uint64_t device_phase_ns_[NUM_AICPU_PHASES] = {0}; // Per-phase start offset (ns) from the earliest sub-phase start; see diff --git a/src/common/platform/sim/host/c_api_shared.cpp b/src/common/platform/sim/host/c_api_shared.cpp index 14103f20fe..302f9e3847 100644 --- a/src/common/platform/sim/host/c_api_shared.cpp +++ b/src/common/platform/sim/host/c_api_shared.cpp @@ -887,6 +887,10 @@ int simpler_run( int supports_concurrent_native_prepare_ctx(DeviceContextHandle) { return 0; } +int supports_queued_native_launch_ctx(DeviceContextHandle) { return 0; } + +int native_run_error_poisons_ctx(DeviceContextHandle, int execution_rc) { return execution_rc != 0 ? 1 : 0; } + uint64_t get_arena_bank_gm_heap_base_ctx(DeviceContextHandle ctx, uint32_t bank_id) { if (ctx == NULL) return 0; return static_cast(ctx)->arena_bank_gm_heap_base(bank_id); diff --git a/src/common/runtime_status/error_names.h b/src/common/runtime_status/error_names.h index 9810da9687..7ca44f5146 100644 --- a/src/common/runtime_status/error_names.h +++ b/src/common/runtime_status/error_names.h @@ -196,4 +196,36 @@ static inline const char *latched_error_field(int32_t orch_error_code, int32_t s return orch_error_code != SIMPLER_ERROR_NONE ? "orch_error_code" : "sched_error_code"; } +// A completed event pair proves both stream prefixes retired. Only timeout +// classifications may have reached that point through STARS reaping or a +// stalled producer that leaves the device generation unsafe to reuse. Unknown +// codes stay conservative so a new runtime failure cannot silently skip reset. +static inline bool latched_error_may_poison_device(int32_t code) { + switch (code) { + case SIMPLER_ERROR_NONE: + return false; + case SIMPLER_ERROR_TENSOR_WAIT_TIMEOUT: + case SIMPLER_ERROR_SCHEDULER_TIMEOUT: + return true; + case SIMPLER_ERROR_SCOPE_DEADLOCK: + case SIMPLER_ERROR_HEAP_RING_DEADLOCK: + case SIMPLER_ERROR_FLOW_CONTROL_DEADLOCK: + case SIMPLER_ERROR_FANIN_CAPACITY_EXCEEDED: + case SIMPLER_ERROR_INVALID_ARGS: + case SIMPLER_ERROR_REQUIRE_SYNC_START_INVALID: + case SIMPLER_ERROR_EXPLICIT_ORCH_FATAL: + case SIMPLER_ERROR_SCOPE_TASKS_OVERFLOW: + case SIMPLER_ERROR_TENSORMAP_OVERFLOW: + case SIMPLER_ERROR_ASYNC_COMPLETION_INVALID: + case SIMPLER_ERROR_ASYNC_WAIT_OVERFLOW: + case SIMPLER_ERROR_ASYNC_REGISTRATION_FAILED: +#ifdef SIMPLER_ERROR_READY_QUEUE_OVERFLOW + case SIMPLER_ERROR_READY_QUEUE_OVERFLOW: +#endif + return false; + default: + return true; + } +} + #endif // SRC_COMMON_RUNTIME_STATUS_ERROR_NAMES_H_ diff --git a/src/common/worker/chip_run_lane.cpp b/src/common/worker/chip_run_lane.cpp index f3d0556701..575e258d60 100644 --- a/src/common/worker/chip_run_lane.cpp +++ b/src/common/worker/chip_run_lane.cpp @@ -79,6 +79,10 @@ struct ChipRunLaneState { return permits_native_successor(predecessor, successor.config); } + bool permits_queued_successor(const ChipRunState &predecessor, const ChipRunState &successor) const { + return worker->supports_queued_native_launch() && permits_native_successor(predecessor, successor); + } + void prepare(const std::shared_ptr &run) { run->native_run = worker->prepare_native_run_for_lane( run->callable_id, &run->args, run->config, run->lease, run->run_id, run->dispatch_id, run->accepted_state, @@ -96,13 +100,18 @@ struct ChipRunLaneState { void finish(const std::shared_ptr &run) noexcept { try { worker->finalize_native_run(run->native_run); + } catch (const NativeRunFailure &e) { + const std::exception_ptr finalize_error = std::current_exception(); + if (run->error == nullptr) run->error = finalize_error; + if (e.poisons_lane()) poison_with(run, finalize_error); } catch (...) { const std::exception_ptr finalize_error = std::current_exception(); if (run->error == nullptr) run->error = finalize_error; poison_with(run, finalize_error); } run->phase = ChipRunState::Phase::TERMINAL; - if (!fifo.empty() && fifo.front() == run) fifo.pop_front(); + auto it = std::find(fifo.begin(), fifo.end(), run); + if (it != fifo.end()) fifo.erase(it); } void fail_launch(const std::shared_ptr &run) noexcept { @@ -165,6 +174,21 @@ struct ChipRunLaneState { } } + void launch_successor_if_eligible() noexcept { + if (fifo.size() != 2) return; + const auto predecessor = fifo.front(); + const auto successor = fifo.back(); + if (!successor->activated || successor->phase != ChipRunState::Phase::PREPARED) return; + if (!permits_queued_successor(*predecessor, *successor)) return; + try { + worker->launch_native_run(successor->native_run); + successor->phase = ChipRunState::Phase::LAUNCHED; + successor->crossed_launch_fence = true; + } catch (...) { + fail_launch(successor); + } + } + bool progress(const std::shared_ptr &target) { if (target->phase == ChipRunState::Phase::TERMINAL) return true; if (fifo.empty()) throw std::runtime_error("chip run lane lost a nonterminal run"); @@ -176,6 +200,7 @@ struct ChipRunLaneState { launch_front(); if (fifo.size() == 2 && fifo.front() == target) prepare_successor_if_eligible(fifo.back()); + launch_successor_if_eligible(); if (target->phase == ChipRunState::Phase::TERMINAL) { launch_front(); return true; @@ -305,6 +330,7 @@ void ChipRun::activate() { if (lane_->fifo.size() == 2 && lane_->fifo.front() == run_) { lane_->prepare_successor_if_eligible(lane_->fifo.back()); } + lane_->launch_successor_if_eligible(); } void ChipRun::abandon() { @@ -411,6 +437,7 @@ ChipRun ChipRunLane::submit( } else { state_->prepare_successor_if_eligible(run); } + state_->launch_successor_if_eligible(); } catch (...) { run->error = std::current_exception(); run->phase = ChipRunState::Phase::TERMINAL; @@ -474,6 +501,7 @@ ChipRun ChipRunLane::submit( } else { state_->prepare_successor_if_eligible(run); } + state_->launch_successor_if_eligible(); return ChipRun(state_, std::move(run)); } diff --git a/src/common/worker/chip_worker.cpp b/src/common/worker/chip_worker.cpp index 20d66ba1bd..6c46647029 100644 --- a/src/common/worker/chip_worker.cpp +++ b/src/common/worker/chip_worker.cpp @@ -222,6 +222,9 @@ void ChipWorker::init( finalize_run_fn_ = load_symbol(handle, "simpler_finalize_run"); supports_concurrent_native_prepare_fn_ = load_symbol(handle, "supports_concurrent_native_prepare_ctx"); + supports_queued_native_launch_fn_ = + load_symbol(handle, "supports_queued_native_launch_ctx"); + native_run_error_poisons_fn_ = load_symbol(handle, "native_run_error_poisons_ctx"); get_arena_bank_gm_heap_base_fn_ = load_symbol(handle, "get_arena_bank_gm_heap_base_ctx"); get_retained_temp_addr_fn_ = load_symbol(handle, "get_retained_temp_addr_ctx"); @@ -348,6 +351,8 @@ void ChipWorker::init( wait_run_fn_ = nullptr; finalize_run_fn_ = nullptr; supports_concurrent_native_prepare_fn_ = nullptr; + supports_queued_native_launch_fn_ = nullptr; + native_run_error_poisons_fn_ = nullptr; get_arena_bank_gm_heap_base_fn_ = nullptr; get_retained_temp_addr_fn_ = nullptr; unregister_callable_fn_ = nullptr; @@ -400,6 +405,8 @@ void ChipWorker::init( wait_run_fn_ = nullptr; finalize_run_fn_ = nullptr; supports_concurrent_native_prepare_fn_ = nullptr; + supports_queued_native_launch_fn_ = nullptr; + native_run_error_poisons_fn_ = nullptr; get_arena_bank_gm_heap_base_fn_ = nullptr; get_retained_temp_addr_fn_ = nullptr; unregister_callable_fn_ = nullptr; @@ -524,6 +531,8 @@ void ChipWorker::finalize() { wait_run_fn_ = nullptr; finalize_run_fn_ = nullptr; supports_concurrent_native_prepare_fn_ = nullptr; + supports_queued_native_launch_fn_ = nullptr; + native_run_error_poisons_fn_ = nullptr; get_arena_bank_gm_heap_base_fn_ = nullptr; get_retained_temp_addr_fn_ = nullptr; unregister_callable_fn_ = nullptr; @@ -658,6 +667,10 @@ bool ChipWorker::supports_concurrent_native_prepare() const { supports_concurrent_native_prepare_fn_(device_ctx_) > 0; } +bool ChipWorker::supports_queued_native_launch() const { + return initialized_ && pipeline_contract_.pipeline_depth > 1 && supports_queued_native_launch_fn_(device_ctx_) > 0; +} + ChipWorkerNativeRun ChipWorker::prepare_native_run_on_slot( int32_t callable_id, const ChipStorageTaskArgs *args, const CallConfig &config, uint32_t slot_id, uint64_t generation, uint64_t run_id, uint64_t dispatch_id, volatile int32_t *accepted_state, @@ -900,8 +913,11 @@ void ChipWorker::finalize_native_run(const ChipWorkerNativeRun &run) { } int rc = finalize_rc != 0 ? finalize_rc : wait_rc; if (rc != 0) { - throw std::runtime_error( - "finalize_native_run failed with code " + std::to_string(rc) + " " + format_native_run_identity(run) + const bool poisons_lane = + native_run_error_poisons_fn_ == nullptr || native_run_error_poisons_fn_(device_ctx_, rc) != 0; + throw NativeRunFailure( + "finalize_native_run failed with code " + std::to_string(rc) + " " + format_native_run_identity(run), + poisons_lane ); } } diff --git a/src/common/worker/chip_worker.h b/src/common/worker/chip_worker.h index 0725a787ab..621ab84a54 100644 --- a/src/common/worker/chip_worker.h +++ b/src/common/worker/chip_worker.h @@ -52,6 +52,18 @@ class UnsupportedRuntimeOperation : public std::runtime_error { using std::runtime_error::runtime_error; }; +class NativeRunFailure : public std::runtime_error { +public: + NativeRunFailure(const std::string &message, bool poisons_lane) : + std::runtime_error(message), + poisons_lane_(poisons_lane) {} + + bool poisons_lane() const noexcept { return poisons_lane_; } + +private: + bool poisons_lane_{true}; +}; + class ChipWorker { public: ChipWorker() = default; @@ -121,9 +133,10 @@ class ChipWorker { * the token, and the caller must still finalize it. The blocking composition * performs that cleanup internally on every exit. * - * Onboard HBG may prepare one distinct-slot successor while another run - * owns the execution claim. Diagnostics and backends without the explicit - * capability remain depth-one. Lease generation gates admission; after a + * A capable onboard backend may prepare and submit one distinct-slot + * successor while another run remains below its completion fence. + * Diagnostics and backends without both capabilities remain depth-one. + * Lease generation gates admission; after a * successful prepare, the slot plus process-unique run epoch prevents a * delayed phase call from touching reused storage, including another run * under the same pipeline lease or on another ChipWorker. @@ -233,6 +246,7 @@ class ChipWorker { unsigned pipeline_depth() const { return pipeline_contract_.pipeline_depth; } size_t runtime_slot_count() const { return runtime_bufs_.size(); } bool supports_concurrent_native_prepare() const; + bool supports_queued_native_launch() const; /// Opaque host native-run storage address for every slot the contract /// asked for. Two slots hold distinct storage; tests read this to prove @@ -269,6 +283,7 @@ class ChipWorker { using SimplerPrepareRunFn = decltype(&simpler_prepare_run); using SimplerNativeRunFn = decltype(&simpler_launch_run); using SupportsConcurrentNativePrepareFn = int (*)(void *); + using NativeRunErrorPoisonsFn = decltype(&native_run_error_poisons_ctx); using GetArenaBankGmHeapBaseFn = uint64_t (*)(void *, uint32_t); using GetRetainedTempAddrFn = uint64_t (*)(void *, uint32_t); using GetPipelineContractFn = const PipelineContract *(*)(); @@ -331,6 +346,8 @@ class ChipWorker { SimplerNativeRunFn wait_run_fn_ = nullptr; SimplerNativeRunFn finalize_run_fn_ = nullptr; SupportsConcurrentNativePrepareFn supports_concurrent_native_prepare_fn_ = nullptr; + SupportsConcurrentNativePrepareFn supports_queued_native_launch_fn_ = nullptr; + NativeRunErrorPoisonsFn native_run_error_poisons_fn_ = nullptr; GetArenaBankGmHeapBaseFn get_arena_bank_gm_heap_base_fn_ = nullptr; GetRetainedTempAddrFn get_retained_temp_addr_fn_ = nullptr; SimplerUnregisterCallableFn unregister_callable_fn_ = nullptr; diff --git a/src/common/worker/runtime_c_api.h b/src/common/worker/runtime_c_api.h index efab062eeb..042211a8d9 100644 --- a/src/common/worker/runtime_c_api.h +++ b/src/common/worker/runtime_c_api.h @@ -33,6 +33,8 @@ * simpler_provision_dma_workspace * - pipeline: get_pipeline_contract, * supports_concurrent_native_prepare_ctx, + * supports_queued_native_launch_ctx, + * native_run_error_poisons_ctx, * get_arena_bank_gm_heap_base_ctx, * get_retained_temp_addr_ctx * - ACL/stream: ensure_acl_ready_ctx, create_comm_stream_ctx, @@ -407,6 +409,12 @@ int simpler_prepare_run( */ int supports_concurrent_native_prepare_ctx(DeviceContextHandle ctx); +/** Return nonzero when two prepared runs may be submitted to the same run streams. */ +int supports_queued_native_launch_ctx(DeviceContextHandle ctx); + +/** Return nonzero when an execution error invalidates the native run lane. */ +int native_run_error_poisons_ctx(DeviceContextHandle ctx, int execution_rc); + /** * Launch a prepared run. Returns only after the platform has published its * real kernel-launch marker, or after execution terminates before that marker. diff --git a/tests/st/a2a3/host_build_graph/concurrent_prepare_stress/test_concurrent_prepare_stress.py b/tests/st/a2a3/host_build_graph/concurrent_prepare_stress/test_concurrent_prepare_stress.py index 412f78320a..18d5c237a8 100644 --- a/tests/st/a2a3/host_build_graph/concurrent_prepare_stress/test_concurrent_prepare_stress.py +++ b/tests/st/a2a3/host_build_graph/concurrent_prepare_stress/test_concurrent_prepare_stress.py @@ -7,13 +7,16 @@ # 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. # ----------------------------------------------------------------------------------------------------------- -"""Concurrent-prepare overlap stress for host_build_graph. +"""Depth-two prepare and queued-launch stress for host_build_graph. Drives a 2-deep native-run pipeline over the two arena banks: run i is *prepared* (which runs its full bind — arena build + host orchestration) while run i-1 is still launched-but-not-finalized. That is exactly the ``overlaps_active_run`` path (a successor prepared into bank B while a predecessor executes in bank A), which the ordinary blocking run() never hits. +The same trace verdict also requires ``runner_run(i)`` to begin before +``runner_run(i-1)`` ends, proving the successor reached the shared streams +before the predecessor's completion fence. Each iteration uses distinct input data, so any cross-run interference between the two banks' runtimes/orchestrators (e.g. a shared or under-initialized @@ -25,7 +28,7 @@ same ``assert_native_overlap`` verdict, so each negative arm differs from the positive one in exactly one variable: -* ``inflight_limit=2``, ordinary config — overlap is required. +* ``inflight_limit=2``, ordinary config — prepare and queued-launch overlap are required. * ``inflight_limit=1`` — the lane's capability is untouched and the submissions simply never coexist, so the verdict must be *rejected*. This is what makes the positive arm a detector rather than a formality: the span chain between 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 8561aa061c..d385e40b85 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 @@ -135,6 +135,7 @@ def _run_and_validate_l2( # noqa: PLR0913, PLR0915 -- lifecycle contract is int chip_worker = worker._chip_worker assert chip_worker is not None supports_concurrent_prepare = bool(chip_worker._impl.supports_concurrent_native_prepare) + supports_queued_launch = bool(chip_worker._impl.supports_queued_native_launch) chip_worker._register_callable_at_slot(_SLOT, callable_obj) native_run = None successor_run = None @@ -150,9 +151,8 @@ def _run_and_validate_l2( # noqa: PLR0913, PLR0915 -- lifecycle contract is int ) first_run = native_run # 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. + # includes the run streams: the pair is readied by launch 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"): @@ -163,7 +163,7 @@ def _run_and_validate_l2( # noqa: PLR0913, PLR0915 -- lifecycle contract is int chip_worker._register_callable_at_slot(1, callable_obj) # A prepared run can be abandoned explicitly. Finalize releases its - # claim and registry dependencies without launching or copying back. + # reservation and registry dependencies without launching or copying back. chip_worker._finalize_native_run(native_run) native_run = None assert torch.count_nonzero(test_args.out) == 0 @@ -229,12 +229,10 @@ 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 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. + assert supports_queued_launch + # The successor owns a distinct bank while A remains in flight. + # Both launches reuse the same stream pair, and per-slot events + # let A retire without waiting for B's queued work. 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 @@ -254,18 +252,14 @@ def build_run_args(): assert bank0 != bank1 assert torch.count_nonzero(successor_args.out) == 0 - with pytest.raises(RuntimeError, match="launch_native_run failed") as claim_error: - chip_worker._launch_native_run(successor_run) - assert "slot=1" in str(claim_error.value) - assert "generation=1" in str(claim_error.value) - assert "run_epoch=" in str(claim_error.value) + chip_worker._launch_native_run(successor_run) + assert chip_worker.run_stream_set_create_count == stream_count chip_worker._wait_native_run(native_run) chip_worker._finalize_native_run(native_run) native_run = None _compare_outputs(active_args, active_golden, active_outputs, self.RTOL, self.ATOL) - chip_worker._launch_native_run(successor_run) chip_worker._wait_native_run(successor_run) chip_worker._finalize_native_run(successor_run) successor_run = None @@ -293,7 +287,7 @@ def build_run_args(): _compare_outputs(active_args, active_golden, active_outputs, self.RTOL, self.ATOL) # A diagnostic predecessor also cannot admit an ordinary - # successor while it owns the execution claim. + # successor while its runner-global diagnostics are active. active_args, active_chip_args, active_outputs, active_golden = build_run_args() native_run = chip_worker._prepare_native_run_with_pipeline_lease( _SLOT, active_chip_args, 0, _GENERATION + 3, config=diagnostic_config diff --git a/tests/st/runtime_fatal_codes/kernels/orchestration/successful_orch.cpp b/tests/st/runtime_fatal_codes/kernels/orchestration/successful_orch.cpp new file mode 100644 index 0000000000..deb07f0ff7 --- /dev/null +++ b/tests/st/runtime_fatal_codes/kernels/orchestration/successful_orch.cpp @@ -0,0 +1,27 @@ +/* + * 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 "orchestration_api.h" // NOLINT(build/include_subdir) + +extern "C" { + +__attribute__((visibility("default"))) OrchestrationConfig aicpu_orchestration_config(const ChipTaskArgs &orch_args) { + (void)orch_args; + return OrchestrationConfig{ + .expected_arg_count = 0, + }; +} + +__attribute__((visibility("default"))) void aicpu_orchestration_entry(const ChipTaskArgs &orch_args) { + (void)orch_args; +} + +} // extern "C" diff --git a/tests/st/runtime_fatal_codes/test_runtime_fatal_codes.py b/tests/st/runtime_fatal_codes/test_runtime_fatal_codes.py index 376bc21df1..aa6b3f1b13 100644 --- a/tests/st/runtime_fatal_codes/test_runtime_fatal_codes.py +++ b/tests/st/runtime_fatal_codes/test_runtime_fatal_codes.py @@ -278,6 +278,20 @@ def _build_chip_callable(platform: str, case: dict) -> ChipCallable: ) +def _build_successful_callable(platform: str) -> ChipCallable: + kc = KernelCompiler(platform=platform) + orch_bytes = kc.compile_orchestration( + runtime_name=RUNTIME, + source_path=os.path.join(ORCH_DIR, "successful_orch.cpp"), + ) + return ChipCallable.build( + signature=[], + func_name="aicpu_orchestration_entry", + binary=orch_bytes, + children=[], + ) + + def _make_worker(platform: str, device_id: int, case_name: str, monkeypatch): case = CASES[case_name] # Per-case timeout-chain overrides (which watchdog must fire, and when). @@ -338,3 +352,23 @@ def test_device_error_class_reaches_host_log(st_platform, st_device_ids, case_na _assert_annotated(log, case) finally: worker.close() + + +@pytest.mark.platforms(["a2a3"]) +@pytest.mark.device_count(1) +@pytest.mark.runtime(RUNTIME) +def test_invalid_args_failure_keeps_worker_usable(st_platform, st_device_ids): + failure_callable = _build_chip_callable(st_platform, CASES["invalid_args"]) + success_callable = _build_successful_callable(st_platform) + worker = Worker(level=2, platform=st_platform, runtime=RUNTIME, device_id=int(st_device_ids[0])) + failure_handle = worker.register(failure_callable) + success_handle = worker.register(success_callable) + worker.init() + try: + config = CallConfig() + config.aicpu_thread_num = 2 + with pytest.raises(RuntimeError, match=r"failed with code -5\b"): + worker.run(failure_handle, None, config) + worker.run(success_handle, None, config) + finally: + worker.close() diff --git a/tests/ut/cpp/common/test_error_code_names.cpp b/tests/ut/cpp/common/test_error_code_names.cpp index 89029b8d41..857dce8e74 100644 --- a/tests/ut/cpp/common/test_error_code_names.cpp +++ b/tests/ut/cpp/common/test_error_code_names.cpp @@ -88,6 +88,15 @@ TEST(RuntimeErrorNames, LatchedCodePicksTheNonZeroField) { EXPECT_EQ(latched_error_code(SIMPLER_ERROR_NONE, SIMPLER_ERROR_NONE), SIMPLER_ERROR_NONE); } +TEST(RuntimeErrorNames, OnlyTimeoutsAndUnknownCodesPoisonTheDevice) { + for (int32_t code : kAllRuntimeCodes) { + const bool expected = code == SIMPLER_ERROR_TENSOR_WAIT_TIMEOUT || code == SIMPLER_ERROR_SCHEDULER_TIMEOUT; + EXPECT_EQ(latched_error_may_poison_device(code), expected) << "code " << code; + } + EXPECT_FALSE(latched_error_may_poison_device(SIMPLER_ERROR_NONE)); + EXPECT_TRUE(latched_error_may_poison_device(9999)); +} + // The stall sub-class already had a name table; keep it and the error table consistent so a // SCHEDULER_TIMEOUT annotation and its sub_class= line cannot disagree. TEST(RuntimeErrorNames, StallDetailStillNamed) { diff --git a/tests/ut/cpp/hierarchical/test_chip_run_lane.cpp b/tests/ut/cpp/hierarchical/test_chip_run_lane.cpp index 4f82f08fbd..8fd73bb608 100644 --- a/tests/ut/cpp/hierarchical/test_chip_run_lane.cpp +++ b/tests/ut/cpp/hierarchical/test_chip_run_lane.cpp @@ -53,6 +53,8 @@ size_t g_poll_count{0}; // on the poll count instead of looping forever. size_t g_poll_completes_after{0}; bool g_supports_successor{true}; +bool g_supports_queued_launch{true}; +bool g_finalize_failure_poisons{true}; std::mutex g_wait_mu; std::condition_variable g_wait_cv; bool g_wait_entered{false}; @@ -111,6 +113,10 @@ int finalize_run(void *, void *runtime) { int supports_successor(void *) { return g_supports_successor ? 1 : 0; } +int supports_queued_launch(void *) { return g_supports_queued_launch ? 1 : 0; } + +int finalize_failure_poisons(void *, int error_code) { return error_code != 0 && g_finalize_failure_poisons ? 1 : 0; } + void prime_worker(ChipWorker &worker) { g_slots.clear(); g_complete = {}; @@ -124,6 +130,8 @@ void prime_worker(ChipWorker &worker) { g_poll_count = 0; g_poll_completes_after = 0; g_supports_successor = true; + g_supports_queued_launch = true; + g_finalize_failure_poisons = true; { std::lock_guard lk(g_wait_mu); g_wait_entered = false; @@ -143,6 +151,8 @@ void prime_worker(ChipWorker &worker) { worker.wait_run_fn_ = wait_run; worker.finalize_run_fn_ = finalize_run; worker.supports_concurrent_native_prepare_fn_ = supports_successor; + worker.supports_queued_native_launch_fn_ = supports_queued_launch; + worker.native_run_error_poisons_fn_ = finalize_failure_poisons; } ChipRun submit(ChipRunLane &lane, uint64_t run_id, uint32_t slot, bool activate = true) { @@ -163,12 +173,13 @@ TEST(ChipRunLaneTest, OwnsFifoPreparationAndLaunch) { EXPECT_EQ(g_events, (std::vector{"prepare0", "launch0", "prepare1"})); second.activate(); + EXPECT_TRUE(second.launched()); + EXPECT_EQ(g_events, (std::vector{"prepare0", "launch0", "prepare1", "launch1"})); EXPECT_FALSE(second.done()); g_complete[0] = true; EXPECT_TRUE(first.done()); EXPECT_FALSE(second.done()); - EXPECT_TRUE(second.launched()); - EXPECT_EQ(g_events, (std::vector{"prepare0", "launch0", "prepare1", "finalize0", "launch1"})); + EXPECT_EQ(g_events, (std::vector{"prepare0", "launch0", "prepare1", "launch1", "finalize0"})); g_complete[1] = true; EXPECT_TRUE(second.done()); @@ -308,9 +319,9 @@ TEST(ChipRunLaneTest, DirectCapacityTwoPreparesSuccessorAndBackpressuresThird) { ChipRun first = lane.submit(1, args, CallConfig{}); ChipRun second = lane.submit(1, args, CallConfig{}); EXPECT_TRUE(first.launched()); - EXPECT_FALSE(second.launched()); + EXPECT_TRUE(second.launched()); EXPECT_EQ(second.preparation_disposition(), ChipRunPreparationDisposition::NATIVE_PREPARED); - EXPECT_EQ(g_events, (std::vector{"prepare0", "launch0", "prepare1"})); + EXPECT_EQ(g_events, (std::vector{"prepare0", "launch0", "prepare1", "launch1"})); { std::lock_guard lk(g_wait_mu); @@ -346,7 +357,7 @@ TEST(ChipRunLaneTest, DirectCapacityTwoPreparesSuccessorAndBackpressuresThird) { ASSERT_EQ(submit_error, nullptr); ASSERT_TRUE(third.has_value()); EXPECT_TRUE(second.launched()); - EXPECT_FALSE(third->launched()); + EXPECT_TRUE(third->launched()); EXPECT_EQ(third->preparation_disposition(), ChipRunPreparationDisposition::NATIVE_PREPARED); EXPECT_EQ(g_prepare_count[0], 2u); @@ -435,6 +446,28 @@ TEST(ChipRunLaneTest, DirectRuntimeWithoutConcurrentPrepareRetainsDepthOne) { worker.finalize(); } +TEST(ChipRunLaneTest, ConcurrentPrepareWithoutQueuedLaunchRetainsDepthOneExecution) { + ChipWorker worker; + prime_worker(worker); + g_supports_queued_launch = false; + ChipRunLane lane(worker); + ChipStorageTaskArgs args{}; + + ChipRun first = lane.submit(1, args, CallConfig{}); + ChipRun second = lane.submit(1, args, CallConfig{}); + + EXPECT_TRUE(first.launched()); + EXPECT_FALSE(second.launched()); + EXPECT_EQ(second.preparation_disposition(), ChipRunPreparationDisposition::NATIVE_PREPARED); + g_complete[0] = true; + EXPECT_TRUE(first.done()); + EXPECT_TRUE(second.launched()); + g_complete[1] = true; + EXPECT_TRUE(second.done()); + lane.close(); + worker.finalize(); +} + TEST(ChipRunLaneTest, PrepareFailureIsTerminalWithoutPoisoningTheLane) { ChipWorker worker; prime_worker(worker); @@ -539,6 +572,35 @@ TEST(ChipRunLaneTest, FinalizeFailurePoisonsAdmissionAndCloseReportsIt) { worker.finalize(); } +TEST(ChipRunLaneTest, RecoverableRunFailureLeavesQueuedSuccessorAndLaneUsable) { + ChipWorker worker; + prime_worker(worker); + ChipRunLane lane(worker); + + ChipRun first = submit(lane, 101, 0); + ChipRun second = submit(lane, 102, 1); + ASSERT_TRUE(first.launched()); + ASSERT_TRUE(second.launched()); + + g_finalize_failure_poisons = false; + g_finalize_rc[0] = -5; + g_complete[0] = true; + EXPECT_TRUE(first.done()); + EXPECT_THROW(first.wait_until(ChipRunLane::Deadline::max()), NativeRunFailure); + EXPECT_FALSE(lane.poisoned()); + + g_complete[1] = true; + EXPECT_TRUE(second.wait_until(ChipRunLane::Deadline::max())); + EXPECT_FALSE(lane.poisoned()); + + g_finalize_rc[0] = 0; + ChipRun third = submit(lane, 103, 0); + g_complete[0] = true; + EXPECT_TRUE(third.wait_until(ChipRunLane::Deadline::max())); + lane.close(); + worker.finalize(); +} + TEST(ChipRunLaneTest, LaunchFailureWhoseFinalizeFailsReportsThePoisonedLane) { ChipWorker worker; prime_worker(worker); diff --git a/tests/ut/cpp/hierarchical/test_run_stream_pair.cpp b/tests/ut/cpp/hierarchical/test_run_stream_pair.cpp index b305678164..bf6daec87a 100644 --- a/tests/ut/cpp/hierarchical/test_run_stream_pair.cpp +++ b/tests/ut/cpp/hierarchical/test_run_stream_pair.cpp @@ -28,8 +28,10 @@ using CompletionStatus = RunStreamPair::CompletionStatus; // constants one address — the tests compare owners by address. const int kRunA = 1; const int kRunB = 2; +const int kRunC = 3; const void *const kOwnerA = &kRunA; const void *const kOwnerB = &kRunB; +const void *const kOwnerC = &kRunC; // 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. @@ -184,9 +186,12 @@ TEST(RunStreamPair, ANonSubmittingOwnerRetiresNothing) { // The predecessor's own poll and retirement still work. EXPECT_EQ( - pair.poll([](void *, void *) { - return SIMPLER_NATIVE_RUN_POLL_COMPLETE; - }), + pair.poll( + kOwnerA, + [](void *, void *) { + return SIMPLER_NATIVE_RUN_POLL_COMPLETE; + } + ), SIMPLER_NATIVE_RUN_POLL_COMPLETE ); ASSERT_EQ(pair.retire(CompletionStatus::Complete, kOwnerA), 0); @@ -215,7 +220,7 @@ TEST(RunStreamPair, CompleteRetirementWithoutAHandleIsBenign) { EXPECT_EQ(pair.created_count(), 0u); } -TEST(RunStreamPair, PollCompletionDoesNotPermitReuseBeforeRetirement) { +TEST(RunStreamPair, CompletedFrontDoesNotPermitStaleReplacementBeforeRetirement) { FakeStreams fake; RunStreamPair pair = make_pair(fake); @@ -223,16 +228,112 @@ TEST(RunStreamPair, PollCompletionDoesNotPermitReuseBeforeRetirement) { void *aicore = pair.aicore(); ASSERT_EQ(pair.mark_submitted(kOwnerA), 0); ASSERT_EQ( - pair.poll([](void *, void *) { - return SIMPLER_NATIVE_RUN_POLL_COMPLETE; - }), + pair.poll( + kOwnerA, + [](void *, void *) { + return SIMPLER_NATIVE_RUN_POLL_COMPLETE; + } + ), SIMPLER_NATIVE_RUN_POLL_COMPLETE ); + pair.mark_stale(); 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_NE(pair.aicore(), aicore); +} + +TEST(RunStreamPair, FrontCompletesWhileItsSuccessorRemainsQueued) { + 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.ensure(), 0); + ASSERT_EQ(pair.mark_submitted(kOwnerB), 0); + + EXPECT_EQ( + pair.poll( + kOwnerA, + [](void *, void *) { + return SIMPLER_NATIVE_RUN_POLL_COMPLETE; + } + ), + SIMPLER_NATIVE_RUN_POLL_COMPLETE + ); + EXPECT_EQ( + pair.poll( + kOwnerB, + [](void *, void *) { + return SIMPLER_NATIVE_RUN_POLL_NOT_READY; + } + ), + SIMPLER_NATIVE_RUN_POLL_NOT_READY + ); + ASSERT_EQ(pair.retire(CompletionStatus::Complete, kOwnerA), 0); EXPECT_EQ(pair.aicore(), aicore); + EXPECT_EQ( + pair.poll( + kOwnerB, + [](void *, void *) { + return SIMPLER_NATIVE_RUN_POLL_COMPLETE; + } + ), + SIMPLER_NATIVE_RUN_POLL_COMPLETE + ); + ASSERT_EQ(pair.retire(CompletionStatus::Complete, kOwnerB), 0); +} + +TEST(RunStreamPair, CompleteRetirementMustFollowSubmissionOrder) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + + ASSERT_EQ(pair.ensure(), 0); + ASSERT_EQ(pair.mark_submitted(kOwnerA), 0); + ASSERT_EQ(pair.mark_submitted(kOwnerB), 0); + EXPECT_NE(pair.retire(CompletionStatus::Complete, kOwnerB), 0); + EXPECT_EQ( + pair.poll( + kOwnerB, + [](void *, void *) { + return SIMPLER_NATIVE_RUN_POLL_COMPLETE; + } + ), + SIMPLER_NATIVE_RUN_POLL_ERROR + ); + EXPECT_NE(pair.ensure(), 0) << "out-of-order retirement makes the shared stream unproven"; + EXPECT_EQ(pair.retire(CompletionStatus::Complete, kOwnerA), 0); + EXPECT_EQ(pair.aicore(), nullptr); +} + +TEST(RunStreamPair, ThirdSubmissionIsRejectedAtPipelineDepthTwo) { + FakeStreams fake; + RunStreamPair pair = make_pair(fake); + + ASSERT_EQ(pair.ensure(), 0); + ASSERT_EQ(pair.mark_submitted(kOwnerA), 0); + ASSERT_EQ(pair.mark_submitted(kOwnerB), 0); + EXPECT_NE(pair.mark_submitted(kOwnerC), 0); +} + +TEST(RunStreamPair, UnprovenFrontDefersDestroyUntilQueuedSuccessorRetires) { + 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.mark_submitted(kOwnerB), 0); + + EXPECT_EQ(pair.retire(CompletionStatus::Unproven, kOwnerA), 0); + EXPECT_EQ(pair.aicore(), aicore); + EXPECT_NE(pair.ensure(), 0) << "an unproven shared stream cannot accept more work"; + + EXPECT_EQ(pair.retire(CompletionStatus::Unproven, kOwnerB), 0); + EXPECT_EQ(pair.aicore(), nullptr); + EXPECT_EQ(fake.live_count(), 1u) << "the AICPU stream remains runner-owned"; } TEST(RunStreamPair, UnprovenRunDestroysTheAicoreStreamAndKeepsTheAicpuOne) { @@ -327,9 +428,12 @@ TEST(RunStreamPair, UnprovenRetirementClearsAnEarlyCompletion) { ASSERT_EQ(pair.mark_submitted(kOwnerA), 0); EXPECT_EQ( - pair.poll([](void *, void *) { - return SIMPLER_NATIVE_RUN_POLL_COMPLETE; - }), + pair.poll( + kOwnerA, + [](void *, void *) { + return SIMPLER_NATIVE_RUN_POLL_COMPLETE; + } + ), SIMPLER_NATIVE_RUN_POLL_COMPLETE ); @@ -337,10 +441,13 @@ TEST(RunStreamPair, UnprovenRetirementClearsAnEarlyCompletion) { // 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; - }), + pair.poll( + kOwnerA, + [](void *, void *) { + ADD_FAILURE() << "an unproven retired stream must not be queried"; + return SIMPLER_NATIVE_RUN_POLL_COMPLETE; + } + ), SIMPLER_NATIVE_RUN_POLL_ERROR ); } @@ -351,16 +458,22 @@ TEST(RunStreamPair, PollRequiresSubmissionAndPropagatesQueryErrors) { ASSERT_EQ(pair.ensure(), 0); EXPECT_EQ( - pair.poll([](void *, void *) { - return SIMPLER_NATIVE_RUN_POLL_COMPLETE; - }), + pair.poll( + kOwnerA, + [](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; - }), + pair.poll( + kOwnerA, + [](void *, void *) { + return SIMPLER_NATIVE_RUN_POLL_ERROR; + } + ), SIMPLER_NATIVE_RUN_POLL_ERROR ); } @@ -385,7 +498,7 @@ TEST(RunStreamPair, RetireWaitsForAnInFlightPoll) { 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 *) { + return pair.poll(kOwnerA, [&](void *, void *) { query_entered.set_value(); release.wait(); return SIMPLER_NATIVE_RUN_POLL_NOT_READY; @@ -406,9 +519,12 @@ TEST(RunStreamPair, RetireWaitsForAnInFlightPoll) { EXPECT_EQ(retire.get(), 0); EXPECT_EQ(fake.live_count(), 1u); EXPECT_EQ( - pair.poll([](void *, void *) { - return SIMPLER_NATIVE_RUN_POLL_ERROR; - }), + pair.poll( + kOwnerA, + [](void *, void *) { + return SIMPLER_NATIVE_RUN_POLL_ERROR; + } + ), SIMPLER_NATIVE_RUN_POLL_ERROR ); } @@ -439,10 +555,13 @@ TEST(RunStreamPair, PollDoesNotWaitBehindRetirement) { // 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; - }), + pair.poll( + kOwnerA, + [](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 ); @@ -477,10 +596,13 @@ TEST(RunStreamPair, AbandonClearsHandlesWithoutDestroyingThem) { 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; - }), + pair.poll( + kOwnerA, + [](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); diff --git a/tests/ut/py/test_strace_timing.py b/tests/ut/py/test_strace_timing.py index 74b7e48be8..1abf1470e2 100644 --- a/tests/ut/py/test_strace_timing.py +++ b/tests/ut/py/test_strace_timing.py @@ -775,7 +775,7 @@ def test_native_overlap_reads_identity_from_the_invocation_root(): `(pid, inv)` is what joins them — the same grouping the TPOT views use. """ lines = _run_records(run_epoch=1, prepare=(50, 20), device=(100, 100), release=200) - lines += _run_records(run_epoch=2, prepare=(150, 30), device=(240, 100), release=340) + lines += _run_records(run_epoch=2, prepare=(150, 30), device=(180, 100), release=280) checks = assert_native_overlap(parse_spans(lines)) @@ -792,7 +792,7 @@ def test_native_overlap_accepts_a_prepare_that_outlasts_the_device_window(): blames the wrong thing. """ lines = _run_records(run_epoch=1, prepare=(50, 20), device=(100, 100), release=200) - lines += _run_records(run_epoch=2, prepare=(190, 40), device=(240, 100), release=340) + lines += _run_records(run_epoch=2, prepare=(190, 40), device=(195, 100), release=295) assert len(assert_native_overlap(parse_spans(lines))) == 1 @@ -802,17 +802,17 @@ def test_native_overlap_accepts_a_prepare_that_outlasts_the_device_window(): def test_native_overlap_rejects_a_prepare_after_the_device_window(): lines = _run_records(run_epoch=1, prepare=(50, 20), device=(100, 100), release=200) - lines += _run_records(run_epoch=2, prepare=(210, 20), device=(240, 100), release=340) + lines += _run_records(run_epoch=2, prepare=(210, 20), device=(220, 100), release=320) with pytest.raises(NativeOverlapError, match="did not overlap"): assert_native_overlap(parse_spans(lines)) -def test_native_overlap_rejects_device_work_before_the_predecessor_release(): - lines = _run_records(run_epoch=1, prepare=(50, 20), device=(100, 100), release=250) +def test_native_overlap_rejects_a_successor_launched_after_predecessor_completion(): + lines = _run_records(run_epoch=1, prepare=(50, 20), device=(100, 100), release=210) lines += _run_records(run_epoch=2, prepare=(150, 30), device=(240, 100), release=400) - with pytest.raises(NativeOverlapError, match="reordered before the claim release"): + with pytest.raises(NativeOverlapError, match="not queued before predecessor completion"): assert_native_overlap(parse_spans(lines)) @@ -822,7 +822,7 @@ def test_native_overlap_orders_by_run_epoch_when_dispatch_id_is_absent(): `run_epoch` is the only identity those runs carry, so it is what orders them. """ lines = _run_records(run_epoch=1, prepare=(50, 20), device=(100, 100), release=200) - lines += _run_records(run_epoch=2, prepare=(150, 30), device=(240, 100), release=340) + lines += _run_records(run_epoch=2, prepare=(150, 30), device=(180, 100), release=280) identities = [check.predecessor for check in assert_native_overlap(parse_spans(lines))] @@ -833,7 +833,7 @@ def test_native_overlap_orders_by_run_epoch_when_dispatch_id_is_absent(): def test_native_overlap_orders_by_dispatch_id_when_the_scheduler_allocates_one(): lines = _run_records(run_epoch=1, dispatch_id=10, run_id=5, prepare=(50, 20), device=(100, 100), release=200) - lines += _run_records(run_epoch=2, dispatch_id=11, run_id=6, prepare=(150, 30), device=(240, 100), release=340) + lines += _run_records(run_epoch=2, dispatch_id=11, run_id=6, prepare=(150, 30), device=(180, 100), release=280) checks = assert_native_overlap(parse_spans(lines)) @@ -843,7 +843,7 @@ def test_native_overlap_orders_by_dispatch_id_when_the_scheduler_allocates_one() def test_native_overlap_skips_an_invocation_that_is_not_a_phased_run(): """A lexical `chip.run` carries no identity attrs, so it orders nothing.""" lines = _run_records(run_epoch=1, prepare=(50, 20), device=(100, 100), release=200) - lines += _run_records(run_epoch=2, prepare=(150, 30), device=(240, 100), release=340) + lines += _run_records(run_epoch=2, prepare=(150, 30), device=(180, 100), release=280) lines += [_record(7, 99, "chip.run", "", depth=0, ts=10, dur=5)] assert len(assert_native_overlap(parse_spans(lines))) == 1 diff --git a/tests/ut/py/test_worker/test_host_worker.py b/tests/ut/py/test_worker/test_host_worker.py index ba03153522..0c48c8e512 100644 --- a/tests/ut/py/test_worker/test_host_worker.py +++ b/tests/ut/py/test_worker/test_host_worker.py @@ -702,6 +702,7 @@ def activate(self) -> None: self.activated = True self._lane._launch_front() self._lane._prepare_successor() + self._lane._launch_successor() def abandon(self) -> None: if self._launched: @@ -720,6 +721,7 @@ def _raise_if_failed(self) -> None: class _FakeNativeRunImpl: def __init__(self, *, supports_concurrent_native_prepare: bool = False) -> None: self.supports_concurrent_native_prepare = supports_concurrent_native_prepare + self.supports_queued_native_launch = supports_concurrent_native_prepare self.events: list[tuple] = [] self.completed = [threading.Event(), threading.Event()] self.prepared = [threading.Event(), threading.Event()] @@ -731,6 +733,7 @@ def __init__(self, *, supports_concurrent_native_prepare: bool = False) -> None: self.prepare_errors: dict[tuple[int, int], BaseException] = {} self.poll_errors: dict[tuple[int, int], BaseException] = {} self.finalize_errors: dict[tuple[int, int], BaseException] = {} + self.recoverable_finalize_errors: set[tuple[int, int]] = set() self.prepare_identities: list[tuple[int, int, int, int]] = [] self.poll_states: list[tuple[int, int]] = [] self.register_calls: list[tuple[int, int]] = [] @@ -853,7 +856,9 @@ def _finish(self, run: _FakeChipRun) -> None: self._finalize_native_run(run.token) except BaseException as error: # noqa: BLE001 run.error = run.error or error - self._lane_poisoned = True + run_key = (int(run.token.slot_id), int(run.token.generation)) + if run_key not in self.recoverable_finalize_errors: + self._lane_poisoned = True run.terminal = True if run in self._runs: self._runs.remove(run) @@ -901,6 +906,26 @@ def _prepare_successor(self) -> None: successor.terminal = True self._runs.remove(successor) + def _launch_successor(self) -> None: + if len(self._runs) != 2: + return + predecessor, successor = self._runs + if ( + predecessor._launched + and successor.activated + and successor.token is not None + and not successor._launched + and self.supports_queued_native_launch + and not self._has_diagnostics(predecessor.submission.config) + and not self._has_diagnostics(successor.submission.config) + ): + try: + self._launch_native_run(successor.token) + successor._launched = True + except BaseException as error: # noqa: BLE001 + successor.error = error + self._finish(successor) + def _progress(self, target: _FakeChipRun) -> bool: if target.terminal: return True @@ -913,6 +938,7 @@ def _progress(self, target: _FakeChipRun) -> bool: return False self._launch_front() self._prepare_successor() + self._launch_successor() if target.terminal: self._launch_front() return True @@ -959,6 +985,7 @@ def _submit_chip_run_materialized( # noqa: PLR0913 -- mirrors the production bi self._runs.sort(key=lambda candidate: candidate.submission.dispatch_id) self._launch_front() self._prepare_successor() + self._launch_successor() return run def _close_chip_run_lane(self) -> None: @@ -1224,14 +1251,12 @@ def test_two_frame_hbg_prepares_b_while_a_runs_but_accepts_only_after_launch(): assert _mailbox_load_i32(harness.accepted_addr(1)) == 0 _mailbox_store_i32(harness.state_addr(1), worker_mod._ACTIVATE) - assert harness.cw._impl.wait_for_non_head_progress(1) - assert not harness.cw._impl.launched[1].is_set() - assert _mailbox_load_i32(harness.accepted_addr(1)) == 0 + assert harness.cw._impl.launched[1].wait(5.0) + assert _mailbox_load_i32(harness.accepted_addr(1)) == worker_mod._TASK_ACCEPTED + assert not harness.cw._impl.finalized[0].is_set() harness.cw._impl.completed[0].set() assert harness.cw._impl.finalized[0].wait(5.0) - assert harness.cw._impl.launched[1].wait(5.0) - assert _mailbox_load_i32(harness.accepted_addr(1)) == worker_mod._TASK_ACCEPTED harness.cw._impl.completed[1].set() harness.wait_state(1, worker_mod._TASK_DONE) @@ -1240,8 +1265,8 @@ def test_two_frame_hbg_prepares_b_while_a_runs_but_accepts_only_after_launch(): ("prepare", 0), ("launch", 0), ("prepare", 1), - ("finalize", 0), ("launch", 1), + ("finalize", 0), ("finalize", 1), ] finally: @@ -1305,16 +1330,16 @@ def test_two_frame_hbg_prepares_and_launches_reverse_ready_frames_by_dispatch_id harness.start() assert harness.cw._impl.launched[1].wait(5.0) - assert not harness.cw._impl.launched[0].is_set() + assert harness.cw._impl.launched[0].wait(5.0) assert [event[:2] for event in harness.cw._impl.events if event[0] in {"prepare", "launch"}] == [ ("prepare", 1), ("launch", 1), ("prepare", 0), + ("launch", 0), ] assert harness.cw._impl.prepare_identities == [(1, 11, 5, 1), (0, 11, 5, 2)] harness.cw._impl.completed[1].set() - assert harness.cw._impl.launched[0].wait(5.0) harness.cw._impl.completed[0].set() harness.wait_state(0, worker_mod._TASK_DONE) finally: @@ -1823,6 +1848,41 @@ def test_two_frame_native_progress_failure_terminalizes_staged_successor(failure harness.close() +def test_two_frame_recoverable_runtime_failure_keeps_successor_and_lane_usable(): + harness = _TwoFrameLoopHarness( + supports_concurrent_native_prepare=True, + chip_runtime="host_build_graph", + ) + try: + harness.publish(0, 1) + harness.start() + assert harness.cw._impl.launched[0].wait(5.0) + + harness.publish(1, 2, state=worker_mod._PREPARE_READY) + harness.wait_state(1, worker_mod._FRAME_STAGED) + _mailbox_store_i32(harness.state_addr(1), worker_mod._ACTIVATE) + assert harness.cw._impl.launched[1].wait(5.0) + + harness.cw._impl.finalize_errors[(0, 11)] = RuntimeError("invalid orchestration args") + harness.cw._impl.recoverable_finalize_errors.add((0, 11)) + harness.cw._impl.completed[0].set() + harness.wait_state(0, worker_mod._TASK_FAILED) + assert harness.thread.is_alive() + + harness.cw._impl.completed[1].set() + harness.wait_state(1, worker_mod._TASK_DONE) + + harness.cw._impl.completed[0].clear() + harness.cw._impl.launched[0].clear() + harness.publish(0, 3, generation=12) + harness.cw._impl.finalize_errors.pop((0, 11)) + assert harness.cw._impl.launched[0].wait(5.0) + harness.cw._impl.completed[0].set() + harness.wait_state(0, worker_mod._TASK_DONE) + finally: + harness.close() + + def _chip_digest(callable_obj: ChipCallable, *, platform: str = "", runtime: str = "") -> bytes: descriptor = build_chip_callable_descriptor(target=callable_obj, platform=platform, runtime=runtime) return hashid_to_digest(compute_callable_hashid(descriptor))