From b2a205d711dd33b2cccd5f75bda4aaf2cbddee9e Mon Sep 17 00:00:00 2001 From: zm Date: Tue, 11 Aug 2026 02:05:31 -0700 Subject: [PATCH] Fix: harden host-built Graph execution failures - Validate Graph wire images and affine replay patch bounds - Separate internal node resolution from host task completion - Surface queue overflow and invalid completion via scheduler shutdown - Pin fatal fanin and no-partial-upload invariants on A2/A3 and A5 --- docs/troubleshooting/device-error-codes.md | 15 +- .../host_build_graph/host/runtime_maker.cpp | 7 +- .../runtime/graph_execution.h | 15 +- .../host_build_graph/runtime/pto_async_wait.h | 15 +- .../runtime/scheduler/graph_execution.cpp | 4 + .../runtime/scheduler/pto_scheduler.h | 54 +++++- .../runtime/scheduler/scheduler_cold_path.cpp | 11 +- .../runtime/scheduler/scheduler_context.h | 7 +- .../runtime/scheduler/scheduler_dispatch.cpp | 61 +++---- .../host_build_graph/host/runtime_maker.cpp | 7 +- .../runtime/graph_execution.h | 15 +- .../host_build_graph/runtime/pto_async_wait.h | 15 +- .../runtime/scheduler/graph_execution.cpp | 4 + .../runtime/scheduler/pto_scheduler.h | 54 +++++- .../runtime/scheduler/scheduler_cold_path.cpp | 11 +- .../runtime/scheduler/scheduler_context.h | 7 +- .../runtime/scheduler/scheduler_dispatch.cpp | 61 +++---- src/common/runtime_status/error_names.h | 13 ++ tests/ut/cpp/CMakeLists.txt | 21 +++ tests/ut/cpp/a2a3/test_graph_cache.cpp | 154 ++++++++++++++++++ tests/ut/cpp/a5/test_graph_cache.cpp | 154 ++++++++++++++++++ .../common/test_hbg_graph_submit_failure.cpp | 98 +++++++++++ 22 files changed, 669 insertions(+), 134 deletions(-) create mode 100644 tests/ut/cpp/common/test_hbg_graph_submit_failure.cpp diff --git a/docs/troubleshooting/device-error-codes.md b/docs/troubleshooting/device-error-codes.md index eb93f66c48..c631aea56a 100644 --- a/docs/troubleshooting/device-error-codes.md +++ b/docs/troubleshooting/device-error-codes.md @@ -29,7 +29,9 @@ grep -E "orch_error_code=|sched_error_code=|sub_class=|error detail:" ## How an error reaches you -The `tensormap_and_ringbuffer` runtime runs orchestration and scheduling on the +The `host_build_graph` runtime runs orchestration on the host, transfers the +prepared image to the AICPU, and runs scheduling there. The +`tensormap_and_ringbuffer` runtime runs both orchestration and scheduling on the AICPU. On a fatal condition the runtime **latches** a code into the shared-memory header; the host reads it back in `validate_runtime_impl` and prints the lines above. @@ -77,6 +79,7 @@ layer to go looking in, which is what these columns are for. | 101 | ASYNC_COMPLETION_INVALID | kernel (async) | | 102 | ASYNC_WAIT_OVERFLOW | kernel (async) | | 103 | ASYNC_REGISTRATION_FAILED | runtime-internal | +| 104 | READY_QUEUE_OVERFLOW | runtime-internal / config | ### SCHEDULER_TIMEOUT sub-classes @@ -139,9 +142,9 @@ always *fallout* — scroll up for the first failure on that device. For classif ## Minimal reproductions -Each code has a live, minimal trigger in `tests/st/runtime_fatal_codes/`. These are -the fastest way to see what a code looks like, and the shape to copy when you -suspect one: +Each publicly triggerable code listed below has a live, minimal trigger in +`tests/st/runtime_fatal_codes/`. These are the fastest way to see what a code +looks like, and the shape to copy when you suspect one: | Code | How the ST provokes it | Fixture | | ---- | ---------------------- | ------- | @@ -166,7 +169,7 @@ enforces coverage. Edit those and the log carries the new code correctly: | ---- | ----- | | runtime code names / descriptions / hints | `src/common/runtime_status/error_names.h` | | host-side CANN names / descriptions / hints | `src/common/platform/include/host/acl_error_names.h` | -| `SCHEDULER_TIMEOUT` sub-class labels | `src/{arch}/runtime/tensormap_and_ringbuffer/common/pto_runtime_status.h` | +| `SCHEDULER_TIMEOUT` sub-class labels | `src/{arch}/runtime/{host_build_graph,tensormap_and_ringbuffer}/common/pto_runtime_status.h` | | completeness test | `tests/ut/cpp/common/test_error_code_names.cpp` | **This page does not need updating for a new code** — deliberately. The tables @@ -176,7 +179,7 @@ time, so there is nothing to drift out of sync. ## References -- Code definitions: `src/{arch}/runtime/tensormap_and_ringbuffer/common/pto_runtime_status.h` +- Code definitions: `src/{arch}/runtime/{host_build_graph,tensormap_and_ringbuffer}/common/pto_runtime_status.h` - Host print site: `.../host/runtime_maker.cpp` (`validate_runtime_impl`) - Sub-class logic: `.../runtime/scheduler/scheduler_cold_path.cpp` (`classify_stall_reason`) - End-to-end negative tests: `tests/st/runtime_fatal_codes/` diff --git a/src/a2a3/runtime/host_build_graph/host/runtime_maker.cpp b/src/a2a3/runtime/host_build_graph/host/runtime_maker.cpp index f23e6d647a..572d3828a1 100644 --- a/src/a2a3/runtime/host_build_graph/host/runtime_maker.cpp +++ b/src/a2a3/runtime/host_build_graph/host/runtime_maker.cpp @@ -406,12 +406,17 @@ bool upload_graph_submissions(Runtime *runtime, const HostApi *api, GraphHostSta const size_t count = graph_host_upload_count(graph_state); for (size_t index = 0; index < count; ++index) { std::optional upload = graph_host_upload(graph_state, index); - if (!upload.has_value() || upload->outer_slot->task_kind != TaskKind::GRAPH || + if (!upload.has_value() || upload->outer_slot == nullptr || upload->data == nullptr || + upload->bytes < sizeof(GraphSubmission) || upload->outer_slot->task_kind != TaskKind::GRAPH || upload->outer_slot->task == nullptr) { LOG_ERROR("host-orch: invalid pending Graph POD image"); return false; } auto *submission = reinterpret_cast(upload->data); + if (!graph_submission_wire_size_valid(*submission, upload->bytes)) { + LOG_ERROR("host-orch: Graph submission size does not match its POD image"); + return false; + } const GraphDefinition *definition = graph_submission_definition(*submission); size_t execution_bytes = 0; if (definition == nullptr || definition->full_key != submission->graph_key || definition->task_count == 0 || diff --git a/src/a2a3/runtime/host_build_graph/runtime/graph_execution.h b/src/a2a3/runtime/host_build_graph/runtime/graph_execution.h index 6b88f3f548..5642d983f8 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/graph_execution.h +++ b/src/a2a3/runtime/host_build_graph/runtime/graph_execution.h @@ -243,9 +243,18 @@ inline const GraphDefinition *graph_submission_definition(const GraphSubmission sizeof(GraphDefinition) > submission.total_bytes - submission.definition_offset) { return nullptr; } - return reinterpret_cast( + const auto *definition = reinterpret_cast( reinterpret_cast(&submission) + submission.definition_offset ); + if (definition->total_bytes < sizeof(GraphDefinition) || + definition->total_bytes > submission.total_bytes - submission.definition_offset) { + return nullptr; + } + return definition; +} + +inline bool graph_submission_wire_size_valid(const GraphSubmission &submission, size_t available_bytes) { + return available_bytes >= sizeof(GraphSubmission) && submission.total_bytes == available_bytes; } inline const GraphTensor *graph_submission_tensors(const GraphSubmission &submission) { @@ -366,6 +375,8 @@ struct GraphExecution { uint32_t boundary_scalar_count{0}; }; +static_assert(offsetof(GraphExecution, storage_magic) == 0); +static_assert(sizeof(GraphExecution::storage_magic) == sizeof(uint64_t)); static_assert(std::is_trivially_destructible_v); static_assert(std::is_trivially_destructible_v); @@ -442,7 +453,7 @@ inline void graph_execution_retire_node(GraphExecution &execution) { inline bool graph_submission_signal(GraphSubmission &submission, uint32_t bit) { constexpr uint32_t BOTH = 0x3; uint32_t observed = __atomic_fetch_or(&submission.activation_gate, bit, __ATOMIC_ACQ_REL); - return (observed | bit) == BOTH; + return observed != BOTH && (observed | bit) == BOTH; } inline GraphExecution *graph_submission_local_execution(GraphSubmission &submission) { diff --git a/src/a2a3/runtime/host_build_graph/runtime/pto_async_wait.h b/src/a2a3/runtime/host_build_graph/runtime/pto_async_wait.h index ac2d54619c..367ce1abb5 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/pto_async_wait.h +++ b/src/a2a3/runtime/host_build_graph/runtime/pto_async_wait.h @@ -9,8 +9,7 @@ * ----------------------------------------------------------------------------------------------------------- */ -#ifndef PTO_ASYNC_WAIT_H -#define PTO_ASYNC_WAIT_H +#pragma once #include #include @@ -124,7 +123,8 @@ struct AsyncWaitEntry { }; struct AsyncPollResult { - int32_t completed{0}; + int32_t completed{0}; // Host-submitted stream tasks completed. + int32_t resolved{0}; // All task completions, including internal Graph nodes. int32_t error_code{PTO2_ERROR_NONE}; PTO2TaskSlotState *failed_slot_state{nullptr}; }; @@ -174,6 +174,8 @@ struct AsyncWaitList { struct DrainCompletionSink { PTO2SchedulerState *sched{nullptr}; int32_t inline_completed{0}; + int32_t inline_resolved{0}; + int32_t error_code{PTO2_ERROR_NONE}; #if SIMPLER_SCHED_PROFILING int32_t thread_idx{0}; #endif @@ -236,7 +238,10 @@ struct AsyncWaitList { // conditions => NotDeferred. Complete it inline when the // sink allows; otherwise fall back to the entry-store path. if (sink.can_inline_complete()) { - (void)try_inline_complete_locked(sink, *slot_state_ptr); + if (!try_inline_complete_locked(sink, *slot_state_ptr)) { + error_code = sink.error_code; + return drained; + } continue; } if (count >= MAX_ASYNC_WAITS) { @@ -295,5 +300,3 @@ struct AsyncWaitList { #endif ); }; - -#endif // PTO_ASYNC_WAIT_H diff --git a/src/a2a3/runtime/host_build_graph/runtime/scheduler/graph_execution.cpp b/src/a2a3/runtime/host_build_graph/runtime/scheduler/graph_execution.cpp index 8c1b176932..109ef5705c 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/scheduler/graph_execution.cpp +++ b/src/a2a3/runtime/host_build_graph/runtime/scheduler/graph_execution.cpp @@ -565,6 +565,10 @@ GraphMaterializeResult graph_execution_materialize_slice( } } else { for (int32_t j = 0; j < payload.tensor_count; ++j) { + if (execution.materialized_tensor_patches >= execution.materialized_tensor_patch_count) { + execution.materialize_busy.store(0, std::memory_order_release); + return GraphMaterializeResult::INVALID; + } const GraphTensorAddressPatch &patch = execution.tensor_patches[execution.materialized_tensor_patches++]; if (patch.source == static_cast(GraphTensorAddressSource::BOUNDARY)) { diff --git a/src/a2a3/runtime/host_build_graph/runtime/scheduler/pto_scheduler.h b/src/a2a3/runtime/host_build_graph/runtime/scheduler/pto_scheduler.h index 3cfad7c4c7..9c1a8bb301 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/scheduler/pto_scheduler.h +++ b/src/a2a3/runtime/host_build_graph/runtime/scheduler/pto_scheduler.h @@ -488,6 +488,25 @@ struct PTO2SchedulerState { // dummy_ready_queue and are retired inline; a ready sync_start cohort goes to // the per-shape ready_sync_queues[] (drained as Tier-0); everything else to // ready_queues[]. + void latch_ready_queue_overflow(int32_t thread_idx = -1) { + int32_t expected = PTO2_ERROR_NONE; + const bool latched = sm_header->sched_error_code.compare_exchange_strong( + expected, PTO2_ERROR_READY_QUEUE_OVERFLOW, std::memory_order_acq_rel, std::memory_order_acquire + ); + if (latched && thread_idx >= 0) { + sm_header->sched_error_thread.store(thread_idx, std::memory_order_release); + } + if (thread_idx >= 0 && thread_idx < 32) { + sm_header->sched_error_bitmap.fetch_or(1U << static_cast(thread_idx), std::memory_order_acq_rel); + } + } + + bool push_graph_prepare(PTO2TaskSlotState *slot_state, uint64_t task_id, int32_t thread_idx) { + if (graph_prepare_queue.push_tagged(slot_state, task_id)) return true; + latch_ready_queue_overflow(thread_idx); + return false; + } + void push_ready_routed(PTO2TaskSlotState *slot_state) { bool pushed; if (slot_state->task_kind == TaskKind::GRAPH) { @@ -511,10 +530,7 @@ struct PTO2SchedulerState { // forward-progress timeout). The graph_ready push is checked identically // so a graph task cannot be dropped either. if (!pushed) { - int32_t expected = PTO2_ERROR_NONE; - sm_header->sched_error_code.compare_exchange_strong( - expected, PTO2_ERROR_READY_QUEUE_OVERFLOW, std::memory_order_acq_rel, std::memory_order_acquire - ); + latch_ready_queue_overflow(); } } @@ -959,6 +975,7 @@ struct PTO2SchedulerState { struct TaskCompletionOutcome { uint32_t fanout_edges{0}; int32_t stream_tasks_completed{0}; + int32_t error_code{PTO2_ERROR_NONE}; }; TaskCompletionOutcome complete_task( @@ -981,11 +998,21 @@ struct PTO2SchedulerState { } GraphExecution *execution = graph_execution_from_slot(slot_state); - if (execution == nullptr || execution->definition == nullptr || execution->nodes == nullptr) return outcome; + if (execution == nullptr || execution->definition == nullptr || execution->nodes == nullptr || + execution->state.load(std::memory_order_acquire) != GraphExecutionState::ACTIVE) { + outcome.error_code = PTO2_ERROR_INVALID_ARGS; + return outcome; + } const int32_t saved_node_index = slot_state.graph_node_index; - if (saved_node_index < 0) return outcome; + if (saved_node_index < 0) { + outcome.error_code = PTO2_ERROR_INVALID_ARGS; + return outcome; + } const uint32_t node_index = static_cast(saved_node_index); - if (node_index >= static_cast(execution->node_count)) return outcome; + if (node_index >= static_cast(execution->node_count)) { + outcome.error_code = PTO2_ERROR_INVALID_ARGS; + return outcome; + } // Publish completion before closing the wake list. A consumer that // loses registration to the sentinel acquires this state when it @@ -1113,7 +1140,12 @@ AsyncWaitList::try_inline_complete_locked(AsyncWaitList::DrainCompletionSink &si #else PTO2SchedulerState::TaskCompletionOutcome outcome = sink.sched->complete_task(slot_state); #endif + if (outcome.error_code != PTO2_ERROR_NONE) { + sink.error_code = outcome.error_code; + return false; + } sink.inline_completed += outcome.stream_tasks_completed; + sink.inline_resolved++; return true; } @@ -1142,6 +1174,7 @@ inline AsyncPollResult AsyncWaitList::poll_and_complete( return result; } result.completed += sink.inline_completed; + result.resolved += sink.inline_resolved; for (int32_t i = count - 1; i >= 0; --i) { AsyncWaitEntry &entry = entries[i]; @@ -1178,8 +1211,15 @@ inline AsyncPollResult AsyncWaitList::poll_and_complete( #else PTO2SchedulerState::TaskCompletionOutcome outcome = sched->complete_task(*entry.slot_state); #endif + if (outcome.error_code != PTO2_ERROR_NONE) { + result.error_code = outcome.error_code; + result.failed_slot_state = entry.slot_state; + unlock(); + return result; + } // Polling: completion is fully published inline; no deferred release. result.completed += outcome.stream_tasks_completed; + result.resolved++; int32_t last = count - 1; if (i != last) entries[i] = entries[last]; diff --git a/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_cold_path.cpp b/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_cold_path.cpp index fe856bb3f3..601528eab9 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_cold_path.cpp +++ b/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_cold_path.cpp @@ -45,6 +45,13 @@ static void latch_scheduler_error(PTO2SharedMemoryHeader *header, int32_t thread } } +void SchedulerContext::fail_scheduler(Runtime *runtime, int32_t thread_idx, int32_t error_code) { + latch_scheduler_error(sched_ == nullptr ? nullptr : sched_->sm_header, thread_idx, error_code); + if (!completed_.exchange(true, std::memory_order_acq_rel)) { + emergency_shutdown(runtime); + } +} + LoopAction SchedulerContext::handle_orchestrator_exit( int32_t thread_idx, PTO2SharedMemoryHeader *header, Runtime *runtime, int32_t &task_count ) { @@ -1140,9 +1147,7 @@ void SchedulerContext::classify_partition(int32_t thread_idx, int32_t nthreads) } PTO2TaskSlotState &slot = ring.get_slot_state_by_task_id(id); if (slot.task_kind == TaskKind::GRAPH) { - while (!sched_->graph_prepare_queue.push_tagged(&slot, slot.task->task_id.raw)) { - SPIN_WAIT_HINT(); - } + if (!sched_->push_graph_prepare(&slot, slot.task->task_id.raw, thread_idx)) return; } int32_t state = sched_->classify_fanin_state(&slot); if (state < 0) { diff --git a/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_context.h b/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_context.h index 3c0bc092ba..7ecc04e544 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_context.h +++ b/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_context.h @@ -8,8 +8,7 @@ * See LICENSE in the root of the software repository for the full text of the License. * ----------------------------------------------------------------------------------------------------------- */ -#ifndef SCHEDULER_CONTEXT_H -#define SCHEDULER_CONTEXT_H +#pragma once #include "aicpu/device_phase_aicpu.h" #include "aicpu/platform_regs.h" @@ -272,6 +271,8 @@ class SchedulerContext { // deinit their AICore register blocks. Idempotent. void emergency_shutdown(Runtime *runtime); + __attribute__((noinline, cold)) void fail_scheduler(Runtime *runtime, int32_t thread_idx, int32_t error_code); + // ========================================================================= // Dispatch (scheduler_dispatch.cpp) // ========================================================================= @@ -575,5 +576,3 @@ class SchedulerContext { return func_id_to_addr_[func_id]; } }; - -#endif // SCHEDULER_CONTEXT_H diff --git a/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_dispatch.cpp b/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_dispatch.cpp index c6db2a6f78..22d23f1d01 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_dispatch.cpp +++ b/src/a2a3/runtime/host_build_graph/runtime/scheduler/scheduler_dispatch.cpp @@ -902,7 +902,7 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread int32_t resolved_this_pass = 0; bool resolved_any = false; - for (int32_t s = 0; s < active_sched_threads_; s++) { + for (int32_t s = 0; s < active_sched_threads_ && !completed_.load(std::memory_order_acquire); s++) { PTO2TaskSlotState *slot; while ((slot = sp_queues_[s].pop()) != nullptr) { #if SIMPLER_SCHED_PROFILING @@ -910,10 +910,15 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread #else PTO2SchedulerState::TaskCompletionOutcome outcome = sched_->complete_task(*slot); #endif + if (outcome.error_code != PTO2_ERROR_NONE) { + fail_scheduler(runtime, thread_idx, outcome.error_code); + break; + } resolved_this_pass += outcome.stream_tasks_completed; resolved_any = true; } } + if (completed_.load(std::memory_order_acquire)) break; // Async deferred completions, moved off the scheduler threads. Every // condition that fires resolves via on_task_complete inside @@ -929,17 +934,11 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread #endif ); if (poll_result.error_code != PTO2_ERROR_NONE) { - int32_t expected = PTO2_ERROR_NONE; - header->sched_error_code.compare_exchange_strong( - expected, poll_result.error_code, std::memory_order_acq_rel, std::memory_order_acquire - ); - if (!completed_.exchange(true, std::memory_order_acq_rel)) { - emergency_shutdown(runtime); - } + fail_scheduler(runtime, thread_idx, poll_result.error_code); break; } resolved_this_pass += poll_result.completed; - resolved_any = resolved_any || poll_result.completed > 0; + resolved_any = resolved_any || poll_result.resolved > 0; } // Dependency-only tasks (empty active_mask, or a predicate that failed) @@ -958,11 +957,17 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread #else PTO2SchedulerState::TaskCompletionOutcome outcome = sched_->complete_task(*dummy_batch[di]); #endif + if (outcome.error_code != PTO2_ERROR_NONE) { + fail_scheduler(runtime, thread_idx, outcome.error_code); + break; + } resolved_this_pass += outcome.stream_tasks_completed; resolved_any = true; } + if (completed_.load(std::memory_order_acquire)) break; } } + if (completed_.load(std::memory_order_acquire)) break; if (resolved_any) { if (resolved_this_pass > 0) { @@ -1288,16 +1293,7 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ (void)sched_->activate_graph_task(*graph_slot); made_progress = true; } else { - int32_t expected = PTO2_ERROR_NONE; - if (header->sched_error_code.compare_exchange_strong( - expected, PTO2_ERROR_INVALID_ARGS, std::memory_order_acq_rel, std::memory_order_acquire - )) { - header->sched_error_thread.store(thread_idx, std::memory_order_release); - } - header->sched_error_bitmap.fetch_or( - 1U << static_cast(thread_idx), std::memory_order_acq_rel - ); - completed_.store(true, std::memory_order_release); + fail_scheduler(runtime, thread_idx, PTO2_ERROR_INVALID_ARGS); break; } } @@ -1308,16 +1304,7 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ const bool valid_slot = prepare_slot->task != nullptr && prepare_slot->task_kind == TaskKind::GRAPH && prepare_slot->task->task_id.raw == prepare_task_id; if (!valid_slot) { - int32_t expected = PTO2_ERROR_NONE; - if (header->sched_error_code.compare_exchange_strong( - expected, PTO2_ERROR_INVALID_ARGS, std::memory_order_acq_rel, std::memory_order_acquire - )) { - header->sched_error_thread.store(thread_idx, std::memory_order_release); - } - header->sched_error_bitmap.fetch_or( - 1U << static_cast(thread_idx), std::memory_order_acq_rel - ); - completed_.store(true, std::memory_order_release); + fail_scheduler(runtime, thread_idx, PTO2_ERROR_INVALID_ARGS); break; } #if SIMPLER_DFX @@ -1328,20 +1315,12 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ GraphMaterializeResult result = sched_->prepare_graph_task(*prepare_slot, GRAPH_MATERIALIZE_SLICE_NODES, &nodes_materialized); if (result == GraphMaterializeResult::PENDING || result == GraphMaterializeResult::BUSY) { - while (!sched_->graph_prepare_queue.push_tagged(prepare_slot, prepare_task_id)) { - SPIN_WAIT_HINT(); + if (!sched_->push_graph_prepare(prepare_slot, prepare_task_id, thread_idx)) { + fail_scheduler(runtime, thread_idx, PTO2_ERROR_READY_QUEUE_OVERFLOW); + break; } } else if (result == GraphMaterializeResult::INVALID) { - int32_t expected = PTO2_ERROR_NONE; - if (header->sched_error_code.compare_exchange_strong( - expected, PTO2_ERROR_INVALID_ARGS, std::memory_order_acq_rel, std::memory_order_acquire - )) { - header->sched_error_thread.store(thread_idx, std::memory_order_release); - } - header->sched_error_bitmap.fetch_or( - 1U << static_cast(thread_idx), std::memory_order_acq_rel - ); - completed_.store(true, std::memory_order_release); + fail_scheduler(runtime, thread_idx, PTO2_ERROR_INVALID_ARGS); break; } if (nodes_materialized > 0 || result == GraphMaterializeResult::PREPARED) { diff --git a/src/a5/runtime/host_build_graph/host/runtime_maker.cpp b/src/a5/runtime/host_build_graph/host/runtime_maker.cpp index f27d1ad88e..52ce127791 100644 --- a/src/a5/runtime/host_build_graph/host/runtime_maker.cpp +++ b/src/a5/runtime/host_build_graph/host/runtime_maker.cpp @@ -456,12 +456,17 @@ bool upload_graph_submissions(Runtime *runtime, const HostApi *api, GraphHostSta const size_t count = graph_host_upload_count(graph_state); for (size_t index = 0; index < count; ++index) { std::optional upload = graph_host_upload(graph_state, index); - if (!upload.has_value() || upload->outer_slot->task_kind != TaskKind::GRAPH || + if (!upload.has_value() || upload->outer_slot == nullptr || upload->data == nullptr || + upload->bytes < sizeof(GraphSubmission) || upload->outer_slot->task_kind != TaskKind::GRAPH || upload->outer_slot->task == nullptr) { LOG_ERROR("host-orch: invalid pending Graph POD image"); return false; } auto *submission = reinterpret_cast(upload->data); + if (!graph_submission_wire_size_valid(*submission, upload->bytes)) { + LOG_ERROR("host-orch: Graph submission size does not match its POD image"); + return false; + } const GraphDefinition *definition = graph_submission_definition(*submission); size_t execution_bytes = 0; if (definition == nullptr || definition->full_key != submission->graph_key || definition->task_count == 0 || diff --git a/src/a5/runtime/host_build_graph/runtime/graph_execution.h b/src/a5/runtime/host_build_graph/runtime/graph_execution.h index 6b88f3f548..5642d983f8 100644 --- a/src/a5/runtime/host_build_graph/runtime/graph_execution.h +++ b/src/a5/runtime/host_build_graph/runtime/graph_execution.h @@ -243,9 +243,18 @@ inline const GraphDefinition *graph_submission_definition(const GraphSubmission sizeof(GraphDefinition) > submission.total_bytes - submission.definition_offset) { return nullptr; } - return reinterpret_cast( + const auto *definition = reinterpret_cast( reinterpret_cast(&submission) + submission.definition_offset ); + if (definition->total_bytes < sizeof(GraphDefinition) || + definition->total_bytes > submission.total_bytes - submission.definition_offset) { + return nullptr; + } + return definition; +} + +inline bool graph_submission_wire_size_valid(const GraphSubmission &submission, size_t available_bytes) { + return available_bytes >= sizeof(GraphSubmission) && submission.total_bytes == available_bytes; } inline const GraphTensor *graph_submission_tensors(const GraphSubmission &submission) { @@ -366,6 +375,8 @@ struct GraphExecution { uint32_t boundary_scalar_count{0}; }; +static_assert(offsetof(GraphExecution, storage_magic) == 0); +static_assert(sizeof(GraphExecution::storage_magic) == sizeof(uint64_t)); static_assert(std::is_trivially_destructible_v); static_assert(std::is_trivially_destructible_v); @@ -442,7 +453,7 @@ inline void graph_execution_retire_node(GraphExecution &execution) { inline bool graph_submission_signal(GraphSubmission &submission, uint32_t bit) { constexpr uint32_t BOTH = 0x3; uint32_t observed = __atomic_fetch_or(&submission.activation_gate, bit, __ATOMIC_ACQ_REL); - return (observed | bit) == BOTH; + return observed != BOTH && (observed | bit) == BOTH; } inline GraphExecution *graph_submission_local_execution(GraphSubmission &submission) { diff --git a/src/a5/runtime/host_build_graph/runtime/pto_async_wait.h b/src/a5/runtime/host_build_graph/runtime/pto_async_wait.h index ac2d54619c..367ce1abb5 100644 --- a/src/a5/runtime/host_build_graph/runtime/pto_async_wait.h +++ b/src/a5/runtime/host_build_graph/runtime/pto_async_wait.h @@ -9,8 +9,7 @@ * ----------------------------------------------------------------------------------------------------------- */ -#ifndef PTO_ASYNC_WAIT_H -#define PTO_ASYNC_WAIT_H +#pragma once #include #include @@ -124,7 +123,8 @@ struct AsyncWaitEntry { }; struct AsyncPollResult { - int32_t completed{0}; + int32_t completed{0}; // Host-submitted stream tasks completed. + int32_t resolved{0}; // All task completions, including internal Graph nodes. int32_t error_code{PTO2_ERROR_NONE}; PTO2TaskSlotState *failed_slot_state{nullptr}; }; @@ -174,6 +174,8 @@ struct AsyncWaitList { struct DrainCompletionSink { PTO2SchedulerState *sched{nullptr}; int32_t inline_completed{0}; + int32_t inline_resolved{0}; + int32_t error_code{PTO2_ERROR_NONE}; #if SIMPLER_SCHED_PROFILING int32_t thread_idx{0}; #endif @@ -236,7 +238,10 @@ struct AsyncWaitList { // conditions => NotDeferred. Complete it inline when the // sink allows; otherwise fall back to the entry-store path. if (sink.can_inline_complete()) { - (void)try_inline_complete_locked(sink, *slot_state_ptr); + if (!try_inline_complete_locked(sink, *slot_state_ptr)) { + error_code = sink.error_code; + return drained; + } continue; } if (count >= MAX_ASYNC_WAITS) { @@ -295,5 +300,3 @@ struct AsyncWaitList { #endif ); }; - -#endif // PTO_ASYNC_WAIT_H diff --git a/src/a5/runtime/host_build_graph/runtime/scheduler/graph_execution.cpp b/src/a5/runtime/host_build_graph/runtime/scheduler/graph_execution.cpp index 8c1b176932..109ef5705c 100644 --- a/src/a5/runtime/host_build_graph/runtime/scheduler/graph_execution.cpp +++ b/src/a5/runtime/host_build_graph/runtime/scheduler/graph_execution.cpp @@ -565,6 +565,10 @@ GraphMaterializeResult graph_execution_materialize_slice( } } else { for (int32_t j = 0; j < payload.tensor_count; ++j) { + if (execution.materialized_tensor_patches >= execution.materialized_tensor_patch_count) { + execution.materialize_busy.store(0, std::memory_order_release); + return GraphMaterializeResult::INVALID; + } const GraphTensorAddressPatch &patch = execution.tensor_patches[execution.materialized_tensor_patches++]; if (patch.source == static_cast(GraphTensorAddressSource::BOUNDARY)) { diff --git a/src/a5/runtime/host_build_graph/runtime/scheduler/pto_scheduler.h b/src/a5/runtime/host_build_graph/runtime/scheduler/pto_scheduler.h index 3cfad7c4c7..9c1a8bb301 100644 --- a/src/a5/runtime/host_build_graph/runtime/scheduler/pto_scheduler.h +++ b/src/a5/runtime/host_build_graph/runtime/scheduler/pto_scheduler.h @@ -488,6 +488,25 @@ struct PTO2SchedulerState { // dummy_ready_queue and are retired inline; a ready sync_start cohort goes to // the per-shape ready_sync_queues[] (drained as Tier-0); everything else to // ready_queues[]. + void latch_ready_queue_overflow(int32_t thread_idx = -1) { + int32_t expected = PTO2_ERROR_NONE; + const bool latched = sm_header->sched_error_code.compare_exchange_strong( + expected, PTO2_ERROR_READY_QUEUE_OVERFLOW, std::memory_order_acq_rel, std::memory_order_acquire + ); + if (latched && thread_idx >= 0) { + sm_header->sched_error_thread.store(thread_idx, std::memory_order_release); + } + if (thread_idx >= 0 && thread_idx < 32) { + sm_header->sched_error_bitmap.fetch_or(1U << static_cast(thread_idx), std::memory_order_acq_rel); + } + } + + bool push_graph_prepare(PTO2TaskSlotState *slot_state, uint64_t task_id, int32_t thread_idx) { + if (graph_prepare_queue.push_tagged(slot_state, task_id)) return true; + latch_ready_queue_overflow(thread_idx); + return false; + } + void push_ready_routed(PTO2TaskSlotState *slot_state) { bool pushed; if (slot_state->task_kind == TaskKind::GRAPH) { @@ -511,10 +530,7 @@ struct PTO2SchedulerState { // forward-progress timeout). The graph_ready push is checked identically // so a graph task cannot be dropped either. if (!pushed) { - int32_t expected = PTO2_ERROR_NONE; - sm_header->sched_error_code.compare_exchange_strong( - expected, PTO2_ERROR_READY_QUEUE_OVERFLOW, std::memory_order_acq_rel, std::memory_order_acquire - ); + latch_ready_queue_overflow(); } } @@ -959,6 +975,7 @@ struct PTO2SchedulerState { struct TaskCompletionOutcome { uint32_t fanout_edges{0}; int32_t stream_tasks_completed{0}; + int32_t error_code{PTO2_ERROR_NONE}; }; TaskCompletionOutcome complete_task( @@ -981,11 +998,21 @@ struct PTO2SchedulerState { } GraphExecution *execution = graph_execution_from_slot(slot_state); - if (execution == nullptr || execution->definition == nullptr || execution->nodes == nullptr) return outcome; + if (execution == nullptr || execution->definition == nullptr || execution->nodes == nullptr || + execution->state.load(std::memory_order_acquire) != GraphExecutionState::ACTIVE) { + outcome.error_code = PTO2_ERROR_INVALID_ARGS; + return outcome; + } const int32_t saved_node_index = slot_state.graph_node_index; - if (saved_node_index < 0) return outcome; + if (saved_node_index < 0) { + outcome.error_code = PTO2_ERROR_INVALID_ARGS; + return outcome; + } const uint32_t node_index = static_cast(saved_node_index); - if (node_index >= static_cast(execution->node_count)) return outcome; + if (node_index >= static_cast(execution->node_count)) { + outcome.error_code = PTO2_ERROR_INVALID_ARGS; + return outcome; + } // Publish completion before closing the wake list. A consumer that // loses registration to the sentinel acquires this state when it @@ -1113,7 +1140,12 @@ AsyncWaitList::try_inline_complete_locked(AsyncWaitList::DrainCompletionSink &si #else PTO2SchedulerState::TaskCompletionOutcome outcome = sink.sched->complete_task(slot_state); #endif + if (outcome.error_code != PTO2_ERROR_NONE) { + sink.error_code = outcome.error_code; + return false; + } sink.inline_completed += outcome.stream_tasks_completed; + sink.inline_resolved++; return true; } @@ -1142,6 +1174,7 @@ inline AsyncPollResult AsyncWaitList::poll_and_complete( return result; } result.completed += sink.inline_completed; + result.resolved += sink.inline_resolved; for (int32_t i = count - 1; i >= 0; --i) { AsyncWaitEntry &entry = entries[i]; @@ -1178,8 +1211,15 @@ inline AsyncPollResult AsyncWaitList::poll_and_complete( #else PTO2SchedulerState::TaskCompletionOutcome outcome = sched->complete_task(*entry.slot_state); #endif + if (outcome.error_code != PTO2_ERROR_NONE) { + result.error_code = outcome.error_code; + result.failed_slot_state = entry.slot_state; + unlock(); + return result; + } // Polling: completion is fully published inline; no deferred release. result.completed += outcome.stream_tasks_completed; + result.resolved++; int32_t last = count - 1; if (i != last) entries[i] = entries[last]; diff --git a/src/a5/runtime/host_build_graph/runtime/scheduler/scheduler_cold_path.cpp b/src/a5/runtime/host_build_graph/runtime/scheduler/scheduler_cold_path.cpp index fe856bb3f3..601528eab9 100644 --- a/src/a5/runtime/host_build_graph/runtime/scheduler/scheduler_cold_path.cpp +++ b/src/a5/runtime/host_build_graph/runtime/scheduler/scheduler_cold_path.cpp @@ -45,6 +45,13 @@ static void latch_scheduler_error(PTO2SharedMemoryHeader *header, int32_t thread } } +void SchedulerContext::fail_scheduler(Runtime *runtime, int32_t thread_idx, int32_t error_code) { + latch_scheduler_error(sched_ == nullptr ? nullptr : sched_->sm_header, thread_idx, error_code); + if (!completed_.exchange(true, std::memory_order_acq_rel)) { + emergency_shutdown(runtime); + } +} + LoopAction SchedulerContext::handle_orchestrator_exit( int32_t thread_idx, PTO2SharedMemoryHeader *header, Runtime *runtime, int32_t &task_count ) { @@ -1140,9 +1147,7 @@ void SchedulerContext::classify_partition(int32_t thread_idx, int32_t nthreads) } PTO2TaskSlotState &slot = ring.get_slot_state_by_task_id(id); if (slot.task_kind == TaskKind::GRAPH) { - while (!sched_->graph_prepare_queue.push_tagged(&slot, slot.task->task_id.raw)) { - SPIN_WAIT_HINT(); - } + if (!sched_->push_graph_prepare(&slot, slot.task->task_id.raw, thread_idx)) return; } int32_t state = sched_->classify_fanin_state(&slot); if (state < 0) { diff --git a/src/a5/runtime/host_build_graph/runtime/scheduler/scheduler_context.h b/src/a5/runtime/host_build_graph/runtime/scheduler/scheduler_context.h index 51addff20f..1d030a7986 100644 --- a/src/a5/runtime/host_build_graph/runtime/scheduler/scheduler_context.h +++ b/src/a5/runtime/host_build_graph/runtime/scheduler/scheduler_context.h @@ -8,8 +8,7 @@ * See LICENSE in the root of the software repository for the full text of the License. * ----------------------------------------------------------------------------------------------------------- */ -#ifndef SCHEDULER_CONTEXT_H -#define SCHEDULER_CONTEXT_H +#pragma once #include "aicpu/device_phase_aicpu.h" #include "aicpu/platform_regs.h" @@ -272,6 +271,8 @@ class SchedulerContext { // deinit their AICore register blocks. Idempotent. void emergency_shutdown(Runtime *runtime); + __attribute__((noinline, cold)) void fail_scheduler(Runtime *runtime, int32_t thread_idx, int32_t error_code); + // ========================================================================= // Dispatch (scheduler_dispatch.cpp) // ========================================================================= @@ -575,5 +576,3 @@ class SchedulerContext { return func_id_to_addr_[func_id]; } }; - -#endif // SCHEDULER_CONTEXT_H diff --git a/src/a5/runtime/host_build_graph/runtime/scheduler/scheduler_dispatch.cpp b/src/a5/runtime/host_build_graph/runtime/scheduler/scheduler_dispatch.cpp index 1f518c7ad1..9b27806ed9 100644 --- a/src/a5/runtime/host_build_graph/runtime/scheduler/scheduler_dispatch.cpp +++ b/src/a5/runtime/host_build_graph/runtime/scheduler/scheduler_dispatch.cpp @@ -908,7 +908,7 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread int32_t resolved_this_pass = 0; bool resolved_any = false; - for (int32_t s = 0; s < active_sched_threads_; s++) { + for (int32_t s = 0; s < active_sched_threads_ && !completed_.load(std::memory_order_acquire); s++) { PTO2TaskSlotState *slot; while ((slot = sp_queues_[s].pop()) != nullptr) { #if SIMPLER_SCHED_PROFILING @@ -916,10 +916,15 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread #else PTO2SchedulerState::TaskCompletionOutcome outcome = sched_->complete_task(*slot); #endif + if (outcome.error_code != PTO2_ERROR_NONE) { + fail_scheduler(runtime, thread_idx, outcome.error_code); + break; + } resolved_this_pass += outcome.stream_tasks_completed; resolved_any = true; } } + if (completed_.load(std::memory_order_acquire)) break; // Async deferred completions, moved off the scheduler threads. Every // condition that fires resolves via on_task_complete inside @@ -935,17 +940,11 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread #endif ); if (poll_result.error_code != PTO2_ERROR_NONE) { - int32_t expected = PTO2_ERROR_NONE; - header->sched_error_code.compare_exchange_strong( - expected, poll_result.error_code, std::memory_order_acq_rel, std::memory_order_acquire - ); - if (!completed_.exchange(true, std::memory_order_acq_rel)) { - emergency_shutdown(runtime); - } + fail_scheduler(runtime, thread_idx, poll_result.error_code); break; } resolved_this_pass += poll_result.completed; - resolved_any = resolved_any || poll_result.completed > 0; + resolved_any = resolved_any || poll_result.resolved > 0; } // Dependency-only tasks (empty active_mask, or a predicate that failed) @@ -964,11 +963,17 @@ int32_t SchedulerContext::run_resolution_thread(Runtime *runtime, int32_t thread #else PTO2SchedulerState::TaskCompletionOutcome outcome = sched_->complete_task(*dummy_batch[di]); #endif + if (outcome.error_code != PTO2_ERROR_NONE) { + fail_scheduler(runtime, thread_idx, outcome.error_code); + break; + } resolved_this_pass += outcome.stream_tasks_completed; resolved_any = true; } + if (completed_.load(std::memory_order_acquire)) break; } } + if (completed_.load(std::memory_order_acquire)) break; if (resolved_any) { int32_t new_total = completed_tasks_.load(std::memory_order_relaxed); @@ -1300,16 +1305,7 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ (void)sched_->activate_graph_task(*graph_slot); made_progress = true; } else { - int32_t expected = PTO2_ERROR_NONE; - if (header->sched_error_code.compare_exchange_strong( - expected, PTO2_ERROR_INVALID_ARGS, std::memory_order_acq_rel, std::memory_order_acquire - )) { - header->sched_error_thread.store(thread_idx, std::memory_order_release); - } - header->sched_error_bitmap.fetch_or( - 1U << static_cast(thread_idx), std::memory_order_acq_rel - ); - completed_.store(true, std::memory_order_release); + fail_scheduler(runtime, thread_idx, PTO2_ERROR_INVALID_ARGS); break; } } @@ -1320,16 +1316,7 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ const bool valid_slot = prepare_slot->task != nullptr && prepare_slot->task_kind == TaskKind::GRAPH && prepare_slot->task->task_id.raw == prepare_task_id; if (!valid_slot) { - int32_t expected = PTO2_ERROR_NONE; - if (header->sched_error_code.compare_exchange_strong( - expected, PTO2_ERROR_INVALID_ARGS, std::memory_order_acq_rel, std::memory_order_acquire - )) { - header->sched_error_thread.store(thread_idx, std::memory_order_release); - } - header->sched_error_bitmap.fetch_or( - 1U << static_cast(thread_idx), std::memory_order_acq_rel - ); - completed_.store(true, std::memory_order_release); + fail_scheduler(runtime, thread_idx, PTO2_ERROR_INVALID_ARGS); break; } #if SIMPLER_DFX @@ -1340,20 +1327,12 @@ int32_t SchedulerContext::resolve_and_dispatch(Runtime *runtime, int32_t thread_ GraphMaterializeResult result = sched_->prepare_graph_task(*prepare_slot, GRAPH_MATERIALIZE_SLICE_NODES, &nodes_materialized); if (result == GraphMaterializeResult::PENDING || result == GraphMaterializeResult::BUSY) { - while (!sched_->graph_prepare_queue.push_tagged(prepare_slot, prepare_task_id)) { - SPIN_WAIT_HINT(); + if (!sched_->push_graph_prepare(prepare_slot, prepare_task_id, thread_idx)) { + fail_scheduler(runtime, thread_idx, PTO2_ERROR_READY_QUEUE_OVERFLOW); + break; } } else if (result == GraphMaterializeResult::INVALID) { - int32_t expected = PTO2_ERROR_NONE; - if (header->sched_error_code.compare_exchange_strong( - expected, PTO2_ERROR_INVALID_ARGS, std::memory_order_acq_rel, std::memory_order_acquire - )) { - header->sched_error_thread.store(thread_idx, std::memory_order_release); - } - header->sched_error_bitmap.fetch_or( - 1U << static_cast(thread_idx), std::memory_order_acq_rel - ); - completed_.store(true, std::memory_order_release); + fail_scheduler(runtime, thread_idx, PTO2_ERROR_INVALID_ARGS); break; } if (nodes_materialized > 0 || result == GraphMaterializeResult::PREPARED) { diff --git a/src/common/runtime_status/error_names.h b/src/common/runtime_status/error_names.h index d858c958f0..402b7b54ac 100644 --- a/src/common/runtime_status/error_names.h +++ b/src/common/runtime_status/error_names.h @@ -70,6 +70,10 @@ static inline const char *error_name(int32_t code) { return "ASYNC_WAIT_OVERFLOW"; case PTO2_ERROR_ASYNC_REGISTRATION_FAILED: return "ASYNC_REGISTRATION_FAILED"; +#ifdef PTO2_ERROR_READY_QUEUE_OVERFLOW + case PTO2_ERROR_READY_QUEUE_OVERFLOW: + return "READY_QUEUE_OVERFLOW"; +#endif default: return "unknown"; } @@ -124,6 +128,10 @@ static inline const char *error_desc(int32_t code) { case PTO2_ERROR_ASYNC_REGISTRATION_FAILED: return "the scheduler received an async completion message of an illegal kind (runtime-internal; " "ASYNC_WAIT_OVERFLOW normally intercepts this first)"; +#ifdef PTO2_ERROR_READY_QUEUE_OVERFLOW + case PTO2_ERROR_READY_QUEUE_OVERFLOW: + return "a scheduler ready queue rejected a task because it had no free slot"; +#endif default: return ""; } @@ -176,6 +184,11 @@ static inline const char *error_hint(int32_t code) { case PTO2_ERROR_ASYNC_WAIT_OVERFLOW: return "cut the number of in-flight async completions per task, and confirm the consumer side " "actually polls and retires them"; +#ifdef PTO2_ERROR_READY_QUEUE_OVERFLOW + case PTO2_ERROR_READY_QUEUE_OVERFLOW: + return "ensure PTO2_RING_TASK_WINDOW does not exceed the ready-queue capacity; otherwise keep the device " + "log and report the scheduler queue-accounting bug"; +#endif default: return ""; } diff --git a/tests/ut/cpp/CMakeLists.txt b/tests/ut/cpp/CMakeLists.txt index 135841c67d..8e73c23bfc 100644 --- a/tests/ut/cpp/CMakeLists.txt +++ b/tests/ut/cpp/CMakeLists.txt @@ -766,6 +766,7 @@ add_a2a3_test(test_acl_hal_device common/test_acl_hal_device.cpp) add_a2a3_runtime_test(test_task_allocator a2a3/test_task_allocator.cpp) add_a2a3_runtime_test(test_scope_deadlock_detection common/test_scope_deadlock_detection.cpp) set(HBG_RUNTIME_DIR ${CMAKE_SOURCE_DIR}/../../../src/a2a3/runtime/host_build_graph/runtime) +set(A5_HBG_RUNTIME_DIR ${CMAKE_SOURCE_DIR}/../../../src/a5/runtime/host_build_graph/runtime) add_a2a3_hbg_runtime_test(test_hbg_task_allocator a2a3/test_task_allocator.cpp) add_a2a3_hbg_runtime_test(test_hbg_tensormap a2a3/test_hbg_tensormap.cpp) add_a2a3_hbg_runtime_test(test_hbg_dep_gen_host_graph a2a3/test_dep_gen_host_graph.cpp) @@ -783,6 +784,26 @@ target_sources(test_hbg_submit_poison PRIVATE ${HBG_RUNTIME_DIR}/shared/runtime.cpp ${CMAKE_SOURCE_DIR}/../../../src/common/platform/shared/aicpu/args_dump_aicpu.cpp ) +add_a2a3_hbg_runtime_test(test_hbg_graph_submit_failure common/test_hbg_graph_submit_failure.cpp) +target_sources(test_hbg_graph_submit_failure PRIVATE + ${HBG_RUNTIME_DIR}/orchestrator_core/pto_orchestrator.cpp + ${HBG_RUNTIME_DIR}/orchestrator_core/pto_ring_buffer.cpp + ${HBG_RUNTIME_DIR}/shared/pto_shared_memory.cpp + ${HBG_RUNTIME_DIR}/shared/pto_tensormap.cpp + ${HBG_RUNTIME_DIR}/shared/pto_runtime2_init.cpp + ${HBG_RUNTIME_DIR}/shared/runtime.cpp + ${CMAKE_SOURCE_DIR}/../../../src/common/platform/shared/aicpu/args_dump_aicpu.cpp +) +add_a5_hbg_runtime_test(test_a5_hbg_graph_submit_failure common/test_hbg_graph_submit_failure.cpp) +target_sources(test_a5_hbg_graph_submit_failure PRIVATE + ${A5_HBG_RUNTIME_DIR}/orchestrator_core/pto_orchestrator.cpp + ${A5_HBG_RUNTIME_DIR}/orchestrator_core/pto_ring_buffer.cpp + ${A5_HBG_RUNTIME_DIR}/shared/pto_shared_memory.cpp + ${A5_HBG_RUNTIME_DIR}/shared/pto_tensormap.cpp + ${A5_HBG_RUNTIME_DIR}/shared/pto_runtime2_init.cpp + ${A5_HBG_RUNTIME_DIR}/shared/runtime.cpp + ${CMAKE_SOURCE_DIR}/../../../src/common/platform/shared/aicpu/args_dump_aicpu.cpp +) add_a5_hbg_runtime_test(test_a5_hbg_core_tracker common/test_hbg_core_tracker.cpp) add_a5_hbg_runtime_test(test_a5_hbg_aicore_completion_mailbox a5/test_aicore_completion_mailbox.cpp) add_a2a3_hbg_runtime_test(test_graph_cache a2a3/test_graph_cache.cpp) diff --git a/tests/ut/cpp/a2a3/test_graph_cache.cpp b/tests/ut/cpp/a2a3/test_graph_cache.cpp index 9428c16817..5a41a59610 100644 --- a/tests/ut/cpp/a2a3/test_graph_cache.cpp +++ b/tests/ut/cpp/a2a3/test_graph_cache.cpp @@ -13,13 +13,17 @@ #include #include +#include #include #include #include +#include #include #include "graph_cache.h" #include "graph_execution.h" +#include "runtime_status/error_names.h" +#include "scheduler/pto_scheduler.h" namespace { @@ -320,4 +324,154 @@ TEST(GraphExecutionReplay, AffineHitRefreshesOnlyDynamicFields) { ); EXPECT_EQ(node.slot.completed_subtasks.load(std::memory_order_relaxed), 0); EXPECT_EQ(node.payload.dispatch_fanin.load(std::memory_order_relaxed), 0); + + graph_execution_mark_completed(*execution); + execution->retired_nodes.store(2, std::memory_order_release); + submission.local_execution = 0; + execution = graph_execution_localize(outer_slot); + ASSERT_NE(execution, nullptr); + ASSERT_TRUE(execution->definition_affine_reuse); + execution->materialized_tensor_patch_count = 1; + + EXPECT_EQ(graph_execution_materialize_slice(outer_slot, *execution, 2), GraphMaterializeResult::INVALID); + EXPECT_EQ(execution->materialized_tensor_patches, 1U); +} + +TEST(GraphSubmissionWire, RejectsDefinitionBeyondSubmission) { + constexpr uint64_t GRAPH_KEY_VALUE = 0x3456; + std::vector image = make_test_submission(GRAPH_KEY_VALUE, 0x1000, 17, 0x2000, 4096); + auto &submission = *reinterpret_cast(image.data()); + auto *definition = reinterpret_cast(image.data() + submission.definition_offset); + definition->total_bytes = submission.total_bytes - submission.definition_offset + 1; + + EXPECT_EQ(graph_submission_definition(submission), nullptr); +} + +TEST(GraphSubmissionWire, RequiresExactAvailableSize) { + constexpr uint64_t GRAPH_KEY_VALUE = 0x4567; + std::vector image = make_test_submission(GRAPH_KEY_VALUE, 0x1000, 17, 0x2000, 4096); + const auto &submission = *reinterpret_cast(image.data()); + + EXPECT_TRUE(graph_submission_wire_size_valid(submission, image.size())); + EXPECT_FALSE(graph_submission_wire_size_valid(submission, image.size() - 1)); + EXPECT_FALSE(graph_submission_wire_size_valid(submission, image.size() + 1)); +} + +TEST(GraphSubmissionActivationGate, ActivatesExactlyOnceUnderContention) { + constexpr int ITERATIONS = 1000; + for (int iteration = 0; iteration < ITERATIONS; ++iteration) { + GraphSubmission submission{}; + std::atomic activations{0}; + std::thread prepared([&] { + if (graph_submission_signal(submission, 0x1)) activations.fetch_add(1, std::memory_order_relaxed); + }); + std::thread ready([&] { + if (graph_submission_signal(submission, 0x2)) activations.fetch_add(1, std::memory_order_relaxed); + }); + prepared.join(); + ready.join(); + EXPECT_EQ(submission.activation_gate, 0x3U); + EXPECT_EQ(activations.load(std::memory_order_relaxed), 1); + } +} + +TEST(GraphSubmissionActivationGate, RetriesDoNotReactivate) { + GraphSubmission submission{}; + + EXPECT_FALSE(graph_submission_signal(submission, 0x1)); + EXPECT_TRUE(graph_submission_signal(submission, 0x2)); + EXPECT_FALSE(graph_submission_signal(submission, 0x1)); + EXPECT_FALSE(graph_submission_signal(submission, 0x2)); +} + +TEST(GraphExecutionErrors, ReadyQueueOverflowHasTriageText) { + EXPECT_STREQ(error_name(PTO2_ERROR_READY_QUEUE_OVERFLOW), "READY_QUEUE_OVERFLOW"); + EXPECT_STRNE(error_desc(PTO2_ERROR_READY_QUEUE_OVERFLOW), ""); + EXPECT_STRNE(error_hint(PTO2_ERROR_READY_QUEUE_OVERFLOW), ""); +} + +TEST(GraphExecutionErrors, GraphReadyQueueOverflowIsReported) { + PTO2SharedMemoryHeader header{}; + PTO2SchedulerState scheduler{}; + scheduler.sm_header = &header; + PTO2ReadyQueueSlot queue_slots[2]{}; + queue_slots[0].sequence.store(0, std::memory_order_relaxed); + queue_slots[1].sequence.store(1, std::memory_order_relaxed); + scheduler.graph_ready_queue.slots = queue_slots; + scheduler.graph_ready_queue.capacity = 2; + scheduler.graph_ready_queue.mask = 1; + scheduler.graph_ready_queue.enqueue_pos.store(0, std::memory_order_relaxed); + scheduler.graph_ready_queue.dequeue_pos.store(0, std::memory_order_relaxed); + PTO2TaskSlotState graph_slots[3]{}; + for (PTO2TaskSlotState &slot : graph_slots) { + slot.task_kind = TaskKind::GRAPH; + } + + scheduler.push_ready_routed(&graph_slots[0]); + scheduler.push_ready_routed(&graph_slots[1]); + scheduler.push_ready_routed(&graph_slots[2]); + + EXPECT_EQ(header.sched_error_code.load(std::memory_order_acquire), PTO2_ERROR_READY_QUEUE_OVERFLOW); +} + +TEST(GraphExecutionErrors, GraphPrepareQueueOverflowIsReported) { + PTO2SharedMemoryHeader header{}; + PTO2SchedulerState scheduler{}; + scheduler.sm_header = &header; + PTO2ReadyQueueSlot queue_slots[2]{}; + queue_slots[0].sequence.store(0, std::memory_order_relaxed); + queue_slots[1].sequence.store(1, std::memory_order_relaxed); + scheduler.graph_prepare_queue.slots = queue_slots; + scheduler.graph_prepare_queue.capacity = 2; + scheduler.graph_prepare_queue.mask = 1; + scheduler.graph_prepare_queue.enqueue_pos.store(0, std::memory_order_relaxed); + scheduler.graph_prepare_queue.dequeue_pos.store(0, std::memory_order_relaxed); + PTO2TaskSlotState graph_slots[3]{}; + + EXPECT_TRUE(scheduler.push_graph_prepare(&graph_slots[0], 10, 3)); + EXPECT_TRUE(scheduler.push_graph_prepare(&graph_slots[1], 11, 3)); + EXPECT_FALSE(scheduler.push_graph_prepare(&graph_slots[2], 12, 3)); + + EXPECT_EQ(header.sched_error_code.load(std::memory_order_acquire), PTO2_ERROR_READY_QUEUE_OVERFLOW); + EXPECT_EQ(header.sched_error_thread.load(std::memory_order_acquire), 3); + EXPECT_EQ(header.sched_error_bitmap.load(std::memory_order_acquire), 1U << 3); +} + +TEST(GraphExecutionErrors, InvalidNodeCompletionIsReported) { + PTO2SchedulerState scheduler{}; + PTO2TaskSlotState slot{}; + slot.task_kind = TaskKind::GRAPH_NODE; + + const PTO2SchedulerState::TaskCompletionOutcome outcome = scheduler.complete_task(slot); + + EXPECT_EQ(outcome.error_code, PTO2_ERROR_INVALID_ARGS); + EXPECT_EQ(outcome.stream_tasks_completed, 0); +} + +TEST(GraphExecutionProgress, InternalNodeResolutionIsNotAHostCompletion) { + PTO2SchedulerState scheduler{}; + GraphDefinition definition{}; + GraphNodeStorage node{}; + GraphExecution execution{}; + execution.definition = &definition; + execution.nodes = &node; + execution.node_storage = &node; + execution.node_count = 1; + execution.remaining_nodes.store(1, std::memory_order_relaxed); + execution.state.store(GraphExecutionState::ACTIVE, std::memory_order_relaxed); + node.slot.task_kind = TaskKind::GRAPH_NODE; + node.slot.graph_context = &execution; + node.slot.graph_node_index = 0; + + AsyncWaitList wait_list{}; + wait_list.entries[0].slot_state = &node.slot; + wait_list.entries[0].task_token = PTO2TaskId::make(0, 1); + wait_list.entries[0].normal_done = true; + wait_list.count = 1; + + const AsyncPollResult result = wait_list.poll_and_complete(nullptr, &scheduler); + + EXPECT_EQ(result.error_code, PTO2_ERROR_NONE); + EXPECT_EQ(result.resolved, 1); + EXPECT_EQ(result.completed, 0); } diff --git a/tests/ut/cpp/a5/test_graph_cache.cpp b/tests/ut/cpp/a5/test_graph_cache.cpp index 9428c16817..5a41a59610 100644 --- a/tests/ut/cpp/a5/test_graph_cache.cpp +++ b/tests/ut/cpp/a5/test_graph_cache.cpp @@ -13,13 +13,17 @@ #include #include +#include #include #include #include +#include #include #include "graph_cache.h" #include "graph_execution.h" +#include "runtime_status/error_names.h" +#include "scheduler/pto_scheduler.h" namespace { @@ -320,4 +324,154 @@ TEST(GraphExecutionReplay, AffineHitRefreshesOnlyDynamicFields) { ); EXPECT_EQ(node.slot.completed_subtasks.load(std::memory_order_relaxed), 0); EXPECT_EQ(node.payload.dispatch_fanin.load(std::memory_order_relaxed), 0); + + graph_execution_mark_completed(*execution); + execution->retired_nodes.store(2, std::memory_order_release); + submission.local_execution = 0; + execution = graph_execution_localize(outer_slot); + ASSERT_NE(execution, nullptr); + ASSERT_TRUE(execution->definition_affine_reuse); + execution->materialized_tensor_patch_count = 1; + + EXPECT_EQ(graph_execution_materialize_slice(outer_slot, *execution, 2), GraphMaterializeResult::INVALID); + EXPECT_EQ(execution->materialized_tensor_patches, 1U); +} + +TEST(GraphSubmissionWire, RejectsDefinitionBeyondSubmission) { + constexpr uint64_t GRAPH_KEY_VALUE = 0x3456; + std::vector image = make_test_submission(GRAPH_KEY_VALUE, 0x1000, 17, 0x2000, 4096); + auto &submission = *reinterpret_cast(image.data()); + auto *definition = reinterpret_cast(image.data() + submission.definition_offset); + definition->total_bytes = submission.total_bytes - submission.definition_offset + 1; + + EXPECT_EQ(graph_submission_definition(submission), nullptr); +} + +TEST(GraphSubmissionWire, RequiresExactAvailableSize) { + constexpr uint64_t GRAPH_KEY_VALUE = 0x4567; + std::vector image = make_test_submission(GRAPH_KEY_VALUE, 0x1000, 17, 0x2000, 4096); + const auto &submission = *reinterpret_cast(image.data()); + + EXPECT_TRUE(graph_submission_wire_size_valid(submission, image.size())); + EXPECT_FALSE(graph_submission_wire_size_valid(submission, image.size() - 1)); + EXPECT_FALSE(graph_submission_wire_size_valid(submission, image.size() + 1)); +} + +TEST(GraphSubmissionActivationGate, ActivatesExactlyOnceUnderContention) { + constexpr int ITERATIONS = 1000; + for (int iteration = 0; iteration < ITERATIONS; ++iteration) { + GraphSubmission submission{}; + std::atomic activations{0}; + std::thread prepared([&] { + if (graph_submission_signal(submission, 0x1)) activations.fetch_add(1, std::memory_order_relaxed); + }); + std::thread ready([&] { + if (graph_submission_signal(submission, 0x2)) activations.fetch_add(1, std::memory_order_relaxed); + }); + prepared.join(); + ready.join(); + EXPECT_EQ(submission.activation_gate, 0x3U); + EXPECT_EQ(activations.load(std::memory_order_relaxed), 1); + } +} + +TEST(GraphSubmissionActivationGate, RetriesDoNotReactivate) { + GraphSubmission submission{}; + + EXPECT_FALSE(graph_submission_signal(submission, 0x1)); + EXPECT_TRUE(graph_submission_signal(submission, 0x2)); + EXPECT_FALSE(graph_submission_signal(submission, 0x1)); + EXPECT_FALSE(graph_submission_signal(submission, 0x2)); +} + +TEST(GraphExecutionErrors, ReadyQueueOverflowHasTriageText) { + EXPECT_STREQ(error_name(PTO2_ERROR_READY_QUEUE_OVERFLOW), "READY_QUEUE_OVERFLOW"); + EXPECT_STRNE(error_desc(PTO2_ERROR_READY_QUEUE_OVERFLOW), ""); + EXPECT_STRNE(error_hint(PTO2_ERROR_READY_QUEUE_OVERFLOW), ""); +} + +TEST(GraphExecutionErrors, GraphReadyQueueOverflowIsReported) { + PTO2SharedMemoryHeader header{}; + PTO2SchedulerState scheduler{}; + scheduler.sm_header = &header; + PTO2ReadyQueueSlot queue_slots[2]{}; + queue_slots[0].sequence.store(0, std::memory_order_relaxed); + queue_slots[1].sequence.store(1, std::memory_order_relaxed); + scheduler.graph_ready_queue.slots = queue_slots; + scheduler.graph_ready_queue.capacity = 2; + scheduler.graph_ready_queue.mask = 1; + scheduler.graph_ready_queue.enqueue_pos.store(0, std::memory_order_relaxed); + scheduler.graph_ready_queue.dequeue_pos.store(0, std::memory_order_relaxed); + PTO2TaskSlotState graph_slots[3]{}; + for (PTO2TaskSlotState &slot : graph_slots) { + slot.task_kind = TaskKind::GRAPH; + } + + scheduler.push_ready_routed(&graph_slots[0]); + scheduler.push_ready_routed(&graph_slots[1]); + scheduler.push_ready_routed(&graph_slots[2]); + + EXPECT_EQ(header.sched_error_code.load(std::memory_order_acquire), PTO2_ERROR_READY_QUEUE_OVERFLOW); +} + +TEST(GraphExecutionErrors, GraphPrepareQueueOverflowIsReported) { + PTO2SharedMemoryHeader header{}; + PTO2SchedulerState scheduler{}; + scheduler.sm_header = &header; + PTO2ReadyQueueSlot queue_slots[2]{}; + queue_slots[0].sequence.store(0, std::memory_order_relaxed); + queue_slots[1].sequence.store(1, std::memory_order_relaxed); + scheduler.graph_prepare_queue.slots = queue_slots; + scheduler.graph_prepare_queue.capacity = 2; + scheduler.graph_prepare_queue.mask = 1; + scheduler.graph_prepare_queue.enqueue_pos.store(0, std::memory_order_relaxed); + scheduler.graph_prepare_queue.dequeue_pos.store(0, std::memory_order_relaxed); + PTO2TaskSlotState graph_slots[3]{}; + + EXPECT_TRUE(scheduler.push_graph_prepare(&graph_slots[0], 10, 3)); + EXPECT_TRUE(scheduler.push_graph_prepare(&graph_slots[1], 11, 3)); + EXPECT_FALSE(scheduler.push_graph_prepare(&graph_slots[2], 12, 3)); + + EXPECT_EQ(header.sched_error_code.load(std::memory_order_acquire), PTO2_ERROR_READY_QUEUE_OVERFLOW); + EXPECT_EQ(header.sched_error_thread.load(std::memory_order_acquire), 3); + EXPECT_EQ(header.sched_error_bitmap.load(std::memory_order_acquire), 1U << 3); +} + +TEST(GraphExecutionErrors, InvalidNodeCompletionIsReported) { + PTO2SchedulerState scheduler{}; + PTO2TaskSlotState slot{}; + slot.task_kind = TaskKind::GRAPH_NODE; + + const PTO2SchedulerState::TaskCompletionOutcome outcome = scheduler.complete_task(slot); + + EXPECT_EQ(outcome.error_code, PTO2_ERROR_INVALID_ARGS); + EXPECT_EQ(outcome.stream_tasks_completed, 0); +} + +TEST(GraphExecutionProgress, InternalNodeResolutionIsNotAHostCompletion) { + PTO2SchedulerState scheduler{}; + GraphDefinition definition{}; + GraphNodeStorage node{}; + GraphExecution execution{}; + execution.definition = &definition; + execution.nodes = &node; + execution.node_storage = &node; + execution.node_count = 1; + execution.remaining_nodes.store(1, std::memory_order_relaxed); + execution.state.store(GraphExecutionState::ACTIVE, std::memory_order_relaxed); + node.slot.task_kind = TaskKind::GRAPH_NODE; + node.slot.graph_context = &execution; + node.slot.graph_node_index = 0; + + AsyncWaitList wait_list{}; + wait_list.entries[0].slot_state = &node.slot; + wait_list.entries[0].task_token = PTO2TaskId::make(0, 1); + wait_list.entries[0].normal_done = true; + wait_list.count = 1; + + const AsyncPollResult result = wait_list.poll_and_complete(nullptr, &scheduler); + + EXPECT_EQ(result.error_code, PTO2_ERROR_NONE); + EXPECT_EQ(result.resolved, 1); + EXPECT_EQ(result.completed, 0); } diff --git a/tests/ut/cpp/common/test_hbg_graph_submit_failure.cpp b/tests/ut/cpp/common/test_hbg_graph_submit_failure.cpp new file mode 100644 index 0000000000..3651b68f15 --- /dev/null +++ b/tests/ut/cpp/common/test_hbg_graph_submit_failure.cpp @@ -0,0 +1,98 @@ +/* + * Copyright (c) PyPTO Contributors. + * This program is free software, you can redistribute it and/or modify it under the terms and conditions of + * CANN Open Software License Agreement Version 2.0 (the "License"). + * Please refer to the License for details. You may not use this file except in compliance with the License. + * THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND, EITHER EXPRESS OR IMPLIED, + * INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT, MERCHANTABILITY, OR FITNESS FOR A PARTICULAR PURPOSE. + * See LICENSE in the root of the software repository for the full text of the License. + * ----------------------------------------------------------------------------------------------------------- + */ + +#include + +#include +#include +#include + +#include "graph_host_state.h" +#include "pto_orchestrator.h" +#include "pto_shared_memory.h" +#include "utils/device_arena.h" + +class HbgGraphSubmitFailureTest : public ::testing::Test { +protected: + DeviceArena sm_arena; + DeviceArena runtime_arena; + PTO2SharedMemoryHandle *sm_handle = nullptr; + PTO2OrchestratorState orch{}; + PTO2SchedulerState sched{}; + PTO2OrchestratorLayout orch_layout{}; + PTO2SchedulerLayout sched_layout{}; + GraphHostStatePtr graph_state; + std::vector gm_heap; + + void SetUp() override { + sm_handle = PTO2SharedMemoryHandle::create_and_init_default(sm_arena); + ASSERT_NE(sm_handle, nullptr); + gm_heap.resize(4096 * PTO2_MAX_RING_DEPTH); + + orch_layout = PTO2OrchestratorState::reserve_layout(runtime_arena, static_cast(PTO2_TASK_WINDOW_SIZE)); + sched_layout = PTO2SchedulerState::reserve_layout(runtime_arena); + ASSERT_NE(runtime_arena.commit(), nullptr); + + ASSERT_TRUE(orch.init_data_from_layout( + orch_layout, runtime_arena, sm_handle->sm_base, gm_heap.data(), 4096, PTO2_TASK_WINDOW_SIZE + )); + ASSERT_TRUE(sched.init_data_from_layout(sched_layout, runtime_arena, sm_handle->sm_base)); + sched.wire_arena_pointers(sched_layout, runtime_arena); + orch.wire_arena_pointers(orch_layout, runtime_arena, &sched); + + graph_state = make_graph_host_state(); + ASSERT_NE(graph_state, nullptr); + orch.graph_host_state = graph_state.get(); + } + + void TearDown() override { + orch.graph_host_state = nullptr; + graph_state.reset(); + orch.destroy(); + sched.destroy(); + runtime_arena.release(); + sm_arena.release(); + } +}; + +TEST_F(HbgGraphSubmitFailureTest, FaninFailureLatchesFatalWithoutPartialUpload) { + std::array storage{}; + uint32_t shape[] = {static_cast(storage.size())}; + ChipTensor boundary = make_tensor_external(storage.data(), shape, 1); + + orch.begin_scope(); + CoreTaskArgs boundary_args; + boundary_args.add_input(boundary); + const GraphScopeResult graph = orch.graph_begin(0x1715, boundary_args, 0x1736); + ASSERT_TRUE(graph.recording); + + CoreTaskArgs node_args; + node_args.add_input(boundary); + ASSERT_TRUE(orch.submit_dummy_task(node_args).task_id().is_valid()); + orch.graph_end(); + ASSERT_FALSE(orch.fatal); + const size_t uploads_before_failure = graph_host_upload_count(*graph_state); + + CoreTaskArgs producer_args; + producer_args.add_output(boundary); + for (int32_t i = 0; i < PTO2_MAX_FANIN + 1; ++i) { + ASSERT_TRUE(orch.submit_dummy_task(producer_args).task_id().is_valid()); + } + + const GraphScopeResult replay = orch.graph_begin(0x1715, boundary_args, 0x1736); + + EXPECT_TRUE(replay.execute_block); + EXPECT_FALSE(replay.recording); + EXPECT_FALSE(replay.task_id.is_valid()); + EXPECT_TRUE(orch.fatal); + EXPECT_EQ(sm_handle->header->orch_error_code.load(std::memory_order_acquire), PTO2_ERROR_DEP_POOL_OVERFLOW); + EXPECT_EQ(graph_host_upload_count(*graph_state), uploads_before_failure); +}