From 979f6bbd893cbeeef9364e0f678f9490f500a86f Mon Sep 17 00:00:00 2001 From: HighCloud Date: Mon, 24 Aug 2026 00:48:25 -0700 Subject: [PATCH] Fix: stabilize depth-two chip run lifecycle Keep preparation, native execution, completion, and stream retirement owned by resident workers so host scheduling gaps do not enter every decode step. - Prestart the two-frame preparation and completion pools - Reuse one native executor pool per device context - Serialize stream retirement and replenishment across pipeline slots - Buffer host timing spans per process to avoid shared stderr contention - Cover concurrent preparation, completion, shutdown, and slot reuse --- python/simpler/worker.py | 249 +++++++++--- .../platform/onboard/host/device_runner.cpp | 86 +++- .../platform/onboard/host/device_runner.h | 2 + .../host/runtime_maker.cpp | 16 +- src/common/log/include/common/strace.h | 79 +++- .../platform/include/host/run_stream_slots.h | 260 ++++++++++-- .../platform/onboard/host/c_api_shared.cpp | 376 ++++++++++++++---- .../onboard/host/device_runner_base.cpp | 12 +- .../onboard/host/device_runner_base.h | 3 + src/common/worker/native_run_launch_signal.h | 8 + src/common/worker/native_run_state.h | 10 + .../hierarchical/test_run_stream_slots.cpp | 88 ++++ tests/ut/py/test_worker/test_host_worker.py | 61 ++- 13 files changed, 1065 insertions(+), 185 deletions(-) diff --git a/python/simpler/worker.py b/python/simpler/worker.py index 3136a1af46..31d98ab608 100644 --- a/python/simpler/worker.py +++ b/python/simpler/worker.py @@ -75,6 +75,7 @@ def my_l4_orch(orch, args, config): import threading import time import uuid +from concurrent.futures import Future, ThreadPoolExecutor from dataclasses import dataclass, field from multiprocessing import resource_tracker from multiprocessing.shared_memory import SharedMemory @@ -2444,11 +2445,39 @@ class _StagedFrame: activated: bool native_run: Any = None published: bool = False + prepare_future: Future[None] | None = None + prepare_done: threading.Event = field(default_factory=threading.Event) + prepare_error: BaseException | None = None + completion_future: Future[None] | None = None + completion_start: threading.Event = field(default_factory=threading.Event) + completion_cancel: threading.Event = field(default_factory=threading.Event) + completion_done: threading.Event = field(default_factory=threading.Event) + completion_error: BaseException | None = None supports_concurrent_native_prepare = bool(cw._impl.supports_concurrent_native_prepare) staged_frames: dict[int, _StagedFrame] = {} active_frame: _StagedFrame | None = None active_run: Any = None + prepare_pool = ThreadPoolExecutor( + max_workers=_TASK_FRAME_COUNT, + thread_name_prefix=f"simpler-prepare-dev{device_id}", + ) + completion_pool = ThreadPoolExecutor( + max_workers=_TASK_FRAME_COUNT, + thread_name_prefix=f"simpler-complete-dev{device_id}", + ) + + def prestart_pool(pool: ThreadPoolExecutor) -> None: + # ThreadPoolExecutor starts workers lazily. Hold one task on each + # worker so all pthread creation is paid before the first frame. + barrier = threading.Barrier(_TASK_FRAME_COUNT + 1) + futures = [pool.submit(barrier.wait) for _ in range(_TASK_FRAME_COUNT)] + barrier.wait() + for future in futures: + future.result() + + prestart_pool(prepare_pool) + prestart_pool(completion_pool) def config_has_diagnostics(config: CallConfig) -> bool: # Mirrors CallConfig::diagnostics_any(); these modes share native @@ -2462,7 +2491,10 @@ def config_has_diagnostics(config: CallConfig) -> bool: ) def has_backend_prepared_frame() -> bool: - return any(frame.native_run is not None for frame in staged_frames.values()) + return any( + frame.native_run is not None or (frame.prepare_future is not None and not frame.prepare_done.is_set()) + for frame in staged_frames.values() + ) def read_identity(frame_buf: memoryview) -> tuple[int, int, int, int, int]: return ( @@ -2507,7 +2539,91 @@ def prepare_frame_native_run(frame: _StagedFrame) -> Any: ) return frame.native_run + def start_frame_native_prepare(frame: _StagedFrame) -> None: + if frame.native_run is not None or frame.prepare_future is not None: + return + + def prepare() -> None: + try: + native_run = prepare_frame_native_run(frame) + start_frame_completion_waiter(frame, native_run) + except BaseException as error: # noqa: BLE001 - progress owner publishes this failure + frame.prepare_error = error + finally: + frame.prepare_done.set() + + frame.prepare_done.clear() + frame.prepare_future = prepare_pool.submit(prepare) + + def start_frame_completion_waiter(frame: _StagedFrame, native_run: Any) -> None: + if frame.completion_future is not None: + return + + def complete() -> None: + frame.completion_start.wait() + if frame.completion_cancel.is_set(): + frame.completion_done.set() + return + try: + cw._impl._wait_native_run(native_run) + except BaseException as error: # noqa: BLE001 - progress owner publishes this failure + frame.completion_error = error + # A failed wait may still leave a live native token. Make + # one cleanup attempt here because ownership transferred + # to this waiter when completion_start was signalled. + try: + cw._impl._finalize_native_run(native_run) + except BaseException as finalize_error: # noqa: BLE001 + if finalize_error is not error: + frame.completion_error = RuntimeError(f"{error}; native finalize: {finalize_error}") + else: + try: + cw._impl._finalize_native_run(native_run) + except BaseException as error: # noqa: BLE001 - progress owner publishes this failure + frame.completion_error = error + finally: + frame.completion_done.set() + + frame.completion_future = completion_pool.submit(complete) + + def finish_frame_native_prepare(frame: _StagedFrame) -> None: + future = frame.prepare_future + if future is None: + return + future.result() + if frame.prepare_error is not None: + error = frame.prepare_error + frame.prepare_error = None + raise error + + def cancel_frame_completion_waiter(frame: _StagedFrame) -> None: + future = frame.completion_future + if future is None or frame.completion_start.is_set(): + return + frame.completion_cancel.set() + frame.completion_start.set() + future.result() + + def finish_frame_native_completion(frame: _StagedFrame, *, block: bool) -> bool: + future = frame.completion_future + if future is None: + return False + if not block and not frame.completion_done.is_set(): + return False + future.result() + if frame.completion_error is not None: + error = frame.completion_error + frame.completion_error = None + raise error + return True + def finalize_frame_native_run(frame: _StagedFrame) -> None: + finish_frame_native_prepare(frame) + if frame.completion_future is not None: + if frame.completion_start.is_set() and not frame.completion_cancel.is_set(): + finish_frame_native_completion(frame, block=True) + return + cancel_frame_completion_waiter(frame) native_run = frame.native_run if native_run is None: return @@ -2635,57 +2751,24 @@ def stage_frame(index: int, initial_state: int) -> _StagedFrame | None: if stop_after_frame_scan: break - next_active = None - if active_frame is None: - activated_frames = [frame for frame in staged_frames.values() if frame.activated] - if activated_frames: - next_active = min(activated_frames, key=lambda frame: frame.identity[4]) - - for staged in sorted(staged_frames.values(), key=lambda frame: frame.identity[4]): - # A frame published before any active claim is validation-only. - # Keep considering it so activation or a later predecessor - # claim can add the missing native token. - native_prepare_now = ( - staged.native_run is None - and supports_concurrent_native_prepare - and not config_has_diagnostics(staged.config) - and ( - (active_frame is None and staged is next_active) - or ( - active_frame is not None - and staged is not active_frame - and not config_has_diagnostics(active_frame.config) - ) - ) - ) - if staged.published and not native_prepare_now: - continue - try: - if native_prepare_now: - prepare_frame_native_run(staged) - if not staged.published: - publish_frame_staged(staged) - except Exception as e: # noqa: BLE001 - prepare_message = _format_exc(f"chip_process dev={device_id}: native prepare", e) - finalize_failed = False - try: - finalize_frame_native_run(staged) - except Exception as finalize_error: # noqa: BLE001 - finalize_failed = True - prepare_message += "; " + _format_exc("native finalize", finalize_error) - fail_frame(staged, prepare_message) - staged_frames.pop(staged.index, None) - if finalize_failed: - shutdown_message = prepare_message - stop_after_frame_scan = True - break - - if stop_after_frame_scan: - break - + # Retire a completed active run before doing potentially slow + # native preparation for its successor. Preparing the staged + # frame first can spend milliseconds in stream/resource setup + # after the active device work has already finished, turning + # successor setup jitter into predecessor finalize/validate + # tail latency. When the active run is still executing we + # continue below and overlap that same preparation with it. if active_frame is not None: try: - run_complete = bool(cw._impl._poll_native_run(active_run)) + if active_frame.completion_future is not None: + # The dedicated waiter is notified by the native + # executor and performs validate/retirement itself. + # Keep this bounded wait so mailbox/control progress + # remains responsive while avoiding a busy spin. + active_frame.completion_done.wait(0.0001) + run_complete = finish_frame_native_completion(active_frame, block=False) + else: + run_complete = bool(cw._impl._poll_native_run(active_run)) except Exception as e: # noqa: BLE001 poll_message = _format_exc(f"chip_process dev={device_id}: native poll", e) try: @@ -2735,9 +2818,66 @@ def stage_frame(index: int, initial_state: int) -> _StagedFrame | None: shutdown_message = msg break + next_active = None + if active_frame is None: + activated_frames = [frame for frame in staged_frames.values() if frame.activated] + if activated_frames: + next_active = min(activated_frames, key=lambda frame: frame.identity[4]) + + for staged in sorted(staged_frames.values(), key=lambda frame: frame.identity[4]): + # A frame published before any active claim is validation-only. + # Keep considering it so activation or a later predecessor + # claim can add the missing native token. + native_prepare_now = ( + staged.native_run is None + and supports_concurrent_native_prepare + and not config_has_diagnostics(staged.config) + and ( + (active_frame is None and staged is next_active) + or ( + active_frame is not None + and staged is not active_frame + and not config_has_diagnostics(active_frame.config) + ) + ) + ) + if staged.published and not native_prepare_now: + continue + try: + if native_prepare_now: + start_frame_native_prepare(staged) + if not staged.prepare_done.is_set(): + continue + finish_frame_native_prepare(staged) + if not staged.published: + publish_frame_staged(staged) + except Exception as e: # noqa: BLE001 + prepare_message = _format_exc(f"chip_process dev={device_id}: native prepare", e) + finalize_failed = False + try: + finalize_frame_native_run(staged) + except Exception as finalize_error: # noqa: BLE001 + finalize_failed = True + prepare_message += "; " + _format_exc("native finalize", finalize_error) + fail_frame(staged, prepare_message) + staged_frames.pop(staged.index, None) + if finalize_failed: + shutdown_message = prepare_message + stop_after_frame_scan = True + break + + if stop_after_frame_scan: + break + if active_frame is None: eligible = sorted( - (frame for frame in staged_frames.values() if frame.activated and frame.published), + ( + frame + for frame in staged_frames.values() + if frame is next_active + and frame.published + and (frame.prepare_future is None or frame.prepare_done.is_set()) + ), key=lambda frame: frame.identity[4], ) if eligible: @@ -2787,6 +2927,11 @@ def stage_frame(index: int, initial_state: int) -> _StagedFrame | None: shutdown_message = launch_message break else: + if next_frame.completion_future is not None: + # Transfer sole ownership of the launched + # token to the already-parked waiter. + next_frame.native_run = None + next_frame.completion_start.set() active_frame = next_frame active_run = native_run _mailbox_store_i32(next_frame.frame_addr + _OFF_STATE, _TASK_LAUNCHED) @@ -2826,6 +2971,8 @@ def stage_frame(index: int, initial_state: int) -> _StagedFrame | None: _mailbox_store_i32(frame_state_addr, _TASK_FAILED) for frame_buf in frame_bufs: frame_buf.release() + prepare_pool.shutdown(wait=True) + completion_pool.shutdown(wait=True) try: if task_frame_count >= 2: diff --git a/src/a2a3/platform/onboard/host/device_runner.cpp b/src/a2a3/platform/onboard/host/device_runner.cpp index c326a521ba..5a3bfa85cf 100644 --- a/src/a2a3/platform/onboard/host/device_runner.cpp +++ b/src/a2a3/platform/onboard/host/device_runner.cpp @@ -40,6 +40,7 @@ #include "callable_protocol.h" #include "call_config.h" #include "chip_callable_layout.h" +#include "common/strace.h" #include "utils/elf_build_id.h" #include "host/host_regs.h" // Register address retrieval #include "host/raii_scope_guard.h" @@ -241,13 +242,18 @@ int DeviceRunner::abandon_native_run_resources(uint32_t pipeline_slot) { return retire_run_aicore_stream(pipeline_slot); } +int DeviceRunner::complete_native_run_resources(uint32_t pipeline_slot) { + return retire_run_aicore_stream_async(pipeline_slot); +} + int DeviceRunner::run(Runtime &runtime, const CallConfig &config) { const unsigned selected_pipeline_slot = pipeline_slot(); // The AICore stream is created during native prepare so its provisioning - // can overlap the predecessor's execution. Once run() is entered, every - // exit owns retirement; the success path reports destroy failure, while - // early-error paths keep the original error and leave a failed-destroy - // handle in the slot so it cannot be reused. + // can overlap the predecessor's execution. Early-error paths retire it + // synchronously and leave a failed-destroy handle in the slot. Success + // hands ownership back to native finalize: only after output D2H completes + // may it start asynchronous retire/replenish. This prevents CANN stream + // control calls from contending with validate's rtMemcpy. bool aicore_stream_retired = false; auto aicore_stream_retire = RAIIScopeGuard([this, selected_pipeline_slot, &aicore_stream_retired]() { if (!aicore_stream_retired) (void)retire_run_aicore_stream(selected_pipeline_slot); @@ -300,6 +306,7 @@ int DeviceRunner::run(Runtime &runtime, const CallConfig &config) { // before the allocs so that an alloc-failure early-return still triggers // cleanup of previously-allocated buffers (the predicates no-op on 0). auto regs_cleanup = RAIIScopeGuard([this]() { + STRACE("simpler_run.runner_run.cleanup.regs"); if (kernel_args_.args.regs != 0) { mem_alloc_.free(reinterpret_cast(kernel_args_.args.regs)); kernel_args_.args.regs = 0; @@ -307,6 +314,7 @@ int DeviceRunner::run(Runtime &runtime, const CallConfig &config) { }); auto pmu_regs_cleanup = RAIIScopeGuard([this]() { + STRACE("simpler_run.runner_run.cleanup.pmu_regs"); if (kernel_args_.args.pmu_reg_addrs != 0) { mem_alloc_.free(reinterpret_cast(kernel_args_.args.pmu_reg_addrs)); kernel_args_.args.pmu_reg_addrs = 0; @@ -401,6 +409,7 @@ int DeviceRunner::run(Runtime &runtime, const CallConfig &config) { } auto runtime_args_cleanup = RAIIScopeGuard([this]() { + STRACE("simpler_run.runner_run.cleanup.runtime_args"); kernel_args_.finalize_device_kernel_args(); kernel_args_.finalize_runtime_args(); }); @@ -455,6 +464,7 @@ int DeviceRunner::run(Runtime &runtime, const CallConfig &config) { // Worker reused across runs (e.g. a pytest session-scoped worker pool) would // otherwise re-enter init_l2_swimlane() with stale state still allocated. auto perf_cleanup = RAIIScopeGuard([this]() { + STRACE("simpler_run.runner_run.cleanup.collectors"); finalize_collectors(); }); @@ -507,21 +517,27 @@ int DeviceRunner::run(Runtime &runtime, const CallConfig &config) { l2_swimlane_collector_.set_core_types(core_types.data(), num_aicore); } - rc = launch_run(runtime, num_aicore, launch_aicpu_num, selected_pipeline_slot); + { + STRACE("simpler_run.runner_run.launch_device"); + rc = launch_run(runtime, num_aicore, launch_aicpu_num, selected_pipeline_slot); + } if (rc != 0) return rc; - rc = reap_run(selected_pipeline_slot); + { + STRACE("simpler_run.runner_run.reap_device"); + rc = reap_run(selected_pipeline_slot); + } if (rc != 0) return rc; - // The run owns its AICore stream, so a destroy this run cannot complete is - // this run's failure: reporting success would leave the caller believing a - // slot is reusable that the next prepare will now refuse. + // Reap has established that no device work remains on this run's AICore + // stream. Native finalize owns its retire/replenish after output D2H. aicore_stream_retired = true; - rc = retire_run_aicore_stream(selected_pipeline_slot); - if (rc != 0) return rc; // Print handshake results (reads from device memory, must be before free) - print_handshake_results(); + { + STRACE("simpler_run.runner_run.print_handshake"); + print_handshake_results(); + } return 0; } @@ -543,6 +559,36 @@ int DeviceRunner::retire_run_aicore_stream(unsigned slot) { return rc; } +int DeviceRunner::retire_run_aicore_stream_async(unsigned slot) { + const unsigned trace_inv = simpler::strace::StraceScope::current_inv(); + const uint64_t trace_hid = simpler::strace::StraceScope::current_hid(); + int rc = run_stream_slots_.retire_aicore_async( + slot, + [this](std::function fn) { + return create_thread(std::move(fn)); + }, + [trace_inv, trace_hid](std::function fn) { + return [fn = std::move(fn), trace_inv, trace_hid]() { + STRACE_CONTEXT(trace_inv, trace_hid, 2); + const long long wall_start = STRACE_NOW_NS(); + const long long cpu_start = STRACE_THREAD_CPU_NOW_NS(); + { + STRACE("simpler_run.runner_run.async_destroy_aicore_stream"); + fn(); + } + STRACE_HOST_SPAN_AT( + "simpler_run.runner_run.async_destroy_aicore_stream.thread_cpu", wall_start, + STRACE_THREAD_CPU_NOW_NS() - cpu_start, 3 + ); + }; + } + ); + if (rc != 0) { + LOG_ERROR("async rtStreamDestroy launch (run AICore slot %u) failed: %d", slot, rc); + } + return rc; +} + int DeviceRunner::destroy_run_stream_sets() { // No pre-destroy sync, for the reason finalize_common() documents for the // bootstrap pair: rtStreamDestroy is the supported teardown for a stream @@ -619,7 +665,11 @@ int DeviceRunner::reap_run(unsigned slot) { LOG_ERROR("reap_run: invalid stream set %u", slot); return -1; } - int rc = sync_stream_pair(run_stream_slots_.aicpu(slot), run_stream_slots_.aicore(slot)); + int rc = 0; + { + STRACE("simpler_run.runner_run.native_fence_wait"); + rc = sync_stream_pair(run_stream_slots_.aicpu(slot), run_stream_slots_.aicore(slot)); + } 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 @@ -639,11 +689,17 @@ int DeviceRunner::reap_run(unsigned slot) { return rc; } - read_device_wall_ns(); + { + STRACE("simpler_run.runner_run.read_device_wall"); + read_device_wall_ns(); + } // 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). - teardown_shared_collectors_after_run(); + { + STRACE("simpler_run.runner_run.teardown_collectors"); + teardown_shared_collectors_after_run(); + } // a2a3-only dep_gen teardown: host-orch emits the graph snapshot adopted // from the prepare thread; device-orch stops the collector, reconciles the diff --git a/src/a2a3/platform/onboard/host/device_runner.h b/src/a2a3/platform/onboard/host/device_runner.h index dfd35984cf..808e07a16d 100644 --- a/src/a2a3/platform/onboard/host/device_runner.h +++ b/src/a2a3/platform/onboard/host/device_runner.h @@ -119,6 +119,7 @@ class DeviceRunner : public DeviceRunnerBase { bool can_accept_run() const override { return !device_unusable_.load(std::memory_order_acquire); } int provision_native_run_resources(uint32_t pipeline_slot) override; int abandon_native_run_resources(uint32_t pipeline_slot) override; + int complete_native_run_resources(uint32_t pipeline_slot) override; // Map/unmap a device buffer into host address space via // halHostRegister(DEV_SVM_MAP_HOST) / halHostUnregister. The returned host @@ -264,6 +265,7 @@ class DeviceRunner : public DeviceRunnerBase { // image's instructions, so the slot must refuse the next run until finalize // reclaims it. int retire_run_aicore_stream(unsigned slot); + int retire_run_aicore_stream_async(unsigned slot); int destroy_run_stream_sets(); // The kernel submission boundary is separate from the stream wait and 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 6f5cdacb8a..e11175e82f 100644 --- a/src/a2a3/runtime/tensormap_and_ringbuffer/host/runtime_maker.cpp +++ b/src/a2a3/runtime/tensormap_and_ringbuffer/host/runtime_maker.cpp @@ -981,14 +981,10 @@ extern "C" int validate_runtime_impl(Runtime *runtime, const HostApi *api, int e int rc = 0; - LOG_INFO("=== Copying Results Back to Host ==="); - // Copy all recorded tensors from device back to host TensorLease *tensor_leases = runtime->tensor_leases_.data(); int tensor_lease_count = static_cast(runtime->tensor_leases_.size()); - LOG_INFO("Tensor leases to process: %d", tensor_lease_count); - bool skip_tensor_copy_back = execution_rc != 0; int32_t runtime_status = 0; PTO2SharedMemoryHeader host_header; @@ -1024,6 +1020,7 @@ extern "C" int validate_runtime_impl(Runtime *runtime, const HostApi *api, int e if (skip_tensor_copy_back) { LOG_WARN("Skipping tensor copy-back because execution failed"); } else { + STRACE("simpler_run.validate.copy_back"); for (int i = 0; i < tensor_lease_count; i++) { const TensorLease &lease = tensor_leases[i]; @@ -1047,6 +1044,9 @@ extern "C" int validate_runtime_impl(Runtime *runtime, const HostApi *api, int e continue; } + char copy_attrs[96]; + std::snprintf(copy_attrs, sizeof(copy_attrs), "index=%d bytes=%zu", i, lease.size); + STRACE_A("simpler_run.validate.copy_back.tensor", copy_attrs); int copy_rc = api->copy_from_device(lease.host_ptr, lease.dev_ptr, lease.size); if (copy_rc != 0) { LOG_ERROR("Failed to copy tensor %d from device: %d", i, copy_rc); @@ -1058,10 +1058,10 @@ extern "C" int validate_runtime_impl(Runtime *runtime, const HostApi *api, int e } // Cleanup device tensors - LOG_INFO("=== Cleaning Up ==="); - release_tensor_leases(runtime, api); - - LOG_INFO("=== Finalize Complete ==="); + { + STRACE("simpler_run.validate.release_leases"); + release_tensor_leases(runtime, api); + } if (rc == 0 && runtime_status != 0) { rc = runtime_status; diff --git a/src/common/log/include/common/strace.h b/src/common/log/include/common/strace.h index 5543e5a15e..7165ac3903 100644 --- a/src/common/log/include/common/strace.h +++ b/src/common/log/include/common/strace.h @@ -59,8 +59,10 @@ #include #include +#include #include #include +#include #include @@ -112,6 +114,62 @@ inline long strace_tid() { #endif } +/** + * Emit one host marker. Profiling can redirect markers to one fully-buffered + * file per process with SIMPLER_HOST_STRACE_DIR. This avoids making every + * short scope synchronously flush the shared stderr pipe -- at eight ranks the + * observer itself otherwise creates millisecond scheduling holes between + * adjacent tensor copies. The outermost marker flushes the private file once + * per invocation, so a completed invocation is always available to the parser. + */ +inline void write_span( + const char *name, long long ts_ns, long long dur_ns, int depth, unsigned inv, uint64_t hid, const char *attrs +) { + const char *directory = std::getenv("SIMPLER_HOST_STRACE_DIR"); + if (directory == nullptr || directory[0] == '\0') { + LOG_TIMING( + "[STRACE] v=1 pid=%d tid=%ld inv=%u hid=%llx depth=%d name=%s ts=%lld dur=%lld %s", + static_cast(getpid()), strace_tid(), inv, static_cast(hid), depth, name, ts_ns, + dur_ns, attrs + ); + return; + } + + static pthread_mutex_t mutex = PTHREAD_MUTEX_INITIALIZER; + static FILE *stream = nullptr; + static pid_t stream_pid = -1; + pthread_mutex_lock(&mutex); + const pid_t pid = getpid(); + if (stream == nullptr || stream_pid != pid) { + // Do not close a stream inherited across fork: another stdio object may + // still own the copied buffer. Each child switches to its own PID file. + char path[4096]; + const int count = std::snprintf(path, sizeof(path), "%s/host-strace.%d.log", directory, pid); + if (count > 0 && static_cast(count) < sizeof(path)) { + stream = std::fopen(path, "a"); + if (stream != nullptr) { + std::setvbuf(stream, nullptr, _IOFBF, 1U << 20U); + stream_pid = pid; + } + } + } + if (stream != nullptr && stream_pid == pid) { + std::fprintf( + stream, "[STRACE] v=1 pid=%d tid=%ld inv=%u hid=%llx depth=%d name=%s ts=%lld dur=%lld %s\n", + static_cast(pid), strace_tid(), inv, static_cast(hid), depth, name, ts_ns, dur_ns, + attrs + ); + if (depth == 0) std::fflush(stream); + pthread_mutex_unlock(&mutex); + return; + } + pthread_mutex_unlock(&mutex); + LOG_TIMING( + "[STRACE] v=1 pid=%d tid=%ld inv=%u hid=%llx depth=%d name=%s ts=%lld dur=%lld %s", static_cast(pid), + strace_tid(), inv, static_cast(hid), depth, name, ts_ns, dur_ns, attrs + ); +} + class StraceScope { public: explicit StraceScope(const char *name, const char *attrs = "") : @@ -131,11 +189,7 @@ class StraceScope { // depth printed is the scope's own level (post-decrement so the // outermost scope prints depth=0). const int d = --depth(); - LOG_TIMING( - "[STRACE] v=1 pid=%d tid=%ld inv=%u hid=%llx depth=%d name=%s ts=%lld dur=%lld %s", - static_cast(getpid()), strace_tid(), inv(), static_cast(hid()), d, name_, ts, dur, - attrs_ - ); + write_span(name_, ts, dur, d, inv(), hid(), attrs_); } StraceScope(const StraceScope &) = delete; @@ -208,6 +262,13 @@ inline long long strace_now_ns() { ); } +/** Current per-thread CPU clock; diagnostic companion to wall-clock spans. */ +inline long long strace_thread_cpu_now_ns() { + struct timespec value{}; + if (clock_gettime(CLOCK_THREAD_CPUTIME_ID, &value) != 0) return 0; + return static_cast(value.tv_sec) * 1000000000LL + value.tv_nsec; +} + /** * Emit a marker for a span whose duration was measured elsewhere (e.g. a device * phase: AICPU cycles → ns). Shares the current thread's inv/hid grouping so the @@ -220,11 +281,7 @@ inline long long strace_now_ns() { */ inline void emit_span_at(const char *name, long long ts_ns, long long dur_ns, int depth, const char *attrs = "clk=dev") { - LOG_TIMING( - "[STRACE] v=1 pid=%d tid=%ld inv=%u hid=%llx depth=%d name=%s ts=%lld dur=%lld %s", static_cast(getpid()), - strace_tid(), StraceScope::current_inv(), static_cast(StraceScope::current_hid()), depth, - name, ts_ns, dur_ns, attrs - ); + write_span(name, ts_ns, dur_ns, depth, StraceScope::current_inv(), StraceScope::current_hid(), attrs); } /** Emit an explicitly timed host-domain span in the active invocation. */ @@ -253,6 +310,7 @@ inline void emit_host_span_at(const char *name, long long ts_ns, long long dur_n ::simpler::strace::StraceContextScope STRACE_CAT(_strace_context_, __LINE__)((inv), (hid), (depth)) /** Read the current host monotonic clock in nanoseconds. */ #define STRACE_NOW_NS() ::simpler::strace::strace_now_ns() +#define STRACE_THREAD_CPU_NOW_NS() ::simpler::strace::strace_thread_cpu_now_ns() /** Emit a host-domain span measured across disjoint API calls. */ #define STRACE_HOST_SPAN_AT(name, ts_ns, dur_ns, depth) \ ::simpler::strace::emit_host_span_at((name), (ts_ns), (dur_ns), (depth)) @@ -272,6 +330,7 @@ inline void emit_host_span_at(const char *name, long long ts_ns, long long dur_n #define STRACE_SET_HID(h) ((void)0) #define STRACE_CONTEXT(inv, hid, depth) ((void)0) #define STRACE_NOW_NS() 0LL +#define STRACE_THREAD_CPU_NOW_NS() 0LL #define STRACE_HOST_SPAN_AT(name, ts_ns, dur_ns, depth) ((void)0) #define STRACE_HOST_SPAN_AT_A(name, ts_ns, dur_ns, depth, attrs) ((void)0) #define STRACE_DEV_SPAN_AT(name, ts_ns, dur_ns, depth) ((void)0) diff --git a/src/common/platform/include/host/run_stream_slots.h b/src/common/platform/include/host/run_stream_slots.h index d5b18c499b..bf0eadf2f5 100644 --- a/src/common/platform/include/host/run_stream_slots.h +++ b/src/common/platform/include/host/run_stream_slots.h @@ -15,7 +15,11 @@ #include #include #include +#include +#include #include +#include +#include #include "pto_runtime_c_api.h" @@ -35,24 +39,30 @@ * to retry. Stream creation and destruction are injected so this state machine * is exercisable without a device. * - * Threading: a `Slot` is touched only by the thread that owns that slot's run, - * and admission gives at most one owner per slot, so the per-slot handles need - * no synchronization. Two slots are therefore serviced concurrently — a native - * prepare acquires the successor's slot while the executor retires the - * predecessor's. `created_count_` is the one field shared across those owners - * and is additionally readable from an unrelated thread through - * `get_run_stream_set_create_count`, so it is atomic. `destroy_all()` walks - * every slot and requires all runs to be quiesced. + * Threading: admission gives at most one run owner per slot. A single resident + * lifecycle worker serializes async retire/replenish for every slot. This keeps + * two slow, CPU-heavy rtStreamDestroy/rtStreamCreate calls from competing with + * each other in one rank while the device pipeline remains depth two. The next + * owner waits only when its own slot is still being replenished. State access is + * protected separately from the CANN calls, so reading the active peer slot is + * never blocked behind a slow lifecycle call. `created_count_` is additionally + * readable from an unrelated thread through `get_run_stream_set_create_count`, + * so it is atomic. `destroy_all()` walks every slot and requires all runs to be + * quiesced. */ class RunStreamSlots { public: using CreateFn = std::function; using DestroyFn = std::function; + using ThreadFactory = std::function)>; + using JobWrapper = std::function(std::function)>; RunStreamSlots(CreateFn create, DestroyFn destroy) : create_(std::move(create)), destroy_(std::move(destroy)) {} + ~RunStreamSlots() { stop_worker(); } + /** * Ready `slot` for a run: its AICPU stream on first use, and always a fresh * AICore stream. Fails when the slot still holds an AICore stream a prior @@ -60,19 +70,50 @@ class RunStreamSlots { */ int acquire(unsigned slot) { if (slot >= slots_.size()) return -1; - Slot &s = slots_[slot]; - if (s.aicpu == nullptr) { - int rc = create_(&s.aicpu); + { + std::unique_lock lock(state_mutex_); + state_cv_.wait(lock, [this, slot]() { + return !slots_[slot].destroying; + }); + Slot &s = slots_[slot]; + if (s.destroy_rc != 0) { + int rc = s.destroy_rc; + s.destroy_rc = 0; + return rc; + } + if (s.aicore_stranded) return -1; + if (s.aicore != nullptr) return 0; + } + + void *new_aicpu = nullptr; + void *new_aicore = nullptr; + { + std::lock_guard lifecycle_lock(lifecycle_mutex_); + bool need_aicpu = false; + { + std::lock_guard state_lock(state_mutex_); + need_aicpu = slots_[slot].aicpu == nullptr; + } + if (need_aicpu) { + int rc = create_(&new_aicpu); + if (rc != 0) return rc; + } + int rc = create_(&new_aicore); if (rc != 0) { - s.aicpu = nullptr; + if (new_aicpu != nullptr) { + std::lock_guard state_lock(state_mutex_); + slots_[slot].aicpu = new_aicpu; + } return rc; } } - if (s.aicore != nullptr) return -1; - int rc = create_(&s.aicore); - if (rc != 0) { - s.aicore = nullptr; - return rc; + { + std::lock_guard lock(state_mutex_); + Slot &s = slots_[slot]; + if (new_aicpu != nullptr) s.aicpu = new_aicpu; + s.aicore = new_aicore; + s.destroy_rc = 0; + s.aicore_stranded = false; } created_count_.fetch_add(1, std::memory_order_relaxed); return 0; @@ -81,34 +122,129 @@ class RunStreamSlots { /** Retire `slot`'s AICore stream. The handle survives a failed destroy. */ int retire_aicore(unsigned slot) { if (slot >= slots_.size()) return -1; + void *retired = nullptr; + { + std::unique_lock lock(state_mutex_); + state_cv_.wait(lock, [this, slot]() { + return !slots_[slot].destroying; + }); + Slot &s = slots_[slot]; + if (s.destroy_rc != 0) { + int rc = s.destroy_rc; + s.destroy_rc = 0; + return rc; + } + retired = s.aicore; + if (retired == nullptr) return 0; + } + int rc = 0; + { + std::lock_guard lifecycle_lock(lifecycle_mutex_); + rc = destroy_(retired); + } + { + std::lock_guard lock(state_mutex_); + Slot &s = slots_[slot]; + // The synchronous caller observes `rc` directly. Keep only the + // poisoned-handle state; otherwise teardown would report the same + // already-observed destroy failure a second time before retrying it. + s.destroy_rc = 0; + s.aicore_stranded = rc != 0; + if (rc == 0) s.aicore = nullptr; + } + return rc; + } + + /** + * Start retirement after the stream has been reaped, then immediately + * create the fresh stream for the slot's next owner. The handle stays in + * the slot until destroy succeeds; acquire() joins and checks both actions. + */ + int retire_aicore_async( + unsigned slot, const ThreadFactory &thread_factory, const JobWrapper &job_wrapper = JobWrapper{} + ) { + if (slot >= slots_.size()) return -1; + std::unique_lock lock(state_mutex_); Slot &s = slots_[slot]; - if (s.aicore == nullptr) return 0; - int rc = destroy_(s.aicore); - if (rc != 0) return rc; - s.aicore = nullptr; + if (stopping_ || s.destroying || s.aicore == nullptr) return -1; + if (!worker_started_) { + try { + retire_worker_ = thread_factory([this]() { + worker_loop(); + }); + } catch (...) { + return -1; + } + if (!retire_worker_.joinable()) return -1; + worker_started_ = true; + } + std::function job = [this, slot]() { + retire_and_replenish(slot); + }; + try { + if (job_wrapper) job = job_wrapper(std::move(job)); + queue_.push_back(std::move(job)); + } catch (...) { + return -1; + } + s.destroying = true; + s.destroy_rc = 0; + lock.unlock(); + worker_cv_.notify_one(); return 0; } /** Destroy every stream, keeping handles whose destroy failed. */ int destroy_all() { int first_error = 0; - for (Slot &s : slots_) { - for (void **stream : {&s.aicpu, &s.aicore}) { - if (*stream == nullptr) continue; - int rc = destroy_(*stream); + for (unsigned slot = 0; slot < slots_.size(); ++slot) { + void *streams[2] = {nullptr, nullptr}; + { + std::unique_lock lock(state_mutex_); + state_cv_.wait(lock, [this, slot]() { + return !slots_[slot].destroying; + }); + Slot &s = slots_[slot]; + if (s.destroy_rc != 0 && first_error == 0) first_error = s.destroy_rc; + streams[0] = s.aicpu; + streams[1] = s.aicore; + } + for (unsigned index = 0; index < 2; ++index) { + if (streams[index] == nullptr) continue; + int rc = 0; + { + std::lock_guard lifecycle_lock(lifecycle_mutex_); + rc = destroy_(streams[index]); + } if (rc != 0) { if (first_error == 0) first_error = rc; continue; } - *stream = nullptr; + std::lock_guard lock(state_mutex_); + Slot &s = slots_[slot]; + if (index == 0 && s.aicpu == streams[index]) s.aicpu = nullptr; + if (index == 1 && s.aicore == streams[index]) { + s.aicore = nullptr; + s.aicore_stranded = false; + } } } return first_error; } - void *aicpu(unsigned slot) const { return slot < slots_.size() ? slots_[slot].aicpu : nullptr; } - void *aicore(unsigned slot) const { return slot < slots_.size() ? slots_[slot].aicore : nullptr; } - bool ready(unsigned slot) const { return aicpu(slot) != nullptr && aicore(slot) != nullptr; } + void *aicpu(unsigned slot) const { + std::lock_guard lock(state_mutex_); + return slot < slots_.size() ? slots_[slot].aicpu : nullptr; + } + void *aicore(unsigned slot) const { + std::lock_guard lock(state_mutex_); + return slot < slots_.size() ? slots_[slot].aicore : nullptr; + } + bool ready(unsigned slot) const { + std::lock_guard lock(state_mutex_); + return slot < slots_.size() && !slots_[slot].destroying && slots_[slot].aicpu != nullptr && + slots_[slot].aicore != nullptr; + } size_t created_count() const { return created_count_.load(std::memory_order_relaxed); } static constexpr size_t capacity() { return PTO_PIPELINE_MAX_DEPTH; } @@ -116,12 +252,78 @@ class RunStreamSlots { struct Slot { void *aicpu{nullptr}; void *aicore{nullptr}; + int destroy_rc{0}; + bool destroying{false}; + bool aicore_stranded{false}; }; + void worker_loop() { + for (;;) { + std::function job; + { + std::unique_lock lock(state_mutex_); + worker_cv_.wait(lock, [this]() { + return stopping_ || !queue_.empty(); + }); + if (stopping_ && queue_.empty()) return; + job = std::move(queue_.front()); + queue_.pop_front(); + } + job(); + } + } + + void retire_and_replenish(unsigned slot) noexcept { + void *retired = nullptr; + { + std::lock_guard lock(state_mutex_); + retired = slots_[slot].aicore; + } + void *fresh = retired; + int rc = -1; + try { + std::lock_guard lifecycle_lock(lifecycle_mutex_); + rc = destroy_(retired); + if (rc == 0) { + fresh = nullptr; + rc = create_(&fresh); + } + } catch (...) { + rc = -1; + } + { + std::lock_guard lock(state_mutex_); + Slot &s = slots_[slot]; + s.aicore = fresh; + s.destroy_rc = rc; + s.destroying = false; + s.aicore_stranded = rc != 0 && fresh == retired; + } + if (rc == 0) created_count_.fetch_add(1, std::memory_order_relaxed); + state_cv_.notify_all(); + } + + void stop_worker() noexcept { + { + std::lock_guard lock(state_mutex_); + stopping_ = true; + } + worker_cv_.notify_all(); + if (retire_worker_.joinable()) retire_worker_.join(); + } + CreateFn create_; DestroyFn destroy_; std::array slots_{}; std::atomic created_count_{0}; + mutable std::mutex state_mutex_; + std::condition_variable state_cv_; + std::condition_variable worker_cv_; + std::mutex lifecycle_mutex_; + std::deque> queue_; + std::thread retire_worker_{}; + bool worker_started_{false}; + bool stopping_{false}; }; #endif // SRC_COMMON_PLATFORM_INCLUDE_HOST_RUN_STREAM_SLOTS_H_ diff --git a/src/common/platform/onboard/host/c_api_shared.cpp b/src/common/platform/onboard/host/c_api_shared.cpp index c6322d65b9..4b71fcd36c 100644 --- a/src/common/platform/onboard/host/c_api_shared.cpp +++ b/src/common/platform/onboard/host/c_api_shared.cpp @@ -36,10 +36,15 @@ #include #include +#include #include #include +#include +#include +#include #include #include +#include #include #include @@ -60,6 +65,115 @@ using OnboardNativeRunState = NativeRunState; // lifetime, so the on-storage magic must remain the leading bytes. static_assert(__builtin_offsetof(OnboardNativeRunState, magic) == 0, "native-run magic must lead runtime storage"); +namespace { + +// Keep two device-bound host executors resident for the depth-two pipeline. +// Per-run std::thread construction occasionally stalls all ranks for tens of +// milliseconds; if one rank stalls longer, the other ranks enter the device +// collective early and the same host jitter merely appears inside runner_run. +class NativeExecutorPool { +public: + explicit NativeExecutorPool(DeviceRunnerBase *runner) : + runner_(runner) { + for (unsigned i = 0; i < PTO_PIPELINE_MAX_DEPTH; ++i) { + workers_.push_back(runner_->create_thread([this]() { + worker_loop(); + })); + } + } + + ~NativeExecutorPool() { stop(); } + + NativeExecutorPool(const NativeExecutorPool &) = delete; + NativeExecutorPool &operator=(const NativeExecutorPool &) = delete; + + bool submit(std::function fn) { + { + std::lock_guard lock(mutex_); + if (stopping_) return false; + queue_.push_back(std::move(fn)); + } + cv_.notify_one(); + return true; + } + + void stop() { + { + std::lock_guard lock(mutex_); + if (stopping_) return; + stopping_ = true; + } + cv_.notify_all(); + for (std::thread &worker : workers_) { + if (worker.joinable()) worker.join(); + } + workers_.clear(); + } + +private: + void worker_loop() { + for (;;) { + std::function fn; + { + std::unique_lock lock(mutex_); + cv_.wait(lock, [this]() { + return stopping_ || !queue_.empty(); + }); + if (stopping_ && queue_.empty()) return; + fn = std::move(queue_.front()); + queue_.pop_front(); + } + try { + fn(); + } catch (...) { + // Per-run closures publish their own failure. Never let one + // malformed run permanently shrink the resident pool. + } + } + } + + DeviceRunnerBase *runner_; + std::mutex mutex_; + std::condition_variable cv_; + std::deque> queue_; + std::vector workers_; + bool stopping_{false}; +}; + +std::mutex g_native_executor_pools_mutex; +std::unordered_map> g_native_executor_pools; + +NativeExecutorPool *get_native_executor_pool(DeviceContextHandle ctx) { + std::lock_guard lock(g_native_executor_pools_mutex); + auto it = g_native_executor_pools.find(ctx); + return it == g_native_executor_pools.end() ? nullptr : it->second.get(); +} + +bool install_native_executor_pool(DeviceContextHandle ctx, DeviceRunnerBase *runner) { + std::unique_ptr pool; + try { + pool = std::make_unique(runner); + } catch (...) { + return false; + } + std::lock_guard lock(g_native_executor_pools_mutex); + return g_native_executor_pools.emplace(ctx, std::move(pool)).second; +} + +void remove_native_executor_pool(DeviceContextHandle ctx) { + std::unique_ptr pool; + { + std::lock_guard lock(g_native_executor_pools_mutex); + auto it = g_native_executor_pools.find(ctx); + if (it == g_native_executor_pools.end()) return; + pool = std::move(it->second); + g_native_executor_pools.erase(it); + } + pool->stop(); +} + +} // namespace + extern "C" { /* =========================================================================== @@ -264,6 +378,7 @@ void destroy_device_context(DeviceContextHandle ctx) { LOG_ERROR("destroy_device_context: refusing to destroy a context with an unfinalized native run"); return; } + remove_native_executor_pool(ctx); delete runner; } @@ -313,6 +428,7 @@ int finalize_device(DeviceContextHandle ctx) { LOG_ERROR("finalize_device: native run must be finalized first"); return -1; } + remove_native_executor_pool(ctx); return runner->finalize(); } catch (...) { return -1; @@ -404,6 +520,10 @@ int simpler_init( } if (rc != 0) return rc; } + if (!install_native_executor_pool(ctx, runner)) { + LOG_ERROR("simpler_init: failed to create resident native executor pool"); + return -1; + } return 0; } @@ -505,6 +625,18 @@ static bool device_profiling_enabled() { return enabled; } +static uint64_t device_profiling_min_wall_ns() { + static const uint64_t min_wall_ns = [] { + const char *value = std::getenv("SIMPLER_DEVICE_STRACE_MIN_WALL_US"); + if (value == nullptr || *value == '\0') return UINT64_C(0); + char *end = nullptr; + unsigned long long us = std::strtoull(value, &end, 10); + if (end == value || *end != '\0') return UINT64_C(0); + return static_cast(us) * UINT64_C(1000); + }(); + return min_wall_ns; +} + // Emit device-domain trace markers for the AICPU phases. RunWall (the whole // on-NPU wall, i.e. the former RunTiming.device_wall) is emitted at depth 2 // under runner_run; its preamble/so_load/graph_build/post_orch subdivisions are @@ -517,6 +649,11 @@ static void emit_device_phase_markers(DeviceRunnerBase *runner) { if (run_wall_ns != 0) { STRACE_DEV_SPAN_AT("simpler_run.runner_run.device_wall", 0, static_cast(run_wall_ns), 2); } + // A single RunWall marker per run is cheap and is required to distinguish + // real device tails from a host waiter that resumed late. Detailed phase + // and task-slot markers remain thresholded to avoid perturbing normal D2H + // and stream-retirement paths. + if (run_wall_ns < device_profiling_min_wall_ns()) return; struct PhaseName { AicpuPhase phase; const char *name; @@ -705,20 +842,33 @@ int simpler_prepare_run( state->trace_start_ns = trace_start_ns; STRACE_CONTEXT(state->trace_inv, state->trace_hid, 1); - int rc = runner->attach_current_thread(runner->device_id()); + int rc = -1; + { + STRACE("simpler_run.prepare.attach"); + rc = runner->attach_current_thread(runner->device_id()); + } if (rc != 0) return cleanup_failed_prepare(state, rc, true); state->runner_resources_owned = true; - rc = runner->provision_native_run_resources(state->pipeline_slot); + { + STRACE("simpler_run.prepare.resources"); + rc = runner->provision_native_run_resources(state->pipeline_slot); + } if (rc != 0) return cleanup_failed_prepare(state, rc, true); - rc = runner->prepare_launch_shape(state->runtime, state->config); + { + STRACE("simpler_run.prepare.launch_shape"); + rc = runner->prepare_launch_shape(state->runtime, state->config); + } if (rc != 0) return cleanup_failed_prepare(state, rc, true); // Diagnostic binding reads runner-global collector configuration. It // is depth-one, while concurrent HBG preparation must leave the active // run's configuration untouched until launch. - if (!overlaps_active_run) runner->apply_call_config(state->config); + if (!overlaps_active_run) { + STRACE("simpler_run.prepare.apply_config"); + runner->apply_call_config(state->config); + } { STRACE("simpler_run.bind"); @@ -728,7 +878,95 @@ int simpler_prepare_run( ); } if (rc != 0) return cleanup_failed_prepare(state, rc, true); - state->host_thread_state = runner->take_native_run_thread_state(); + { + STRACE("simpler_run.prepare.take_thread_state"); + state->host_thread_state = runner->take_native_run_thread_state(); + } + + // Create and attach the blocking executor while this successor is + // prepared. In the two-frame HBG path prepare runs independently from + // the progress owner, so thread startup overlaps the active device run + // instead of delaying the successor after the execution claim opens. + { + STRACE("simpler_run.prepare.create_executor"); + NativeExecutorPool *executor_pool = get_native_executor_pool(ctx); + if (executor_pool == nullptr) return cleanup_failed_prepare(state, -1, true); + const DeviceRunnerBase::NativeRunThreadSelection executor_selection = + state->runner->capture_native_run_thread_selection(); + state->executor_submitted = executor_pool->submit([state, ctx, executor_selection]() { + auto task_done_guard = RAIIScopeGuard([state]() { + state->executor_task_done_signal.notify(); + }); + pthread_once(&g_runner_key_once, create_runner_key); + pthread_setspecific(g_runner_key, ctx); + state->runner->restore_native_run_thread_selection(executor_selection); + STRACE_CONTEXT(state->trace_inv, state->trace_hid, 1); + int rc = -1; + bool entered_run = false; + int attach_rc = -1; + try { + { + STRACE("simpler_run.prepare.executor_attach"); + attach_rc = state->runner->attach_current_thread(state->runner->device_id()); + } + } catch (...) { + attach_rc = -1; + } + state->executor_attach_rc.store(attach_rc, std::memory_order_release); + state->executor_ready_signal.notify(); + state->executor_start_signal.wait(); + if (state->executor_cancelled.load(std::memory_order_acquire)) { + pthread_setspecific(g_runner_key, nullptr); + return; + } + try { + if (attach_rc == 0) { + { + STRACE("simpler_run.launch.adopt_thread_state"); + state->adopt_host_thread_state(); + } + { + STRACE("simpler_run.launch.activate_shape"); + state->runner->activate_launch_shape(state->runtime); + } + { + STRACE("simpler_run.runner_run"); + entered_run = true; + rc = state->runner->run(state->runtime, state->config); + } + } else { + rc = attach_rc; + } + } catch (...) { + rc = -1; + } + if (entered_run && rc != 0) { + // Error exits retire synchronously inside DeviceRunner::run. + // A successful run keeps ownership until finalize has + // completed output D2H, then starts async replenish. + state->runner_resources_owned = false; + } else if (!entered_run && state->runner_resources_owned) { + int resources_rc = -1; + try { + resources_rc = state->runner->abandon_native_run_resources(state->pipeline_slot); + } catch (...) {} + state->runner_resources_owned = false; + if (rc == 0) rc = resources_rc; + } + pthread_setspecific(g_runner_key, nullptr); + state->execution_rc.store(rc, std::memory_order_relaxed); + state->execution_done.store(true, std::memory_order_release); + state->completion_signal.notify(); + state->launch_signal.notify(); + }); + if (!state->executor_submitted) return cleanup_failed_prepare(state, -1, true); + } + { + STRACE("simpler_run.prepare.executor_ready_wait"); + state->executor_ready_signal.wait(); + } + rc = state->executor_attach_rc.load(std::memory_order_acquire); + if (rc != 0) return cleanup_failed_prepare(state, rc, true); return 0; } catch (...) { if (state != nullptr) return cleanup_failed_prepare(state, -1, true); @@ -740,7 +978,13 @@ int simpler_launch_run(DeviceContextHandle ctx, RuntimeHandle runtime) { OnboardNativeRunState *state = native_run_state(ctx, runtime, "simpler_launch_run"); if (state == nullptr || state->phase.load(std::memory_order_acquire) != NativeRunPhase::Prepared) return -1; if (!state->runner->can_accept_run() || !state->runner_reserved) return -1; - if (!state->runner->try_acquire_native_run(state, &state->launch_signal)) { + STRACE_CONTEXT(state->trace_inv, state->trace_hid, 1); + bool acquired = false; + { + STRACE("simpler_run.launch.claim"); + acquired = state->runner->try_acquire_native_run(state, &state->launch_signal); + } + if (!acquired) { LOG_ERROR("simpler_launch_run: execution claim is occupied (%s)", state->trace_attrs); return -1; } @@ -761,8 +1005,15 @@ int simpler_launch_run(DeviceContextHandle ctx, RuntimeHandle runtime) { auto selection_guard = RAIIScopeGuard([runner = state->runner, caller_selection]() { runner->restore_native_run_thread_selection(caller_selection); }); - if (state->runner->select_pipeline_slot(state->pipeline_slot) != 0 || - state->runner->select_arena_bank(state->arena_bank) != 0) { + int select_rc = 0; + { + STRACE("simpler_run.launch.select_resources"); + if (state->runner->select_pipeline_slot(state->pipeline_slot) != 0 || + state->runner->select_arena_bank(state->arena_bank) != 0) { + select_rc = -1; + } + } + if (select_rc != 0) { state->runner->release_native_run(state); state->runner_claimed = false; return -1; @@ -770,55 +1021,11 @@ int simpler_launch_run(DeviceContextHandle ctx, RuntimeHandle runtime) { state->phase.store(NativeRunPhase::Launching, std::memory_order_release); - try { - // The compatibility backend uses one blocking executor per run. The - // prepare-through-finalize runner claim limits it to one per context. - state->executor = state->runner->create_thread([state, ctx]() { - pthread_once(&g_runner_key_once, create_runner_key); - pthread_setspecific(g_runner_key, ctx); - STRACE_CONTEXT(state->trace_inv, state->trace_hid, 1); - int rc = -1; - bool entered_run = false; - try { - int attach_rc = state->runner->attach_current_thread(state->runner->device_id()); - if (attach_rc == 0) { - state->adopt_host_thread_state(); - state->runner->activate_launch_shape(state->runtime); - { - STRACE("simpler_run.runner_run"); - entered_run = true; - rc = state->runner->run(state->runtime, state->config); - } - } else { - rc = attach_rc; - } - } catch (...) { - rc = -1; - } - if (entered_run) { - // run() owns stream retirement on every exit once entered. - state->runner_resources_owned = false; - } else if (state->runner_resources_owned) { - int resources_rc = -1; - try { - resources_rc = state->runner->abandon_native_run_resources(state->pipeline_slot); - } catch (...) {} - state->runner_resources_owned = false; - if (rc == 0) rc = resources_rc; - } - pthread_setspecific(g_runner_key, nullptr); - state->execution_rc.store(rc, std::memory_order_relaxed); - state->execution_done.store(true, std::memory_order_release); - state->launch_signal.notify(); - }); - } catch (...) { - state->runner->release_native_run(state); - state->runner_claimed = false; - state->phase.store(NativeRunPhase::Prepared, std::memory_order_release); - return -1; + state->executor_start_signal.notify(); + { + STRACE("simpler_run.launch.handoff_wait"); + state->launch_signal.wait(); } - - state->launch_signal.wait(); if (state->execution_done.load(std::memory_order_acquire)) { state->phase.store(NativeRunPhase::Complete, std::memory_order_release); return state->execution_rc.load(std::memory_order_relaxed); @@ -836,6 +1043,17 @@ int simpler_poll_run(DeviceContextHandle ctx, RuntimeHandle runtime) { state->phase.store(NativeRunPhase::Complete, std::memory_order_release); return SIMPLER_NATIVE_RUN_POLL_COMPLETE; } + // Keep mailbox progress responsive while avoiding a Python-side busy loop. + // The executor wakes this wait immediately on completion; the timeout is a + // bound for control/shutdown scans, not an added completion latency. + // The 100 us timeout keeps control/shutdown latency bounded. A 1 ms + // experiment reduced finalize polling but moved large tails into successor + // resource preparation and launch, so it is intentionally not used. + state->completion_signal.wait_for(std::chrono::microseconds(100)); + if (state->execution_done.load(std::memory_order_acquire)) { + state->phase.store(NativeRunPhase::Complete, std::memory_order_release); + return SIMPLER_NATIVE_RUN_POLL_COMPLETE; + } return SIMPLER_NATIVE_RUN_POLL_NOT_READY; } @@ -844,7 +1062,7 @@ int simpler_wait_run(DeviceContextHandle ctx, RuntimeHandle runtime) { if (state == nullptr) return -1; NativeRunPhase phase = state->phase.load(std::memory_order_acquire); if (phase == NativeRunPhase::Prepared || phase == NativeRunPhase::Launching) return -1; - if (state->executor.joinable()) state->executor.join(); + state->completion_signal.wait(); state->phase.store(NativeRunPhase::Complete, std::memory_order_release); return state->execution_rc.load(std::memory_order_relaxed); } @@ -875,22 +1093,32 @@ int simpler_finalize_run(DeviceContextHandle ctx, RuntimeHandle runtime) { auto selection_guard = RAIIScopeGuard([runner = state->runner, caller_selection]() { runner->restore_native_run_thread_selection(caller_selection); }); - if (state->runner->select_pipeline_slot(state->pipeline_slot) != 0 || - state->runner->select_arena_bank(state->arena_bank) != 0) { - return -1; + { + STRACE("simpler_run.finalize.select_resources"); + if (state->runner->select_pipeline_slot(state->pipeline_slot) != 0 || + state->runner->select_arena_bank(state->arena_bank) != 0) { + return -1; + } } int execution_rc = -1; const bool launched = phase != NativeRunPhase::Prepared; if (launched) { - if (state->executor.joinable()) state->executor.join(); + { + STRACE("simpler_run.finalize.join_executor"); + state->completion_signal.wait(); + } execution_rc = state->execution_rc.load(std::memory_order_relaxed); } int validation_rc = -1; try { if (!launched) state->runtime.set_gm_sm_ptr(nullptr); - int attach_rc = state->runner->attach_current_thread(state->runner->device_id()); + int attach_rc = -1; + { + STRACE("simpler_run.finalize.attach_thread"); + attach_rc = state->runner->attach_current_thread(state->runner->device_id()); + } if (attach_rc == 0) { { STRACE("simpler_run.validate"); @@ -905,9 +1133,14 @@ int simpler_finalize_run(DeviceContextHandle ctx, RuntimeHandle runtime) { } int resources_rc = 0; - if (!launched && state->runner_resources_owned) { + if (state->runner_resources_owned) { try { - resources_rc = state->runner->abandon_native_run_resources(state->pipeline_slot); + if (launched && execution_rc == 0) { + STRACE("simpler_run.finalize.retire_resources"); + resources_rc = state->runner->complete_native_run_resources(state->pipeline_slot); + } else if (!launched) { + resources_rc = state->runner->abandon_native_run_resources(state->pipeline_slot); + } } catch (...) { resources_rc = -1; } @@ -915,14 +1148,23 @@ int simpler_finalize_run(DeviceContextHandle ctx, RuntimeHandle runtime) { } if (state->runner_claimed) { - state->runner->release_native_run(state); + { + STRACE("simpler_run.finalize.release_claim"); + state->runner->release_native_run(state); + } state->runner_claimed = false; } if (state->runner_reserved) { - state->runner->release_native_run_reservation(state); + { + STRACE("simpler_run.finalize.release_reservation"); + state->runner->release_native_run_reservation(state); + } state->runner_reserved = false; } - destroy_native_run_state(state); + { + STRACE("simpler_run.finalize.destroy_state"); + destroy_native_run_state(state); + } emit_native_run_host_wall(trace_inv, trace_hid, trace_start_ns, trace_attrs); if (validation_rc != 0) return validation_rc; if (resources_rc != 0) return resources_rc; diff --git a/src/common/platform/onboard/host/device_runner_base.cpp b/src/common/platform/onboard/host/device_runner_base.cpp index a8f9b4c3de..056c3bbd86 100644 --- a/src/common/platform/onboard/host/device_runner_base.cpp +++ b/src/common/platform/onboard/host/device_runner_base.cpp @@ -43,6 +43,7 @@ #include "callable.h" #include "callable_protocol.h" #include "call_config.h" +#include "common/strace.h" #include "chip_callable_layout.h" #include "common/core_type.h" #include "common/host_api.h" @@ -1409,7 +1410,11 @@ int DeviceRunnerBase::sync_run_streams() { return sync_stream_pair(stream_aicpu_ int DeviceRunnerBase::sync_stream_pair(rtStream_t aicpu_stream, rtStream_t aicore_stream) { LOG_INFO("=== aclrtSynchronizeStreamWithTimeout AICPU stream ==="); - int rc = aclrtSynchronizeStreamWithTimeout(aicpu_stream, timeout_config_.stream_sync_timeout_ms); + int rc = 0; + { + STRACE("simpler_run.runner_run.native_fence_wait.aicpu"); + rc = aclrtSynchronizeStreamWithTimeout(aicpu_stream, timeout_config_.stream_sync_timeout_ms); + } if (rc == ACL_ERROR_RT_STREAM_SYNC_TIMEOUT) { LOG_ERROR( "Stream sync timeout: stream=AICPU timeout_ms=%d device_id=%d block_dim=%d", @@ -1425,7 +1430,10 @@ int DeviceRunnerBase::sync_stream_pair(rtStream_t aicpu_stream, rtStream_t aicor } LOG_INFO("=== aclrtSynchronizeStreamWithTimeout AICore stream ==="); - rc = aclrtSynchronizeStreamWithTimeout(aicore_stream, timeout_config_.stream_sync_timeout_ms); + { + STRACE("simpler_run.runner_run.native_fence_wait.aicore"); + rc = aclrtSynchronizeStreamWithTimeout(aicore_stream, timeout_config_.stream_sync_timeout_ms); + } if (rc == ACL_ERROR_RT_STREAM_SYNC_TIMEOUT) { LOG_ERROR( "Stream sync timeout: stream=AICore timeout_ms=%d device_id=%d block_dim=%d", diff --git a/src/common/platform/onboard/host/device_runner_base.h b/src/common/platform/onboard/host/device_runner_base.h index acbf910e2b..d60862d9d6 100644 --- a/src/common/platform/onboard/host/device_runner_base.h +++ b/src/common/platform/onboard/host/device_runner_base.h @@ -542,6 +542,9 @@ class DeviceRunnerBase { /** Provision/abandon platform resources owned by one prepared native run. */ virtual int provision_native_run_resources(uint32_t /*pipeline_slot*/) { return 0; } virtual int abandon_native_run_resources(uint32_t /*pipeline_slot*/) { return 0; } + virtual int complete_native_run_resources(uint32_t pipeline_slot) { + return abandon_native_run_resources(pipeline_slot); + } /** * Execute a Runtime. Each arch implements its own `run()` — the bodies diff --git a/src/common/worker/native_run_launch_signal.h b/src/common/worker/native_run_launch_signal.h index ef17a46348..a3a6de071d 100644 --- a/src/common/worker/native_run_launch_signal.h +++ b/src/common/worker/native_run_launch_signal.h @@ -12,6 +12,7 @@ #ifndef SRC_COMMON_WORKER_NATIVE_RUN_LAUNCH_SIGNAL_H_ #define SRC_COMMON_WORKER_NATIVE_RUN_LAUNCH_SIGNAL_H_ +#include #include #include @@ -37,6 +38,13 @@ class NativeRunLaunchSignal { cv_.notify_one(); } + bool wait_for(std::chrono::microseconds timeout) { + std::unique_lock lock(mutex_); + return cv_.wait_for(lock, timeout, [this]() { + return notified_; + }); + } + private: std::mutex mutex_; std::condition_variable cv_; diff --git a/src/common/worker/native_run_state.h b/src/common/worker/native_run_state.h index 831c4f9e75..c036981bb2 100644 --- a/src/common/worker/native_run_state.h +++ b/src/common/worker/native_run_state.h @@ -44,7 +44,10 @@ struct NativeRunState { trace_hid(trace_hid_in) {} ~NativeRunState() { + executor_cancelled.store(true, std::memory_order_release); + executor_start_signal.notify(); if (executor.joinable()) executor.join(); + if (executor_submitted) executor_task_done_signal.wait(); if (host_thread_state != nullptr) { runner->destroy_native_run_thread_state(host_thread_state); } @@ -69,6 +72,13 @@ struct NativeRunState { std::atomic execution_done{false}; std::atomic phase{NativeRunPhase::Prepared}; NativeRunLaunchSignal launch_signal{}; + NativeRunLaunchSignal completion_signal{}; + NativeRunLaunchSignal executor_ready_signal{}; + NativeRunLaunchSignal executor_start_signal{}; + NativeRunLaunchSignal executor_task_done_signal{}; + std::atomic executor_attach_rc{-1}; + std::atomic executor_cancelled{false}; + bool executor_submitted{false}; void *host_thread_state{nullptr}; uint64_t run_id{0}; uint64_t generation{0}; diff --git a/tests/ut/cpp/hierarchical/test_run_stream_slots.cpp b/tests/ut/cpp/hierarchical/test_run_stream_slots.cpp index 85358178b5..0db2678a3f 100644 --- a/tests/ut/cpp/hierarchical/test_run_stream_slots.cpp +++ b/tests/ut/cpp/hierarchical/test_run_stream_slots.cpp @@ -11,7 +11,10 @@ #include +#include +#include #include +#include #include #include "host/run_stream_slots.h" @@ -69,6 +72,12 @@ RunStreamSlots make_slots(FakeStreams &fake) { ); } +RunStreamSlots::ThreadFactory thread_factory() { + return [](std::function fn) { + return std::thread(std::move(fn)); + }; +} + // The AICPU stream is the slot's for the runner's lifetime; the AICore stream // belongs to one run, so the count advances once per acquire. TEST(RunStreamSlots, EveryAcquireCreatesAnAicoreStreamAndKeepsTheAicpuOne) { @@ -91,6 +100,85 @@ TEST(RunStreamSlots, EveryAcquireCreatesAnAicoreStreamAndKeepsTheAicpuOne) { EXPECT_EQ(slots.created_count(), 2u); } +TEST(RunStreamSlots, AsyncRetireReplenishesTheNextFreshStreamOffTheAcquirePath) { + FakeStreams fake; + RunStreamSlots slots = make_slots(fake); + + ASSERT_EQ(slots.acquire(0), 0); + void *first_aicore = slots.aicore(0); + ASSERT_EQ(slots.retire_aicore_async(0, thread_factory()), 0); + EXPECT_FALSE(slots.ready(0)); + EXPECT_EQ(slots.aicore(0), first_aicore) << "the in-flight destroy retains ownership"; + + ASSERT_EQ(slots.acquire(0), 0) << "acquire joins the pending replenish"; + EXPECT_TRUE(slots.ready(0)); + EXPECT_NE(slots.aicore(0), first_aicore); + EXPECT_EQ(slots.created_count(), 2u); +} + +TEST(RunStreamSlots, AsyncRetireUsesOneResidentSerialWorkerAcrossSlots) { + FakeStreams fake; + RunStreamSlots slots = make_slots(fake); + ASSERT_EQ(slots.acquire(0), 0); + ASSERT_EQ(slots.acquire(1), 0); + + std::atomic workers_created{0}; + std::atomic active_jobs{0}; + std::atomic max_active_jobs{0}; + RunStreamSlots::ThreadFactory factory = [&workers_created](std::function fn) { + workers_created.fetch_add(1); + return std::thread(std::move(fn)); + }; + RunStreamSlots::JobWrapper wrapper = [&active_jobs, &max_active_jobs](std::function fn) { + return [fn = std::move(fn), &active_jobs, &max_active_jobs]() { + int active = active_jobs.fetch_add(1) + 1; + int observed = max_active_jobs.load(); + while (active > observed && !max_active_jobs.compare_exchange_weak(observed, active)) {} + std::this_thread::sleep_for(std::chrono::milliseconds(5)); + fn(); + active_jobs.fetch_sub(1); + }; + }; + + ASSERT_EQ(slots.retire_aicore_async(0, factory, wrapper), 0); + ASSERT_EQ(slots.retire_aicore_async(1, factory, wrapper), 0); + ASSERT_EQ(slots.acquire(0), 0); + ASSERT_EQ(slots.acquire(1), 0); + EXPECT_EQ(workers_created.load(), 1); + EXPECT_EQ(max_active_jobs.load(), 1); +} + +TEST(RunStreamSlots, AsyncReplenishCreateFailureLeavesTheSlotEmpty) { + FakeStreams fake; + RunStreamSlots slots = make_slots(fake); + + ASSERT_EQ(slots.acquire(0), 0); + fake.fail_next_creates(1); + ASSERT_EQ(slots.retire_aicore_async(0, thread_factory()), 0); + + EXPECT_EQ(slots.acquire(0), -7); + EXPECT_EQ(slots.aicore(0), nullptr); + EXPECT_EQ(slots.created_count(), 1u); + EXPECT_EQ(slots.destroy_all(), 0); + EXPECT_EQ(fake.live_count(), 0u); +} + +TEST(RunStreamSlots, AsyncDestroyFailureLocksTheSlotUntilTeardownRetriesIt) { + FakeStreams fake; + RunStreamSlots slots = make_slots(fake); + + ASSERT_EQ(slots.acquire(0), 0); + void *stranded = slots.aicore(0); + fake.fail_next_destroys(1); + ASSERT_EQ(slots.retire_aicore_async(0, thread_factory()), 0); + + EXPECT_EQ(slots.acquire(0), -13); + EXPECT_EQ(slots.aicore(0), stranded); + EXPECT_EQ(slots.created_count(), 1u); + EXPECT_EQ(slots.destroy_all(), 0); + EXPECT_EQ(fake.live_count(), 0u); +} + // The three consequences a failed destroy must have. TEST(RunStreamSlots, AFailedDestroyReportsKeepsTheHandleAndLocksTheSlot) { FakeStreams fake; diff --git a/tests/ut/py/test_worker/test_host_worker.py b/tests/ut/py/test_worker/test_host_worker.py index 93533f5415..92253bd0fe 100644 --- a/tests/ut/py/test_worker/test_host_worker.py +++ b/tests/ut/py/test_worker/test_host_worker.py @@ -263,6 +263,7 @@ def __init__(self, *, supports_concurrent_native_prepare: bool = False) -> None: self.finalized = [threading.Event(), threading.Event()] self.launch_errors: dict[tuple[int, int], BaseException] = {} self.prepare_errors: dict[tuple[int, int], BaseException] = {} + self.prepare_gates: dict[int, tuple[threading.Event, threading.Event]] = {} self.poll_errors: dict[tuple[int, int], BaseException] = {} self.finalize_errors: dict[tuple[int, int], BaseException] = {} self.prepare_identities: list[tuple[int, int, int, int]] = [] @@ -293,6 +294,11 @@ def _prepare_native_run_from_blob( self.prepare_identities.append((slot, int(generation), int(_run_id), int(_dispatch_id))) token = SimpleNamespace(slot_id=slot, generation=int(generation), run_epoch=slot + 1) self.events.append(("prepare", slot)) + gate = self.prepare_gates.get(slot) + if gate is not None: + entered, release = gate + entered.set() + assert release.wait(5.0) self.prepared[slot].set() return token @@ -324,6 +330,11 @@ def _poll_native_run(self, token) -> bool: self.events.append(("poll", slot)) return self.completed[slot].is_set() + def _wait_native_run(self, token) -> None: + slot = int(token.slot_id) + self.events.append(("wait", slot)) + assert self.completed[slot].wait(5.0) + def _finalize_native_run(self, token) -> None: slot = int(token.slot_id) run_key = (slot, int(token.generation)) @@ -556,6 +567,8 @@ def test_two_frame_hbg_prepares_b_while_a_runs_but_accepts_only_after_launch(): 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) + assert ("wait", 0) in harness.cw._impl.events + assert ("poll", 0) not in harness.cw._impl.events lifecycle = [event[:2] for event in harness.cw._impl.events if event[0] in {"prepare", "launch", "finalize"}] assert lifecycle == [ @@ -570,6 +583,48 @@ def test_two_frame_hbg_prepares_b_while_a_runs_but_accepts_only_after_launch(): harness.close() +def test_two_frame_hbg_finalizes_active_while_successor_prepare_is_blocked(): + harness = _TwoFrameLoopHarness( + supports_concurrent_native_prepare=True, + chip_runtime="host_build_graph", + ) + prepare_entered = threading.Event() + release_prepare = threading.Event() + harness.cw._impl.prepare_gates[1] = (prepare_entered, release_prepare) + 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) + assert prepare_entered.wait(5.0) + harness.cw._impl.completed[0].set() + assert harness.cw._impl.finalized[0].wait(5.0) + assert not release_prepare.is_set() + assert not harness.cw._impl.launched[1].is_set() + + release_prepare.set() + 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.completed[1].set() + harness.wait_state(1, worker_mod._TASK_DONE) + + lifecycle = [event[:2] for event in harness.cw._impl.events if event[0] in {"prepare", "launch", "finalize"}] + expected_lifecycle = [ + ("prepare", 0), + ("launch", 0), + ("prepare", 1), + ("finalize", 0), + ("launch", 1), + ("finalize", 1), + ] + assert lifecycle == expected_lifecycle + finally: + release_prepare.set() + harness.close() + + def test_two_frame_hbg_publishes_failure_instead_of_staged_when_prepare_fails(): harness = _TwoFrameLoopHarness( supports_concurrent_native_prepare=True, @@ -599,7 +654,7 @@ def test_two_frame_hbg_waits_for_first_token_to_launch_before_preparing_second() assert harness.cw._impl.launched[0].wait(5.0) harness.wait_state(1, worker_mod._FRAME_STAGED) - assert harness.cw._impl.prepared[1].is_set() + assert harness.cw._impl.prepared[1].wait(5.0) assert not harness.cw._impl.finalized[0].is_set() assert [event[:2] for event in harness.cw._impl.events if event[0] in {"prepare", "launch"}] == [ ("prepare", 0), @@ -659,7 +714,7 @@ def test_two_frame_hbg_does_not_prepare_high_dispatch_successor_before_active_fr assert harness.cw._impl.launched[1].wait(5.0) harness.wait_state(0, worker_mod._FRAME_STAGED) - assert harness.cw._impl.prepared[0].is_set() + assert harness.cw._impl.prepared[0].wait(5.0) assert not harness.cw._impl.launched[0].is_set() assert [event[:2] for event in harness.cw._impl.events if event[0] in {"prepare", "launch"}] == [ ("prepare", 1), @@ -991,7 +1046,7 @@ def test_two_frame_shutdown_finalizes_backend_prepared_successor_once(): assert harness.cw._impl.prepared[1].is_set() _mailbox_store_i32(harness.mailbox_addr + _OFF_STATE, worker_mod._SHUTDOWN) - harness.thread.join(5.0) + harness.thread.join(10.0) assert not harness.thread.is_alive() assert _mailbox_load_i32(harness.state_addr(1)) == worker_mod._TASK_FAILED assert not harness.cw._impl.launched[1].is_set()