diff --git a/src/a2a3/runtime/host_build_graph/common/pto_runtime_status.h b/src/a2a3/runtime/host_build_graph/common/pto_runtime_status.h index c81a121d80..5344728199 100644 --- a/src/a2a3/runtime/host_build_graph/common/pto_runtime_status.h +++ b/src/a2a3/runtime/host_build_graph/common/pto_runtime_status.h @@ -39,6 +39,7 @@ #define PTO2_ERROR_ASYNC_COMPLETION_INVALID 101 #define PTO2_ERROR_ASYNC_WAIT_OVERFLOW 102 #define PTO2_ERROR_ASYNC_REGISTRATION_FAILED 103 +#define PTO2_ERROR_READY_QUEUE_OVERFLOW 104 // push into a ready queue found no free slot (full, or window > capacity) static inline int32_t runtime_status_from_error_codes(int32_t orch_error_code, int32_t sched_error_code) { if (orch_error_code != PTO2_ERROR_NONE) { diff --git a/src/a2a3/runtime/host_build_graph/docs/RUNTIME_LOGIC.md b/src/a2a3/runtime/host_build_graph/docs/RUNTIME_LOGIC.md index b5322371bc..8e76a3107e 100644 --- a/src/a2a3/runtime/host_build_graph/docs/RUNTIME_LOGIC.md +++ b/src/a2a3/runtime/host_build_graph/docs/RUNTIME_LOGIC.md @@ -81,6 +81,42 @@ The host/device boundary is POD and position-independent. Fanins are integer producer IDs, not pointers. The only per-slot pointers are rebound to their final device addresses before H2D. +### 3.1 Bounded H2D Upload + +The shared-memory mirror and the prebuilt runtime arena are both sized to ring +capacity (task window, 65536-slot ready-queue pools), but a run only ever touches +`[0, total_tasks)`. The device boots scheduler-only and reads no slot past +`total_tasks`, so the H2D shipped each run is bounded, not capacity-sized. This is +the contract that keeps `bind` proportional to the workload; it is the single most +breakable invariant in the image, so it is stated here in one place. + +- **Shared memory** — the header is zeroed on the host; `descriptors`, `payloads`, + `slot_states` and `completion_flags` are each written per task at submit and + H2D-uploaded bounded to `[0, total_tasks)`. Per-slot reset is init-on-write in + `orch::prepare_task` as each slot is claimed — there is no window-wide reset. + +- **Runtime arena** — three regions, shipped differently: + - The **orchestrator block** (`fanin_seen_epoch` / `scope_tasks` / TensorMap, + ~8.5 MB) is **not shipped at all**: it is host-only dep-computation scratch, + and the AICPU scheduler holds zero references to it. (The scalar + `inline_completed_tasks` the scheduler does read lives in the runtime header, + inside the region that still ships whole.) + - The **big ready-queue slot pools** ship bounded to a prefix of + `min(total_tasks + 1, capacity)`. The `+1` is a **sentinel**: + `PTO2ReadyQueue::pop_batch_tagged` reads one slot past `dequeue_pos` — the slot + at `enqueue_pos`, which reaches `total_tasks` — to detect the empty boundary. A + batched dequeue that finds a stale (too-large) Vyukov sequence there spins + forever, so that one boundary slot must carry its seeded empty sequence. A + lock-free queue's read set extends one element past its write set; bound to the + read set, not the write set. + - Everything else (small early-dispatch queues, the runtime header, the + completion mailbox) ships whole — it is small and carries per-run control state. + +The arena slicing assumes the reservation order of `runtime_reserve_layout` +(`sm_handle → orch → sched(big queues → small queues) → runtime → mailbox`); +`bind_callable_to_runtime_impl` `always_assert`s that order before uploading, so a +future reordering faults loudly instead of shipping a misaligned image. + ## 4. Whole-Graph Capacity The runtime uses one task ring, one graph heap, and one TensorMap pool. They are 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 6780863bad..60bd70d41e 100644 --- a/src/a2a3/runtime/host_build_graph/host/runtime_maker.cpp +++ b/src/a2a3/runtime/host_build_graph/host/runtime_maker.cpp @@ -34,6 +34,8 @@ #include #include +#include +#include #include #include #include @@ -474,8 +476,18 @@ int32_t run_host_orchestration( ) { dep_gen_host_graph_begin_capture(); - std::vector host_sm_buffer(sm_size, 0); - void *host_sm = host_sm_buffer.data(); + // Init-on-write: descriptors, payloads, slot_states and completion_flags are + // each written per task at submit and read only for [0, total_tasks). Zero + // only the fixed-size header here; the per-slot segments are initialized in + // orch::prepare_task and shipped bounded to total_tasks below. + const pto2_sm_layout::PTO2RingSegmentOffsets sm_segs = + pto2_sm_layout::ring_segment_offsets(eff_task_window_sizes[0]); + std::unique_ptr host_sm_buf(new uint8_t[sm_size]); + void *host_sm = host_sm_buf.get(); + std::memset(host_sm, 0, sm_segs.descriptors); + + // Re-point the orchestrator half at the host SM (scheduler keeps device SM). + // init_data_from_layout resets the orchestrator state, so this is safe. if (!rt->orchestrator.init_data_from_layout( layout.orch, host_arena, host_sm, gm_heap, eff_heap_sizes[0], eff_task_window_sizes[0] )) { @@ -524,6 +536,18 @@ int32_t run_host_orchestration( const int32_t total_tasks = pto2_sm_layout::ring_current_task_index_addr(host_sm)->load(std::memory_order_acquire); if (!upload_graph_submissions(runtime, api, *graph_state)) return -1; + // total_tasks sizes the bounded per-segment H2D copies below; a value outside + // [0, task_window] would make those copies read/write out of bounds. + if (total_tasks < 0 || static_cast(total_tasks) > eff_task_window_sizes[0]) { + LOG_ERROR("host-orch: total_tasks %d out of range [0, %" PRIu64 "]", total_tasks, eff_task_window_sizes[0]); + return -1; + } + + // Relocate the host-DDR cross-task pointers to their final DEVICE addresses + // on the host, before the SM and arena leave for the device. Pointers into + // the SM shift by sm_delta; pointers into the arena (fanout adjacency, wiring + // queue) shift by arena_delta. After this both the SM and arena carry device + // addresses, so the device boots scheduler-only. const int64_t sm_delta = static_cast(reinterpret_cast(device_sm)) - static_cast(reinterpret_cast(host_sm)); const int64_t arena_delta = static_cast(reinterpret_cast(device_arena)) - @@ -536,7 +560,23 @@ int32_t run_host_orchestration( return -1; } - if (api->copy_to_device(device_sm, host_sm, sm_size) != 0) { + // Ship only the live prefix of each segment: the device reads no slot past + // total_tasks, so upload header + descriptors[0,N), payloads[0,N), + // slot_states[0,N) and completion_flags[0,N) — never the ring-sized tails. + // header + descriptors[0,N) are contiguous, so that is a single copy. + const uint64_t nt = static_cast(total_tasks); + const uint64_t hdr_desc_bytes = sm_segs.descriptors + nt * sizeof(PTO2TaskDescriptor); + char *host_base = static_cast(host_sm); + char *dev_base = static_cast(device_sm); + if (api->copy_to_device(dev_base, host_base, hdr_desc_bytes) != 0 || + api->copy_to_device(dev_base + sm_segs.payloads, host_base + sm_segs.payloads, nt * sizeof(PTO2TaskPayload)) != + 0 || + api->copy_to_device( + dev_base + sm_segs.slot_states, host_base + sm_segs.slot_states, nt * sizeof(PTO2TaskSlotState) + ) != 0 || + api->copy_to_device( + dev_base + sm_segs.completion_flags, host_base + sm_segs.completion_flags, nt * sizeof(std::atomic) + ) != 0) { LOG_ERROR("host-orch: H2D of populated SM failed"); return -1; } @@ -880,7 +920,63 @@ extern "C" int bind_callable_to_runtime_impl( // *before* it can dereference the image. rt->prebuilt_layout = layout; - int rc_upload = api->copy_to_device(runtime_arena_dev, host_arena.base(), layout.arena_size); + // Skip uploading the orchestrator block (fanin_seen_epoch / scope / tensormap, + // ~8.5 MB): it is host-only dep-computation scratch that the AICPU scheduler + // never reads. Ship [0, orch_start) (sm_handle); skip the host-only orch block; + // then each big ready-queue's slots bounded to a live prefix; then the rest + // (small early queues + runtime header + mailbox) whole. orch_start is the first + // orch reservation (runtime_reserve_layout order: sm_handle -> orch -> sched). + // + // Prefix length = total_tasks + 1, clamped to capacity. A queue takes at most + // total_tasks pushes, so its enqueue_pos never exceeds total_tasks; but + // PTO2ReadyQueue::pop_batch_tagged reads one slot past dequeue_pos — the slot at + // enqueue_pos — to detect the empty boundary, and a batched dequeue that finds a + // stale (too-large) sequence there spins retrying forever. Seeding that sentinel + // slot to its empty sequence makes the boundary read resolve to "empty", so the + // ring-sized tail beyond it is never touched. + const auto &sq = layout.sched; + const size_t orch_start = layout.orch.off_fanin_seen_epoch; + const size_t slot_sz = sizeof(PTO2ReadyQueueSlot); + const uint64_t q_span = static_cast(runtime->host_total_tasks) + 1; + const uint64_t q_prefix = q_span < sq.ready_queue_capacity ? q_span : sq.ready_queue_capacity; + const std::array big_q_slots = { + sq.off_ready_queue_slots[0], sq.off_ready_queue_slots[1], sq.off_ready_queue_slots[2], + sq.off_ready_sync_queue_slots[0], sq.off_ready_sync_queue_slots[1], sq.off_ready_sync_queue_slots[2], + sq.off_dummy_ready_queue_slots, + }; + // Small early queues start right after the big queues; from there to arena end is + // shipped whole (early queues are tiny; runtime header + mailbox are per-run state). + const size_t rest_start = sq.off_early_dispatch_queue_slots[0]; + // The slicing assumes runtime_reserve_layout's order: sm_handle -> orch -> + // sched (big queues -> small queues) -> runtime -> mailbox. Pin it so a future + // reordering faults here instead of silently shipping a misaligned image. + always_assert(orch_start < big_q_slots.front()); + for (size_t qi = 1; qi < big_q_slots.size(); qi++) { + always_assert(big_q_slots[qi - 1] < big_q_slots[qi]); + } + always_assert(rest_start >= big_q_slots.back() + sq.ready_queue_capacity * slot_sz); + char *arena_host = static_cast(host_arena.base()); + char *arena_dev = static_cast(runtime_arena_dev); + // Seed the big-queue slots [0, q_prefix) that the build skipped (it did headers + // only). Matches ready_queue_init_data_from_layout: sequence=i, empty slot_state. + for (size_t slots_off : big_q_slots) { + auto *slots = reinterpret_cast(arena_host + slots_off); + for (uint64_t i = 0; i < q_prefix; i++) { + slots[i].sequence.store(static_cast(i), std::memory_order_relaxed); + slots[i].slot_state = nullptr; + } + } + int rc_upload = api->copy_to_device(arena_dev, arena_host, orch_start); + for (size_t slots_off : big_q_slots) { + if (rc_upload != 0) { + break; + } + rc_upload = api->copy_to_device(arena_dev + slots_off, arena_host + slots_off, q_prefix * slot_sz); + } + if (rc_upload == 0) { + rc_upload = + api->copy_to_device(arena_dev + rest_start, arena_host + rest_start, layout.arena_size - rest_start); + } if (rc_upload != 0) { LOG_ERROR("Failed to rtMemcpy prebuilt runtime arena to device (rc=%d)", rc_upload); return -1; diff --git a/src/a2a3/runtime/host_build_graph/runtime/orchestrator_core/pto_orchestrator.cpp b/src/a2a3/runtime/host_build_graph/runtime/orchestrator_core/pto_orchestrator.cpp index cd5913b3f8..0510e76989 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/orchestrator_core/pto_orchestrator.cpp +++ b/src/a2a3/runtime/host_build_graph/runtime/orchestrator_core/pto_orchestrator.cpp @@ -866,6 +866,14 @@ static bool prepare_task( out->task = &orch->sm_header->ring.task_descriptors[out->alloc_result.slot]; out->payload = &orch->sm_header->ring.task_payloads[out->alloc_result.slot]; + // Init-on-write: this slot's dynamic scheduling fields and completion flag are + // initialized here, as the orchestrator claims the slot. whole-graph-resident + // hbg claims slots [0, total_tasks) exactly once and the device reads no slot + // past total_tasks, so this claim-time write is the only per-slot SM reset and + // the unclaimed tail is neither initialized nor read. + out->slot_state->reset_for_reuse(); + orch->sm_header->ring.completion_flags[out->alloc_result.slot].store(0, std::memory_order_relaxed); + graph_record_begin_task(orch, out->task_id); out->payload->prefetch(args.tensor_count(), args.scalar_count()); @@ -884,7 +892,7 @@ static bool prepare_task( // early-dispatch fields) is initialized in PTO2TaskPayload::init, the // single payload-init point, which runs before Orch-side wiring publish. - // Fields already zeroed by reset_for_reuse() at slot init: + // Fields already zeroed by the reset_for_reuse() above: // wake_list_head=nullptr, next_in_wake_list=nullptr, // any_subtask_deferred=false, completed_subtasks=0, next_block_idx=0 // Fields immutable after RingSchedState::init(): @@ -900,8 +908,8 @@ static bool prepare_task( out->slot_state->task_kind = active_mask ? TaskKind::KERNEL : TaskKind::DUMMY; // Reclaim gate: seed last_consumer to self, so a producer with no consumers // is retirable once completed_watermark >= its own id. Each fanin edge bumps - // it in append_fanin_or_fail. completion_flags for this slot are already 0 - // (zeroed once at init; whole-graph-resident hbg never reuses a slot). + // it in append_fanin_or_fail. completion_flags for this slot were cleared + // above (whole-graph-resident hbg never reuses a slot). out->slot_state->last_consumer_local_id = static_cast(out->task_id.local()); // payload.fanin_count is set in submit_task_common's STEP 6. scope_tasks_push(orch, out->slot_state); diff --git a/src/a2a3/runtime/host_build_graph/runtime/pto_runtime2_types.h b/src/a2a3/runtime/host_build_graph/runtime/pto_runtime2_types.h index 8e02d611ea..a109e0685b 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/pto_runtime2_types.h +++ b/src/a2a3/runtime/host_build_graph/runtime/pto_runtime2_types.h @@ -538,9 +538,9 @@ struct alignas(64) PTO2TaskSlotState { bool has_any_subtask_deferred() const { return any_subtask_deferred.load(std::memory_order_acquire); } /** - * Reset dynamic scheduling fields to their pristine values. Runs once per - * slot at init (pto_shared_memory.cpp) — whole-graph-resident hbg has no - * execution-time slot recycle. Skips payload/task (bound once) and + * Reset dynamic scheduling fields to their pristine values. Called once per + * slot as the orchestrator claims it in prepare_task — whole-graph-resident + * hbg has no execution-time slot recycle. Skips payload/task (bound once) and * task_state (the orchestrator sets PENDING when it populates the slot). * wake_list_head starts nullptr (open for registration), NOT SENTINEL. * Graph-affine replay passes preserve_graph_binding=true because its node diff --git a/src/a2a3/runtime/host_build_graph/runtime/pto_shared_memory.h b/src/a2a3/runtime/host_build_graph/runtime/pto_shared_memory.h index bd404c3bc2..cadb1f582a 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/pto_shared_memory.h +++ b/src/a2a3/runtime/host_build_graph/runtime/pto_shared_memory.h @@ -102,7 +102,8 @@ struct alignas(64) PTO2SharedMemoryRingHeader { // Polling-completion state (device-addressed array, one byte per slot). // 0 = pending, 1 = task fully COMPLETED. Writer = the task's completer at // on_mixed_task_complete; reader = consumer fanin polling (is_completion_flag_set). - // Zeroed host-side at init. Indexed by local_id & task_window_mask. + // Cleared per-slot in orch::prepare_task as each slot is claimed. Indexed by + // local_id & task_window_mask. std::atomic *completion_flags; bool is_completion_flag_set(int32_t local_id, std::memory_order order = std::memory_order_acquire) const { 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 baa54fe788..33445af46c 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 @@ -395,7 +395,15 @@ size_t ready_queue_reserve_layout(DeviceArena &arena, uint64_t capacity); // arena.region_ptr(slots_off) only to address the slot array for writes; // does NOT store the pointer in `queue->slots`. Call // `ready_queue_wire_arena_pointers` afterwards to set the field itself. -bool ready_queue_init_data_from_layout(PTO2ReadyQueue *queue, DeviceArena &arena, size_t slots_off, uint64_t capacity); +// slot_init_count bounds the per-slot Vyukov-sequence init loop (defaults to the +// full capacity). host_build_graph passes 0 for the big ready queues and seeds +// only [0, total_tasks) of them after orchestration, since the device reads no +// slot past its push count; the queue header (capacity/mask/positions) is always +// initialized. +bool ready_queue_init_data_from_layout( + PTO2ReadyQueue *queue, DeviceArena &arena, size_t slots_off, uint64_t capacity, + uint64_t slot_init_count = ~static_cast(0) +); // Stores queue->slots = arena.region_ptr(slots_off). Idempotent. void ready_queue_wire_arena_pointers(PTO2ReadyQueue *queue, DeviceArena &arena, size_t slots_off); void ready_queue_destroy(PTO2ReadyQueue *queue); @@ -494,13 +502,26 @@ struct PTO2SchedulerState { return; } PTO2ResourceShape shape = slot_state->active_mask.to_shape(); + bool pushed; if (shape == PTO2ResourceShape::DUMMY || (slot_state->task_attrs.has_predicate() && !slot_state->payload->predicate.pass())) { - dummy_ready_queue.push(slot_state); + pushed = dummy_ready_queue.push(slot_state); } else if (slot_state->task_attrs.requires_sync_start()) { - ready_sync_queues[static_cast(shape)].push(slot_state); + pushed = ready_sync_queues[static_cast(shape)].push(slot_state); } else { - ready_queues[static_cast(shape)].push(slot_state); + pushed = ready_queues[static_cast(shape)].push(slot_state); + } + // A queue is sized for the whole task window and each task is routed to one + // queue exactly once, so push cannot legitimately fail. A false return means + // the target slot fell outside the shipped prefix, or the window genuinely + // exceeds queue capacity — either way the task is dropped and the run would + // otherwise stall. Latch a named error so it surfaces as READY_QUEUE_OVERFLOW + // rather than an anonymous forward-progress timeout. + 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 + ); } } diff --git a/src/a2a3/runtime/host_build_graph/runtime/shared/pto_runtime2_init.cpp b/src/a2a3/runtime/host_build_graph/runtime/shared/pto_runtime2_init.cpp index 9f5108f040..7074c7ebf8 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/shared/pto_runtime2_init.cpp +++ b/src/a2a3/runtime/host_build_graph/runtime/shared/pto_runtime2_init.cpp @@ -55,7 +55,9 @@ size_t ready_queue_reserve_layout(DeviceArena &arena, uint64_t capacity) { return arena.reserve(capacity * sizeof(PTO2ReadyQueueSlot), PTO2_ALIGN_SIZE); } -bool ready_queue_init_data_from_layout(PTO2ReadyQueue *queue, DeviceArena &arena, size_t slots_off, uint64_t capacity) { +bool ready_queue_init_data_from_layout( + PTO2ReadyQueue *queue, DeviceArena &arena, size_t slots_off, uint64_t capacity, uint64_t slot_init_count +) { // Address the slots region for data writes without storing the pointer in // queue->slots — that field is set by ready_queue_wire_arena_pointers. auto *slots_arena = static_cast(arena.region_ptr(slots_off)); @@ -64,7 +66,8 @@ bool ready_queue_init_data_from_layout(PTO2ReadyQueue *queue, DeviceArena &arena queue->enqueue_pos.store(0, std::memory_order_relaxed); queue->dequeue_pos.store(0, std::memory_order_relaxed); - for (uint64_t i = 0; i < capacity; i++) { + const uint64_t n = slot_init_count < capacity ? slot_init_count : capacity; + for (uint64_t i = 0; i < n; i++) { slots_arena[i].sequence.store((int64_t)i, std::memory_order_relaxed); slots_arena[i].slot_state = nullptr; } @@ -92,9 +95,9 @@ bool PTO2SchedulerState::RingSchedState::init_data_from_layout(void *sm_dev_base last_task_alive = 0; advance_lock.store(0, std::memory_order_relaxed); - // Per-slot SM-side initialization (reset_for_reuse + fanin_count/active_mask - // zero) lives in PTO2SharedMemoryHandle::init_header_per_ring so the AICPU - // performs it during SM reset; host prebuilt-arena init skips SM access here. + // Per-slot SM-side initialization (reset_for_reuse + active_mask, and clearing + // the completion flag) happens init-on-write in orch::prepare_task as each slot + // is claimed; host prebuilt-arena init skips SM access here. return true; } @@ -138,21 +141,24 @@ bool PTO2SchedulerState::init_data_from_layout( } for (int i = 0; i < PTO2_NUM_RESOURCE_SHAPES; i++) { + // slot_init_count = 0: big-queue slots are seeded post-orchestration, + // bounded to [0, total_tasks) (see runtime_maker.cpp). if (!ready_queue_init_data_from_layout( - &sched->ready_queues[i], arena, layout.off_ready_queue_slots[i], layout.ready_queue_capacity + &sched->ready_queues[i], arena, layout.off_ready_queue_slots[i], layout.ready_queue_capacity, 0 )) { return false; } } for (int i = 0; i < PTO2_NUM_RESOURCE_SHAPES; i++) { if (!ready_queue_init_data_from_layout( - &sched->ready_sync_queues[i], arena, layout.off_ready_sync_queue_slots[i], layout.ready_queue_capacity + &sched->ready_sync_queues[i], arena, layout.off_ready_sync_queue_slots[i], layout.ready_queue_capacity, + 0 )) { return false; } } if (!ready_queue_init_data_from_layout( - &sched->dummy_ready_queue, arena, layout.off_dummy_ready_queue_slots, layout.ready_queue_capacity + &sched->dummy_ready_queue, arena, layout.off_dummy_ready_queue_slots, layout.ready_queue_capacity, 0 )) { return false; } diff --git a/src/a2a3/runtime/host_build_graph/runtime/shared/pto_shared_memory.cpp b/src/a2a3/runtime/host_build_graph/runtime/shared/pto_shared_memory.cpp index fcff0a154c..80b32e8e19 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/shared/pto_shared_memory.cpp +++ b/src/a2a3/runtime/host_build_graph/runtime/shared/pto_shared_memory.cpp @@ -178,22 +178,11 @@ void PTO2SharedMemoryHandle::init_header_per_ring( header->sched_error_code.store(PTO2_ERROR_NONE, std::memory_order_relaxed); header->sched_error_thread.store(-1, std::memory_order_relaxed); - // Per-ring slot_states reset. Previously lived in - // PTO2SchedulerState::RingSchedState::init(), but it writes into - // ring->slot_states[] which is SM-side storage — keeping it here lets - // host-side prebuilt-arena init skip all SM dereferences. - // reset_for_reuse() prepares dynamic fanout/refcount fields so the first - // submit doesn't need an explicit reset. - auto &ring = header->ring; - for (uint64_t i = 0; i < task_window_sizes[0]; i++) { - ring.slot_states[i].reset_for_reuse(); - ring.slot_states[i].active_mask = ActiveMask{}; - } - - // Polling completion flags: 0 = pending. Shared memory is not guaranteed - // zero on device; stale non-zero bytes would make consumers observe a - // producer as already completed. Zero the whole per-ring array once. - __builtin_memset((void *)ring.completion_flags, 0, task_window_sizes[0] * sizeof(std::atomic)); + // Per-slot init (slot_states.reset_for_reuse() + active_mask, and clearing the + // completion flag) happens init-on-write in orch::prepare_task as each slot + // [0, total_tasks) is claimed, so the SM init/upload cost tracks the task + // count, not ring capacity. The device reads no slot past total_tasks, so the + // unclaimed tail is left uninitialized. } // ============================================================================= diff --git a/src/a2a3/runtime/host_build_graph/runtime/shared/pto_tensormap.cpp b/src/a2a3/runtime/host_build_graph/runtime/shared/pto_tensormap.cpp index d10702ef21..339def0ab6 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/shared/pto_tensormap.cpp +++ b/src/a2a3/runtime/host_build_graph/runtime/shared/pto_tensormap.cpp @@ -94,13 +94,11 @@ bool PTO2TensorMap::init_data_from_layout(const PTO2TensorMapLayout &layout, Dev // The pool's persistent invariant after init is "bucket_index == -1 means // not linked", set explicitly below. memset(entry_pool_arena, 0, static_cast(pool_size) * sizeof(PTO2TensorMapEntry)); + // The memset already zeroed every field (all four link pointers -> nullptr, + // producer_task_id -> {}); only bucket_index needs its non-zero "not linked" + // marker, so the per-entry loop writes just that. for (int32_t i = 0; i < pool_size; i++) { entry_pool_arena[i].bucket_index = -1; - entry_pool_arena[i].next_in_bucket = nullptr; - entry_pool_arena[i].prev_in_bucket = nullptr; - entry_pool_arena[i].next_in_task = nullptr; - entry_pool_arena[i].prev_in_task = nullptr; - entry_pool_arena[i].producer_task_id = PTO2TaskId{}; } // free_entry_list: zeroed (was calloc'd before); contents become meaningful diff --git a/tests/ut/cpp/CMakeLists.txt b/tests/ut/cpp/CMakeLists.txt index d16d0c1ffe..5601273c63 100644 --- a/tests/ut/cpp/CMakeLists.txt +++ b/tests/ut/cpp/CMakeLists.txt @@ -741,11 +741,27 @@ add_a2a3_test(test_acl_hal_device common/test_acl_hal_device.cpp) # PTO2 runtime-linked tests 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) 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) add_a2a3_hbg_runtime_test(test_hbg_tensor_access a2a3/test_hbg_tensor_access.cpp) add_a2a3_hbg_runtime_test(test_hbg_core_tracker common/test_hbg_core_tracker.cpp) +# PTO2ReadyQueue is header-only (pto_scheduler.h); this test seeds a vector-backed +# queue directly and needs no runtime .cpp beyond the shared test stubs. +add_a2a3_hbg_runtime_test(test_hbg_ready_queue a2a3/test_hbg_ready_queue.cpp) +# The submit-poison test drives the real orchestrator submit path, so it links the +# orchestrator + shared runtime out-of-line members (mirrors a2a3_rt_objs for hbg). +add_a2a3_hbg_runtime_test(test_hbg_submit_poison a2a3/test_hbg_submit_poison.cpp) +target_sources(test_hbg_submit_poison 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_core_tracker common/test_hbg_core_tracker.cpp) add_a2a3_hbg_runtime_test(test_graph_cache a2a3/test_graph_cache.cpp) target_sources(test_graph_cache PRIVATE diff --git a/tests/ut/cpp/a2a3/test_hbg_ready_queue.cpp b/tests/ut/cpp/a2a3/test_hbg_ready_queue.cpp new file mode 100644 index 0000000000..b850da9c0b --- /dev/null +++ b/tests/ut/cpp/a2a3/test_hbg_ready_queue.cpp @@ -0,0 +1,130 @@ +/* + * 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. + * ----------------------------------------------------------------------------------------------------------- + */ +/** + * Regression tests for the host_build_graph bounded ready-queue seeding contract. + * + * runtime_maker.cpp seeds and H2D-uploads each ready queue's slots bounded to a + * prefix of min(total_tasks + 1, capacity). The +1 is a sentinel: PTO2ReadyQueue's + * batched dequeue (pop_batch_tagged) reads one slot past dequeue_pos — the slot at + * enqueue_pos, which reaches total_tasks — to detect the empty boundary. If that + * boundary slot carries a stale, too-large Vyukov sequence (as it does on reuse of + * the persistent device arena when it was never re-seeded), pop_batch computes + * diff > 0 and spins forever, and the ready task is never dequeued — the + * intermittent SCHEDULER_TIMEOUT this bound once produced. + * + * These tests pin the queue-side semantics the seeding relies on, so a future + * bound to exactly total_tasks (dropping the sentinel) fails in-tree. + */ + +#include + +#include +#include +#include +#include +#include + +#include "scheduler/pto_scheduler.h" + +namespace { + +constexpr uint64_t CAP = 64; // ready-queue capacity (power of two), >> N +constexpr int N = 5; // tasks pushed; the empty boundary lands at slot N + +// A prior run pops slot i by advancing its sequence to i + capacity. That value is +// "too large" for the next run's boundary read (diff > 0) and is exactly what makes +// pop_batch spin when the boundary slot is not re-seeded. +int64_t stale_seq(uint64_t i) { return static_cast(i + CAP); } + +// Mirrors ready_queue_init_data_from_layout for a vector-backed queue: sets the +// header (capacity / mask / positions) and seeds slots [0, seeded) with the empty +// Vyukov sequence (slot i -> sequence i). Slots [seeded, capacity) are left as the +// caller planted them, so a test can model an un-reseeded stale tail. +void seed_queue(PTO2ReadyQueue &q, std::vector &backing, uint64_t seeded) { + q.slots = backing.data(); + q.capacity = CAP; + q.mask = CAP - 1; + q.enqueue_pos.store(0, std::memory_order_relaxed); + q.dequeue_pos.store(0, std::memory_order_relaxed); + for (uint64_t i = 0; i < seeded; i++) { + backing[i].sequence.store(static_cast(i), std::memory_order_relaxed); + backing[i].slot_state = nullptr; + } +} + +// Fill the whole backing array with a stale, too-large sequence before seeding, so +// any slot the seed does not cover models an un-reseeded prior-run slot. +void plant_stale_tail(std::vector &backing) { + for (uint64_t i = 0; i < CAP; i++) { + backing[i].sequence.store(stale_seq(i), std::memory_order_relaxed); + backing[i].slot_state = nullptr; + } +} + +} // namespace + +// With the sentinel (seed [0, N+1)), pop_batch terminates and returns exactly the +// pushed tasks even when the entire tail beyond the sentinel holds stale sequences. +TEST(HbgReadyQueueSentinel, PopBatchTerminatesWithStaleTail) { + std::vector backing(CAP); + plant_stale_tail(backing); + + PTO2ReadyQueue q; + seed_queue(q, backing, static_cast(N) + 1); // N task slots + one empty sentinel at slot N + + std::vector items(N); + for (int i = 0; i < N; i++) { + ASSERT_TRUE(q.push(&items[i])); + } + + PTO2TaskSlotState *out[N + 8]; + const int got = q.pop_batch(out, N + 8); // scans the sentinel at slot N, breaks cleanly + EXPECT_EQ(got, N); + for (int i = 0; i < N; i++) { + EXPECT_EQ(out[i], &items[i]); + } +} + +// Control: bounding to exactly total_tasks (no sentinel) leaves the boundary slot +// at enqueue_pos holding a stale too-large sequence, and pop_batch spins on it. +// Demonstrate the hang under a timeout, then write the empty sentinel sequence the +// fix would have seeded and confirm the SAME spinning call unblocks and returns N. +TEST(HbgReadyQueueSentinel, WithoutSentinelPopBatchSpinsUntilBoundarySeeded) { + std::vector backing(CAP); + plant_stale_tail(backing); + + PTO2ReadyQueue q; + seed_queue(q, backing, static_cast(N)); // seed only [0, N): slot N stays stale + + std::vector items(N); + for (int i = 0; i < N; i++) { + ASSERT_TRUE(q.push(&items[i])); + } + + PTO2TaskSlotState *out[N + 8]; + std::atomic got{-1}; + std::thread popper([&] { + got.store(q.pop_batch(out, N + 8), std::memory_order_release); + }); + + // The stale boundary slot makes pop_batch spin: it must not complete promptly. + std::this_thread::sleep_for(std::chrono::milliseconds(200)); + EXPECT_EQ(got.load(std::memory_order_acquire), -1); + + // Apply what the +1 sentinel writes: the empty sequence at the boundary slot. + // The spinning pop_batch now sees diff < 0 there, breaks, and returns. + backing[N].sequence.store(static_cast(N), std::memory_order_release); + popper.join(); + EXPECT_EQ(got.load(std::memory_order_acquire), N); + for (int i = 0; i < N; i++) { + EXPECT_EQ(out[i], &items[i]); + } +} diff --git a/tests/ut/cpp/a2a3/test_hbg_submit_poison.cpp b/tests/ut/cpp/a2a3/test_hbg_submit_poison.cpp new file mode 100644 index 0000000000..3d764d6674 --- /dev/null +++ b/tests/ut/cpp/a2a3/test_hbg_submit_poison.cpp @@ -0,0 +1,180 @@ +/* + * 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. + * ----------------------------------------------------------------------------------------------------------- + */ +/** + * Poison test for the "every device-read SM field is written at submit" contract. + * + * host_build_graph no longer zero-fills the shared-memory task window (init-on-write): + * init_header_per_ring writes only the header, and each slot's device-read fields are + * written per task at submit (prepare_task + submit_task_common + PTO2TaskPayload::init). + * Nothing else clears the window, so a device-read field a submit forgets to write would + * read as 0 only by allocator accident — passing every zero-backed test and failing + * non-deterministically on device. + * + * This test fills the whole per-slot window with a 0xAA poison byte before submitting a + * representative mix, then asserts that for every claimed slot [0, total_tasks) the + * device-read fields carry real values, not poison. Add a device-read field and forget + * its submit-path write, and this fails in-tree. + */ + +#include + +#include +#include +#include + +#include "utils/device_arena.h" +#include "pto_orchestrator.h" +#include "pto_shared_memory.h" + +namespace { + +constexpr uint8_t POISON = 0xAA; +// A void* / int32 whose bytes are all 0xAA — what an unwritten field would read as. +void *const POISON_PTR = reinterpret_cast(static_cast(0xAAAAAAAAAAAAAAAAULL)); + +} // namespace + +class HbgSubmitPoisonTest : public ::testing::Test { +protected: + DeviceArena sm_arena; + DeviceArena runtime_arena; + PTO2SharedMemoryHandle *sm_handle = nullptr; + PTO2OrchestratorState orch{}; + PTO2SchedulerState sched{}; + PTO2OrchestratorLayout orch_layout{}; + PTO2SchedulerLayout sched_layout{}; + 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); + } + + void TearDown() override { + orch.destroy(); + sched.destroy(); + runtime_arena.release(); + sm_arena.release(); + } + + // Fill the per-slot window (descriptors / payloads / slot_states / completion_flags) + // with poison. init_header_per_ring wrote only the header, so this is the state the + // window is in before any submit writes it — modelling the never-zeroed device SM. + void poison_window() { + auto &ring = sm_handle->header->ring; // host_build_graph is single-ring + const size_t n = static_cast(ring.task_window_mask) + 1; + std::memset(ring.task_descriptors, POISON, n * sizeof(PTO2TaskDescriptor)); + std::memset(ring.task_payloads, POISON, n * sizeof(PTO2TaskPayload)); + std::memset(ring.slot_states, POISON, n * sizeof(PTO2TaskSlotState)); + std::memset(ring.completion_flags, POISON, n * sizeof(std::atomic)); + } +}; + +TEST_F(HbgSubmitPoisonTest, EveryDeviceReadFieldIsWrittenOverPoison) { + poison_window(); + orch.begin_scope(); + + // 1. Zero-fanin root: a real mixed (AIV0) task with an output tensor and a scalar. + std::vector create_infos; + create_infos.reserve(4); + uint32_t shape[] = {16}; + CoreTaskArgs root_args; + create_infos.emplace_back(shape, 1, DataType::FLOAT32); + root_args.add_output(create_infos.back()); + root_args.add_scalar(static_cast(42)); + MixedKernels root_mixed{}; + root_mixed.aiv0_kernel_id = 0; + TaskOutputTensors root = orch.submit_task(root_mixed, root_args); + ASSERT_TRUE(root.task_id().is_valid()); + + // 2. Multi-fanin dummy consumer (duplicate dep deduped to one fanin). + PTO2TaskId deps[] = {root.task_id(), root.task_id()}; + CoreTaskArgs consumer_args; + consumer_args.set_dependencies(deps, 2); + TaskOutputTensors consumer = orch.submit_dummy_task(consumer_args); + ASSERT_TRUE(consumer.task_id().is_valid()); + + // 3. Hidden-alloc convenience (allocates an output, no kernel). + CoreTaskArgs alloc_args; + create_infos.emplace_back(shape, 1, DataType::FLOAT32); + alloc_args.add_output(create_infos.back()); + TaskOutputTensors allocated = orch.alloc_tensors(alloc_args); + ASSERT_TRUE(allocated.task_id().is_valid()); + + // 4. Plain dummy. + CoreTaskArgs plain_args; + TaskOutputTensors plain = orch.submit_dummy_task(plain_args); + ASSERT_TRUE(plain.task_id().is_valid()); + + orch.end_scope(); + + auto &ring = sm_handle->header->ring; + const int32_t total = ring.fc.current_task_index.load(std::memory_order_acquire); + ASSERT_GE(total, 4); + + // Every claimed slot's device-read fields must carry real values, not poison. + for (int32_t local = 0; local < total; local++) { + SCOPED_TRACE(testing::Message() << "slot local_id=" << local); + const int32_t slot = ring.get_slot_by_task_id(local); + const PTO2TaskDescriptor &desc = ring.task_descriptors[slot]; + const PTO2TaskPayload &pl = ring.task_payloads[slot]; + const PTO2TaskSlotState &st = ring.slot_states[slot]; + + // Descriptor: the task id is written to this exact local id. + EXPECT_EQ(desc.task_id.local(), static_cast(local)); + // task_state is written at submit (reset_for_reuse skips it): PENDING for a + // dispatchable task, COMPLETED for a pre-completed hidden-alloc. Either way a + // real enum, never poison. + const PTO2TaskState state = st.task_state.load(std::memory_order_relaxed); + EXPECT_TRUE(state == PTO2_TASK_PENDING || state == PTO2_TASK_COMPLETED); + // Completion flag is written to a real 0/1 (pending vs pre-completed), not a + // poison byte (0xAA). + const uint8_t cflag = ring.completion_flags[slot].load(std::memory_order_relaxed); + EXPECT_LE(cflag, uint8_t{1}); + // Payload counts are real, not the poison bit pattern. + EXPECT_GE(pl.fanin_count, 0); + EXPECT_LE(pl.fanin_count, PTO2_MAX_FANIN); + EXPECT_GE(pl.tensor_count, 0); + EXPECT_GE(pl.scalar_count, 0); + // predicate.op is a dispatch-time field, read only for tasks the device + // actually dispatches. submit_task_common writes it (NONE when unset); a + // pre-completed hidden-alloc is never dispatched, so it does not. + if (state == PTO2_TASK_PENDING) { + EXPECT_LE(static_cast(pl.predicate.op), static_cast(PredicateOp::LE)); + } + } + + // Field-specific coverage on the real task: tensors, scalar, packed output buffer. + const PTO2TaskDescriptor &root_desc = ring.task_descriptors[ring.get_slot_by_task_id(root.task_id().local())]; + const PTO2TaskPayload &root_pl = ring.task_payloads[ring.get_slot_by_task_id(root.task_id().local())]; + EXPECT_EQ(root_pl.tensor_count, 1); + EXPECT_EQ(root_pl.scalar_count, 1); + EXPECT_NE(root_desc.packed_buffer_base, POISON_PTR); + EXPECT_NE(root_desc.packed_buffer_base, nullptr); + EXPECT_EQ(root_desc.kernel_id[static_cast(PTO2SubtaskSlot::AIV0)], 0); + + // The consumer's fanin is written: two duplicate deps dedupe to one. + const PTO2TaskPayload &cons_pl = ring.task_payloads[ring.get_slot_by_task_id(consumer.task_id().local())]; + EXPECT_EQ(cons_pl.fanin_count, 1); + EXPECT_EQ(cons_pl.fanin_local_ids[0], static_cast(root.task_id().local())); +}