diff --git a/docs/dfx/dep-gen.md b/docs/dfx/dep-gen.md index 06d1153def..58f1cc4d68 100644 --- a/docs/dfx/dep-gen.md +++ b/docs/dfx/dep-gen.md @@ -91,8 +91,8 @@ nothing to capture-then-reconstruct. (`dep_gen_host_graph_active()` tells the runner). Nothing is dropped under back-pressure because nothing is streamed. - **Output.** The same `deps.json`, written during the device-runner drain. - After prepare's host orchestration builds the graph, the phased runtime moves - it into run-owned storage and the executor adopts it before enqueue. + The graph is thread-local, and prepare's host orchestration and the drain that + emits it both run on the child progress loop's single thread. --- @@ -395,7 +395,7 @@ list; only the dep_gen replay graph loses the tail. | Capture call site (device-orch) | `src/{a2a3,a5}/runtime/tensormap_and_ringbuffer/runtime/pto_orchestrator.cpp` `submit_task_common` | One conditional block that snapshots inputs into the ring when `is_dep_gen_enabled()`; fires for both `submit_task` and `submit_dummy_task`. The schema carries `kernel_ids[3] = {aic, aiv0, aiv1}` so the swimlane post-processor can resolve `task_id → kernel` from `deps.json` at level=1 where the AICore record is the sole device-side identity source. Inactive subslots stay at `INVALID_KERNEL_ID = -1`. It also carries the SPMD logical block num (`block_num` on a2a3, `core_num` on a5's launch spec) as `tasks[].block_num`. | | Replay | `src/{a2a3,a5}/runtime/tensormap_and_ringbuffer/host/dep_gen_replay.{h,cpp}` | Pure CPU; runs dual-pass differential replay — `compute_task_fanin` (oracle) + inlined STEP A/B mirror (annotated) against two `PTO2TensorMap` instances. Emits `deps.json` when both passes agree per record. Platform-agnostic — a5 reuses the a2a3 source verbatim. | | Host-direct capture (host-orch) | `src/a2a3/runtime/host_build_graph/runtime/dep_gen_host_graph.h`, `src/a2a3/runtime/host_build_graph/host/dep_gen_host_graph.cpp` | Task / tensor / edge tables filled from `submit_task_common` + `compute_task_fanin`'s `Annotate` hooks (`src/a2a3/runtime/host_build_graph/runtime/pto_dep_compute.h`), reset per orchestration by `run_host_orchestration`, serialized by the same `deps.json` writer. The runtime translation unit carries weak no-op fallbacks so the AICPU build links without it. | -| Device-runner hookup | `src/{a2a3,a5}/platform/{onboard,sim}/host/device_runner.cpp` | `dep_gen_host_graph_active()` picks the shape: host-orch moves the prepare thread's completed capture into run-owned storage, adopts it on the executor, calls `dep_gen_host_graph_emit(deps_path)` at teardown, and skips collector init/start/reconcile entirely; device-orch calls `dep_gen_replay_emit_deps_json(records.data(), records.size(), deps_path)` post-`reconcile_counters`. The c_api latches the CallConfig before the bind so host capture is armed before the orchestration it records. | +| Device-runner hookup | `src/{a2a3,a5}/platform/{onboard,sim}/host/device_runner.cpp` | `dep_gen_host_graph_active()` picks the shape: host-orch calls `dep_gen_host_graph_emit(deps_path)` at teardown — reading the thread-local graph its own orchestration built on the same child progress thread — and skips collector init/start/reconcile entirely; device-orch calls `dep_gen_replay_emit_deps_json(records.data(), records.size(), deps_path)` post-`reconcile_counters`. The c_api latches the CallConfig before the bind so host capture is armed before the orchestration it records. | | Viewer | `simpler_setup/tools/deps_viewer.py` | `deps.json` → text (default) or pan/zoom HTML | | Test | `tests/st/{a2a3,a5}/tensormap_and_ringbuffer/dfx/dep_gen/test_dep_gen.py` + `test_dep_gen_chain.py`, `tests/st/a2a3/host_build_graph/dfx/dep_gen/test_dep_gen.py` | Smoke test + 6-edge validation against `vector_example` orchestration (both platforms share byte-identical orchestration code). The host_build_graph case runs the *same* orchestration through host-direct capture and asserts the same 6 edges, so a divergence between the two shapes fails a test. | diff --git a/src/a2a3/platform/onboard/host/device_runner.cpp b/src/a2a3/platform/onboard/host/device_runner.cpp index bea83c91be..c3e52b591a 100644 --- a/src/a2a3/platform/onboard/host/device_runner.cpp +++ b/src/a2a3/platform/onboard/host/device_runner.cpp @@ -717,8 +717,8 @@ int DeviceRunner::reap_run(unsigned slot) { // order (mgmt's final-drain pass into L2 has poll as its consumer). 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 + // a2a3-only dep_gen teardown: host-orch emits the graph its orchestration + // built on this same thread; device-orch stops the collector, reconciles the // ring, and replays the records. if (enable_dep_gen_) { const std::string deps = make_deps_json_path(output_prefix_); diff --git a/src/a2a3/platform/sim/host/device_runner.cpp b/src/a2a3/platform/sim/host/device_runner.cpp index 01925fcecf..493955e223 100644 --- a/src/a2a3/platform/sim/host/device_runner.cpp +++ b/src/a2a3/platform/sim/host/device_runner.cpp @@ -671,7 +671,7 @@ int DeviceRunner::drain_execution(ActiveExecution &) { pmu_collector_.reconcile_counters(); } - // Host-orch emits the graph snapshot adopted from the prepare thread; + // Host-orch emits the graph its orchestration built on this same thread; // device-orch stops the collector, reconciles the ring, and replays. if (enable_dep_gen_) { const std::string deps = make_deps_json_path(output_prefix_); diff --git a/src/a2a3/runtime/host_build_graph/host/dep_gen_host_graph.cpp b/src/a2a3/runtime/host_build_graph/host/dep_gen_host_graph.cpp index 47daaf1454..a11f05f3d5 100644 --- a/src/a2a3/runtime/host_build_graph/host/dep_gen_host_graph.cpp +++ b/src/a2a3/runtime/host_build_graph/host/dep_gen_host_graph.cpp @@ -195,7 +195,7 @@ void fill_producer(EdgeAnnot &e, const PTO2TensorMapEntry &entry) { } // --------------------------------------------------------------------------- -// Capture state — thread-local while built, then moved with its native run +// Capture state — thread-local, built and emitted on the same thread // --------------------------------------------------------------------------- struct HostGraphState { @@ -495,25 +495,6 @@ extern "C" void dep_gen_host_graph_set_enabled(bool enable) { state().enabled = extern "C" bool dep_gen_host_graph_active() { return true; } -extern "C" void *dep_gen_host_graph_take_capture() { - HostGraphState ¤t = state(); - if (!current.enabled) return nullptr; - auto *capture = new HostGraphState(std::move(current)); - current = HostGraphState{}; - return capture; -} - -extern "C" void dep_gen_host_graph_adopt_capture(void *capture) noexcept { - if (capture == nullptr) return; - auto *captured_state = static_cast(capture); - state() = std::move(*captured_state); - delete captured_state; -} - -extern "C" void dep_gen_host_graph_destroy_capture(void *capture) noexcept { - delete static_cast(capture); -} - extern "C" int dep_gen_host_graph_emit(const char *deps_json_path) { if (deps_json_path == nullptr) { LOG_ERROR("dep_gen host graph: null deps_json_path"); @@ -523,11 +504,9 @@ extern "C" int dep_gen_host_graph_emit(const char *deps_json_path) { if (!s.captured) { // An empty graph here is not "the orchestration submitted nothing" — // begin_task() would have set captured even for a graph of one task. - // It means capture was never armed or the run-owned snapshot was not - // adopted onto this progress thread before teardown. - LOG_ERROR( - "dep_gen host graph: no capture was adopted on this thread — deps.json not written to %s", deps_json_path - ); + // The graph is thread-local, so it means either capture was never armed + // or this run's orchestration ran on a different thread than this emit. + LOG_ERROR("dep_gen host graph: no capture on this thread — deps.json not written to %s", deps_json_path); return -3; } if (!write_deps_json(deps_json_path, s.tasks, s.tensors, s.edges)) { diff --git a/src/a2a3/runtime/host_build_graph/runtime/dep_gen_host_graph.h b/src/a2a3/runtime/host_build_graph/runtime/dep_gen_host_graph.h index d59ac7ff62..8212f217d4 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/dep_gen_host_graph.h +++ b/src/a2a3/runtime/host_build_graph/runtime/dep_gen_host_graph.h @@ -31,17 +31,17 @@ * end_task() — closes the task, after its last dependency step * * Control surface, called from the device runner (same host_runtime.so): - * set_enabled() / active() / take_capture() / adopt_capture() / emit() + * set_enabled() / active() / emit() * * The runtime translation unit links weak no-op fallbacks (pto_orchestrator.cpp) * so the AICPU build, which has no host graph, resolves without this .cpp. * - * The graph is per-thread state while it is being built. After bind, prepare - * moves the completed graph into run-owned storage; launch adopts that snapshot - * into the progress thread's local state before enqueue, and drain emits it. - * This keeps capture lock-free while allowing serialized lifecycle calls to - * use different host threads and preventing two prepared contexts on one - * thread from overwriting one another. + * The graph lives in thread-local state, so capture is lock-free and two + * prepared contexts on different threads cannot overwrite one another. Emit + * reads the calling thread's state, so a run's orchestration (which builds the + * graph) and its drain (which emits it) must land on the same thread. The child + * progress loop satisfies this by being single-threaded; emit returns -3 if the + * invariant is ever broken. * * Per-task producer dedup mirrors PTO2FaninBuilder, which keys on (ring, slot); * this keys on producer task id. The two agree only because host_build_graph is @@ -124,15 +124,6 @@ void dep_gen_host_graph_set_enabled(bool enable); */ bool dep_gen_host_graph_active(); -/** Move the current thread's capture into an opaque, caller-owned snapshot. */ -void *dep_gen_host_graph_take_capture(); - -/** Adopt and consume a snapshot on the current execution thread. */ -void dep_gen_host_graph_adopt_capture(void *capture) noexcept; - -/** Destroy a snapshot that will not be launched. */ -void dep_gen_host_graph_destroy_capture(void *capture) noexcept; - /** * Write the captured graph to `deps_json_path`. Returns 0 on success, non-zero * if capture was off/empty or the file could not be written. diff --git a/src/a5/runtime/host_build_graph/host/dep_gen_host_graph.cpp b/src/a5/runtime/host_build_graph/host/dep_gen_host_graph.cpp index 47daaf1454..a11f05f3d5 100644 --- a/src/a5/runtime/host_build_graph/host/dep_gen_host_graph.cpp +++ b/src/a5/runtime/host_build_graph/host/dep_gen_host_graph.cpp @@ -195,7 +195,7 @@ void fill_producer(EdgeAnnot &e, const PTO2TensorMapEntry &entry) { } // --------------------------------------------------------------------------- -// Capture state — thread-local while built, then moved with its native run +// Capture state — thread-local, built and emitted on the same thread // --------------------------------------------------------------------------- struct HostGraphState { @@ -495,25 +495,6 @@ extern "C" void dep_gen_host_graph_set_enabled(bool enable) { state().enabled = extern "C" bool dep_gen_host_graph_active() { return true; } -extern "C" void *dep_gen_host_graph_take_capture() { - HostGraphState ¤t = state(); - if (!current.enabled) return nullptr; - auto *capture = new HostGraphState(std::move(current)); - current = HostGraphState{}; - return capture; -} - -extern "C" void dep_gen_host_graph_adopt_capture(void *capture) noexcept { - if (capture == nullptr) return; - auto *captured_state = static_cast(capture); - state() = std::move(*captured_state); - delete captured_state; -} - -extern "C" void dep_gen_host_graph_destroy_capture(void *capture) noexcept { - delete static_cast(capture); -} - extern "C" int dep_gen_host_graph_emit(const char *deps_json_path) { if (deps_json_path == nullptr) { LOG_ERROR("dep_gen host graph: null deps_json_path"); @@ -523,11 +504,9 @@ extern "C" int dep_gen_host_graph_emit(const char *deps_json_path) { if (!s.captured) { // An empty graph here is not "the orchestration submitted nothing" — // begin_task() would have set captured even for a graph of one task. - // It means capture was never armed or the run-owned snapshot was not - // adopted onto this progress thread before teardown. - LOG_ERROR( - "dep_gen host graph: no capture was adopted on this thread — deps.json not written to %s", deps_json_path - ); + // The graph is thread-local, so it means either capture was never armed + // or this run's orchestration ran on a different thread than this emit. + LOG_ERROR("dep_gen host graph: no capture on this thread — deps.json not written to %s", deps_json_path); return -3; } if (!write_deps_json(deps_json_path, s.tasks, s.tensors, s.edges)) { diff --git a/src/a5/runtime/host_build_graph/runtime/dep_gen_host_graph.h b/src/a5/runtime/host_build_graph/runtime/dep_gen_host_graph.h index 9f9300a568..3f86de4bd8 100644 --- a/src/a5/runtime/host_build_graph/runtime/dep_gen_host_graph.h +++ b/src/a5/runtime/host_build_graph/runtime/dep_gen_host_graph.h @@ -31,17 +31,17 @@ * end_task() — closes the task, after its last dependency step * * Control surface, called from the device runner (same host_runtime.so): - * set_enabled() / active() / take_capture() / adopt_capture() / emit() + * set_enabled() / active() / emit() * * The runtime translation unit links weak no-op fallbacks (pto_orchestrator.cpp) * so the AICPU build, which has no host graph, resolves without this .cpp. * - * The graph is per-thread state while it is being built. After bind, prepare - * moves the completed graph into run-owned storage; launch adopts that snapshot - * into the progress thread's local state before enqueue, and drain emits it. - * This keeps capture lock-free while allowing serialized lifecycle calls to - * use different host threads and preventing two prepared contexts on one - * thread from overwriting one another. + * The graph lives in thread-local state, so capture is lock-free and two + * prepared contexts on different threads cannot overwrite one another. Emit + * reads the calling thread's state, so a run's orchestration (which builds the + * graph) and its drain (which emits it) must land on the same thread. The child + * progress loop satisfies this by being single-threaded; emit returns -3 if the + * invariant is ever broken. * * Per-task producer dedup mirrors PTO2FaninBuilder, which keys on (ring, slot); * this keys on producer task id. The two agree only because host_build_graph is @@ -123,15 +123,6 @@ void dep_gen_host_graph_set_enabled(bool enable); */ bool dep_gen_host_graph_active(); -/** Move the current thread's capture into an opaque, caller-owned snapshot. */ -void *dep_gen_host_graph_take_capture(); - -/** Adopt and consume a snapshot on the current execution thread. */ -void dep_gen_host_graph_adopt_capture(void *capture) noexcept; - -/** Destroy a snapshot that will not be launched. */ -void dep_gen_host_graph_destroy_capture(void *capture) noexcept; - /** * Write the captured graph to `deps_json_path`. Returns 0 on success, non-zero * if capture was off/empty or the file could not be written. diff --git a/src/common/platform/onboard/host/c_api_shared.cpp b/src/common/platform/onboard/host/c_api_shared.cpp index 1924b60c0b..10543db63e 100644 --- a/src/common/platform/onboard/host/c_api_shared.cpp +++ b/src/common/platform/onboard/host/c_api_shared.cpp @@ -816,12 +816,24 @@ int simpler_wait_run(DeviceContextHandle ctx, RuntimeHandle runtime) { NativeRunPhase phase = state->phase.load(std::memory_order_acquire); if (phase == NativeRunPhase::Prepared) return -1; if (phase == NativeRunPhase::Complete) return state->completion_rc; + // drain_execution() synchronizes and destroys streams, reads device memory + // and frees device allocations, all of which need this thread's CANN + // device context. rtSetDevice is idempotent on an already-attached thread. int drain_rc = -1; try { - if (state->active_execution != nullptr) { - drain_rc = state->runner->drain_execution(*state->active_execution); + drain_rc = state->runner->attach_current_thread(state->runner->device_id()); + if (drain_rc != 0) { + LOG_ERROR("simpler_wait_run: attach_current_thread failed: %d (%s)", drain_rc, state->trace_attrs); + } else { + drain_rc = -1; + if (state->active_execution != nullptr) { + drain_rc = state->runner->drain_execution(*state->active_execution); + } } - } catch (...) {} + } catch (...) { + drain_rc = -1; + LOG_ERROR("simpler_wait_run: drain threw (%s)", state->trace_attrs); + } 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); @@ -842,11 +854,28 @@ int simpler_finalize_run(DeviceContextHandle ctx, RuntimeHandle runtime) { int execution_rc = state->completion_rc; const bool launched = state->active_execution != nullptr; + // Both drain_execution() and validate_runtime_impl() touch the device, so + // the attach covers each of them. rtSetDevice is idempotent on an + // already-attached thread. + int attach_rc = -1; + try { + attach_rc = state->runner->attach_current_thread(state->runner->device_id()); + } catch (...) { + attach_rc = -1; + } + if (attach_rc != 0) { + LOG_ERROR("simpler_finalize_run: attach_current_thread failed: %d (%s)", attach_rc, state->trace_attrs); + } if (phase == NativeRunPhase::Running && launched) { - int drain_rc = -1; - try { - drain_rc = state->runner->drain_execution(*state->active_execution); - } catch (...) {} + int drain_rc = attach_rc; + if (attach_rc == 0) { + drain_rc = -1; + try { + drain_rc = state->runner->drain_execution(*state->active_execution); + } catch (...) { + LOG_ERROR("simpler_finalize_run: drain_execution threw (%s)", state->trace_attrs); + } + } if (execution_rc == 0) execution_rc = drain_rc; state->completion_rc = execution_rc; state->phase.store(NativeRunPhase::Complete, std::memory_order_release); @@ -856,7 +885,6 @@ int simpler_finalize_run(DeviceContextHandle ctx, RuntimeHandle runtime) { 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()); if (attach_rc == 0) { { STRACE("simpler_run.validate"); diff --git a/src/common/platform/sim/host/c_api_shared.cpp b/src/common/platform/sim/host/c_api_shared.cpp index b429278abb..30a6491acd 100644 --- a/src/common/platform/sim/host/c_api_shared.cpp +++ b/src/common/platform/sim/host/c_api_shared.cpp @@ -723,12 +723,25 @@ int simpler_wait_run(DeviceContextHandle ctx, RuntimeHandle runtime) { NativeRunPhase phase = state->phase.load(std::memory_order_acquire); if (phase == NativeRunPhase::Prepared) return -1; if (phase == NativeRunPhase::Complete) return state->completion_rc; + // Bind the calling thread to this runner's simulated device before the + // drain, so sim's lifecycle entry points carry the same contract as + // onboard's. attach_current_thread() is idempotent for a thread already + // bound to this device. int drain_rc = -1; try { - if (state->active_execution != nullptr) { - drain_rc = state->runner->drain_execution(*state->active_execution); + drain_rc = state->runner->attach_current_thread(state->runner->device_id()); + if (drain_rc != 0) { + LOG_ERROR("simpler_wait_run: attach_current_thread failed: %d (%s)", drain_rc, state->trace_attrs); + } else { + drain_rc = -1; + if (state->active_execution != nullptr) { + drain_rc = state->runner->drain_execution(*state->active_execution); + } } - } catch (...) {} + } catch (...) { + drain_rc = -1; + LOG_ERROR("simpler_wait_run: drain threw (%s)", state->trace_attrs); + } 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); @@ -747,11 +760,28 @@ int simpler_finalize_run(DeviceContextHandle ctx, RuntimeHandle runtime) { int execution_rc = state->completion_rc; const bool launched = state->active_execution != nullptr; + // Bind the calling thread to this runner's simulated device before the + // drain and the validation below. attach_current_thread() is idempotent + // for a thread already bound to this device. + int attach_rc = -1; + try { + attach_rc = state->runner->attach_current_thread(state->runner->device_id()); + } catch (...) { + attach_rc = -1; + } + if (attach_rc != 0) { + LOG_ERROR("simpler_finalize_run: attach_current_thread failed: %d (%s)", attach_rc, state->trace_attrs); + } if (phase == NativeRunPhase::Running && launched) { - int drain_rc = -1; - try { - drain_rc = state->runner->drain_execution(*state->active_execution); - } catch (...) {} + int drain_rc = attach_rc; + if (attach_rc == 0) { + drain_rc = -1; + try { + drain_rc = state->runner->drain_execution(*state->active_execution); + } catch (...) { + LOG_ERROR("simpler_finalize_run: drain_execution threw (%s)", state->trace_attrs); + } + } if (execution_rc == 0) execution_rc = drain_rc; state->completion_rc = execution_rc; state->phase.store(NativeRunPhase::Complete, std::memory_order_release); @@ -761,7 +791,6 @@ int simpler_finalize_run(DeviceContextHandle ctx, RuntimeHandle runtime) { 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()); if (attach_rc == 0) { { STRACE("simpler_run.validate"); diff --git a/tests/ut/cpp/CMakeLists.txt b/tests/ut/cpp/CMakeLists.txt index e4ab8abd20..8d31ed6733 100644 --- a/tests/ut/cpp/CMakeLists.txt +++ b/tests/ut/cpp/CMakeLists.txt @@ -416,6 +416,29 @@ target_include_directories(test_native_run_execution PRIVATE target_link_libraries(test_native_run_execution PRIVATE ${GTEST_MAIN_LIB} ${GTEST_LIB} pthread) add_test(NAME test_native_run_execution COMMAND test_native_run_execution) set_tests_properties(test_native_run_execution PROPERTIES LABELS "no_hardware") + +# NativeRunContext::publish_acceptance is the only path that marks a task +# accepted, and it is keyed on the launch receipt's identity. +add_executable(test_native_run_acceptance + common/test_native_run_acceptance.cpp + ${CMAKE_SOURCE_DIR}/stubs/test_stubs.cpp + ${CMAKE_SOURCE_DIR}/../../../src/a2a3/runtime/host_build_graph/runtime/shared/runtime.cpp +) +target_include_directories(test_native_run_acceptance PRIVATE + ${GTEST_INCLUDE_DIRS} + ${CMAKE_SOURCE_DIR}/../../../src/common/worker + ${CMAKE_SOURCE_DIR}/../../../src/common/task_interface + ${CMAKE_SOURCE_DIR}/../../../src/common/platform/include + ${CMAKE_SOURCE_DIR}/../../../src/common/log/include + ${CMAKE_SOURCE_DIR}/../../../src/common + ${CMAKE_SOURCE_DIR}/../../../src/a2a3/runtime/host_build_graph/runtime + ${CMAKE_SOURCE_DIR}/../../../src/a2a3/runtime/host_build_graph/common + ${CMAKE_SOURCE_DIR}/../../../src/a2a3/platform/include + ${CMAKE_SOURCE_DIR}/../../../src/common/platform/sim/aicpu +) +target_link_libraries(test_native_run_acceptance PRIVATE ${GTEST_MAIN_LIB} ${GTEST_LIB} pthread) +add_test(NAME test_native_run_acceptance COMMAND test_native_run_acceptance) +set_tests_properties(test_native_run_acceptance PROPERTIES LABELS "no_hardware") # A HostApi value keeps one runner and one run's resource selection even when # callers invoke it from different host threads. add_common_utils_test(test_host_api common/test_host_api.cpp) diff --git a/tests/ut/cpp/a2a3/test_dep_gen_host_graph.cpp b/tests/ut/cpp/a2a3/test_dep_gen_host_graph.cpp index b97dd331bb..7d0f53ad0b 100644 --- a/tests/ut/cpp/a2a3/test_dep_gen_host_graph.cpp +++ b/tests/ut/cpp/a2a3/test_dep_gen_host_graph.cpp @@ -42,26 +42,15 @@ std::string read_file(const std::filesystem::path &path) { } // namespace -TEST(DepGenHostGraphTest, CaptureMovesFromPrepareThreadToExecutorThread) { - const std::filesystem::path path = output_path("handoff"); +TEST(DepGenHostGraphTest, EmitWritesTheGraphCapturedOnTheSameThread) { + const std::filesystem::path path = output_path("same_thread"); std::filesystem::remove(path); dep_gen_host_graph_set_enabled(true); dep_gen_host_graph_begin_capture(); capture_task(11); capture_task(12, 11); - void *capture = dep_gen_host_graph_take_capture(); - ASSERT_NE(capture, nullptr); - EXPECT_EQ(dep_gen_host_graph_emit(path.c_str()), -3); - - int emit_rc = -1; - std::thread executor([&]() { - dep_gen_host_graph_adopt_capture(capture); - emit_rc = dep_gen_host_graph_emit(path.c_str()); - }); - executor.join(); - - ASSERT_EQ(emit_rc, 0); + ASSERT_EQ(dep_gen_host_graph_emit(path.c_str()), 0); const std::string json = read_file(path); EXPECT_NE(json.find("\"task_id\":\"11\""), std::string::npos); EXPECT_NE(json.find("\"task_id\":\"12\""), std::string::npos); @@ -69,20 +58,35 @@ TEST(DepGenHostGraphTest, CaptureMovesFromPrepareThreadToExecutorThread) { std::filesystem::remove(path); } -TEST(DepGenHostGraphTest, DestroyedCaptureDoesNotContaminateNextRun) { - const std::filesystem::path path = output_path("abandoned"); +TEST(DepGenHostGraphTest, EmitOnAnotherThreadWritesNothingAndReports) { + const std::filesystem::path path = output_path("cross_thread"); + std::filesystem::remove(path); + dep_gen_host_graph_set_enabled(true); + dep_gen_host_graph_begin_capture(); + capture_task(21); + + // The graph is thread-local: a run whose orchestration and drain land on + // different threads emits nothing rather than a partial deps.json. + int emit_rc = 0; + std::thread other([&]() { + emit_rc = dep_gen_host_graph_emit(path.c_str()); + }); + other.join(); + + EXPECT_EQ(emit_rc, -3); + EXPECT_FALSE(std::filesystem::exists(path)); +} + +TEST(DepGenHostGraphTest, BeginCaptureClearsThePreviousRunsGraph) { + const std::filesystem::path path = output_path("reset"); std::filesystem::remove(path); dep_gen_host_graph_set_enabled(true); dep_gen_host_graph_begin_capture(); capture_task(101); - dep_gen_host_graph_destroy_capture(dep_gen_host_graph_take_capture()); dep_gen_host_graph_set_enabled(true); dep_gen_host_graph_begin_capture(); capture_task(202); - void *capture = dep_gen_host_graph_take_capture(); - ASSERT_NE(capture, nullptr); - dep_gen_host_graph_adopt_capture(capture); ASSERT_EQ(dep_gen_host_graph_emit(path.c_str()), 0); const std::string json = read_file(path); diff --git a/tests/ut/cpp/common/test_native_run_acceptance.cpp b/tests/ut/cpp/common/test_native_run_acceptance.cpp new file mode 100644 index 0000000000..ea4f96b121 --- /dev/null +++ b/tests/ut/cpp/common/test_native_run_acceptance.cpp @@ -0,0 +1,90 @@ +/* + * Copyright (c) PyPTO Contributors. + * This program is free software, you can redistribute it and/or modify it under the terms and conditions of + * CANN Open Software License Agreement Version 2.0 (the "License"). + * Please refer to the License for details. You may not use this file except in compliance with the License. + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, + * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. + * See LICENSE in the root of the software repository for the full text of the License. + * ----------------------------------------------------------------------------------------------------------- + */ + +#include + +#include + +#include "native_run_context.h" +#include "native_run_execution_test_peer.h" + +namespace { + +// NativeRunContext is a template over the runner only for its execution types; +// publish_acceptance touches neither, so the smallest runner that names them is +// enough to exercise the acceptance gate. +struct FakeRunner { + struct PreparedExecution {}; + struct ActiveExecution {}; +}; + +using FakeRunContext = NativeRunContext; + +constexpr uint32_t kPipelineSlot = 1; +constexpr int32_t kAcceptedValue = 17; + +NativeRunDescriptor make_descriptor(volatile int32_t *accepted_state) { + NativeRunDescriptor descriptor{}; + descriptor.run_epoch = 11; + descriptor.generation = 13; + descriptor.dispatch_id = 17; + descriptor.pipeline_slot = kPipelineSlot; + descriptor.accepted_state = accepted_state; + descriptor.accepted_value = kAcceptedValue; + return descriptor; +} + +LaunchReceipt complete_receipt(const NativeRunIdentity &identity) { + LaunchTransactionResult launched = exact_launch_transaction( + identity, NativeRunExecutionTestPeer::mint(identity), + []() { + return 0; + }, + []() { + return 0; + } + ); + EXPECT_EQ(launched.progress, LaunchProgress::Complete); + return std::move(launched.receipt); +} + +} // namespace + +TEST(NativeRunAcceptanceTest, MatchingReceiptStoresTheAcceptedValue) { + volatile int32_t accepted_state = 0; + FakeRunner runner; + CallConfig config{}; + FakeRunContext context(&runner, config, 0, make_descriptor(&accepted_state), nullptr); + + EXPECT_TRUE(context.publish_acceptance(complete_receipt(context.identity()))); + EXPECT_EQ(__atomic_load_n(&accepted_state, __ATOMIC_ACQUIRE), kAcceptedValue); +} + +TEST(NativeRunAcceptanceTest, StaleReceiptCannotPublishAcceptance) { + volatile int32_t accepted_state = 5; + FakeRunner runner; + CallConfig config{}; + FakeRunContext context(&runner, config, 0, make_descriptor(&accepted_state), nullptr); + + NativeRunIdentity stale = context.identity(); + stale.generation++; + + EXPECT_FALSE(context.publish_acceptance(complete_receipt(stale))); + EXPECT_EQ(__atomic_load_n(&accepted_state, __ATOMIC_ACQUIRE), 5); +} + +TEST(NativeRunAcceptanceTest, AbsentAcceptedStateIsAccepted) { + FakeRunner runner; + CallConfig config{}; + FakeRunContext context(&runner, config, 0, make_descriptor(nullptr), nullptr); + + EXPECT_TRUE(context.publish_acceptance(complete_receipt(context.identity()))); +}