diff --git a/docs/design/hbg-transitive-reduction.md b/docs/design/hbg-transitive-reduction.md new file mode 100644 index 0000000000..9be0fb31b4 --- /dev/null +++ b/docs/design/hbg-transitive-reduction.md @@ -0,0 +1,209 @@ +# HBG: host-side transitive reduction of the recorded task DAG + +**Status**: design +**Target**: `host_build_graph` (a2a3 + a5, mirrored) +**Baseline**: upstream/main `66ba5c4a` + +## Problem + +`host_build_graph` records a task DAG into a `GraphRecording` and packs it into +the device-resident Definition's fanin CSR. Every HBG edge carries exactly one +semantics — **ordering/readiness** — because Graph Execution is +whole-graph-resident: node slots are never reclaimed or rebound mid-run +(`scheduler.h`: *"on_task_release is gone… host-orch never reclaimed slots on +device"*), and outputs live until the execution object is torn down. There is +no per-edge resource-lifetime semantic to preserve. + +DAGs from generated orchestrations (qwen3-class decode: ~5240 tasks; dsv4: +43-layer MoE) contain redundant ordering edges: a direct edge P→C whose +ordering is already implied by a longer path P→…→C through other nodes of C's +own fanin. On TMR these could only be cleared 1-hop and at a measured per-submit +cost (PR #1830: Orch +2.4–3.8%, no Effective gain on chain-shaped corpora). HBG +pays nothing per submit — the whole graph exists on the host at once — so the +reduction can be exact and run once per Definition build. + +Cost today, per redundant edge, paid on the device at every execution of the +Definition: + +- `graph_first_unmet_producer` scans the consumer's CSR row from the front on + every wake-list re-registration (`scheduler.h`); +- `drain_graph_wake_list` re-runs that scan for every waiter on the producer's + wake list each time the producer completes; +- the `fanout_offsets`/`fanout_indices` section of the image grows with the + raw edge count. + +## Non-goals + +- No change to what `deps.json` records: dep_gen capture is wired to the + *recording* path (annotate hooks in `record_submit`), not to the packed + Definition. `deps.json` keeps the as-constructed edge set — same convention + the TMR side documents in `docs/dfx/dep-gen.md` ("Flags are the + as-constructed set"). +- No cross-Definition reduction. #1968's per-block Definitions connect through + the outer Graph shell's external dependencies; a diamond spanning two blocks + is not visible to either block's recording. Intra-Definition only, stated as + a scope limit here. +- No behavior gate / env knob (per `.claude/rules/env-macro-gating.md`): the + reduction preserves reachability, and reachability is the only semantics an + HBG edge has, so it is unconditionally correct. + +## Design + +### Where + +One pass in `graph_build_definition` (`orchestrator_core/orchestrator.cpp`), +after the size-computation loop and **before** the CSR fill loop, operating on +`recording.nodes[*].fanin_offset/fanin_count` + `recording.internal_fanins`. +The recording itself is left untouched — dep_gen reads it afterwards (same +function in a different call chain? no: dep_gen hooks live in `record_submit`, +which has already run by build time), and the packed image is what the device +consumes. Reduction is a build-time projection of the recording, not a +mutation of it. + +`graph_build_definition` runs once per recording on the recorder thread +(`graph_end`, behind `ORCH_PHASE(BuildDefinition)`), and the resulting +Definition is cached and shared across invocations (#1968 single-upload). +Steady-state replay cost of the reduction is therefore zero. + +### Algorithm + +Nodes are recorded in topological order already — the CSR fill loop checks +`producer >= i → return false` — so no topo-sort is needed. + +Exact transitive reduction on a DAG via the standard reachability method, +iterated in reverse topological order: + +```text +state: bitmask reach[N][N/64] # GRAPH_MAX_NODES = 1024 → 1024 × 16 words = 128 KB scratch +for i = N-1 .. 0: # reverse topo order + for each producer p in row(i): # after reduction of later rows + if reach[i] ⊇ contains(p): # p already reachable via another path + drop edge p→i + else: + reach[i] |= reach[p] + reach[i].set(p) +``` + +Correctness: processing in reverse topological order means `reach[p]` is final +(a node's descendants are all later nodes) when row *i* consumes it. An edge +p→i is dropped iff `p` is reachable from some *other* producer row-entry of *i* +through its (already reduced) descendant sets — i.e. the edge is a transitive +shortcut. Reachability is preserved by construction; since ordering is the only +edge semantics, the reduced graph admits exactly the same executions. + +Scratch: `GRAPH_MAX_NODES = 1024` nodes → 1024 rows × 16 × `uint64` = 128 KB, +host-side `std::vector`, freed after the pass. At the median Definition size +(seven Definitions over 43 layers, #1968 — hundreds of nodes each) this is far +below the recording itself. For `node_count` below ~256 the dense bitset is +allocated to the actual row count, not the cap. + +Complexity: O(V · E / 64) word-ORs — for 1024 nodes with 8 edges/node average, +~130 K word operations, well under a millisecond on the host. This is the +"persistent per-slot ancestor closure" the TMR investigation +(`docs/investigations/2026-08-tmr-transitive-reduction-depth.md`) could not +afford on the AICPU; on the host at build time it is free. + +### What changes in the packed image + +- `definition.edge_count` shrinks to the reduced count. +- `fanin_offsets/fanin_indices` rows drop the removed producer entries. +- `fanout_offsets/fanout_indices` are rebuilt from the reduced edge set (they + are pure derived data: `bind_graph_topology` only validates them; no + scheduler code reads them — the wake machinery walks fanin via + `graph_first_unmet_producer`). +- `root_count` is unaffected (a root has no producers; reduction cannot create + or destroy one). +- The Definition content hash changes for graphs with redundant edges — + expected: the hash identifies image bytes, and the image legitimately + changed. Existing cached Definitions (from a previous process run) are + keyed by `full_key` + hash; a rebuild with different bytes is a different + Definition object, uploaded once. No migration concern. + +### What must NOT change + +- The recording (`GraphRecording`), hence `deps.json`, hence the dep_gen + differential gate and every downstream tool (deps_viewer, swimlane join). +- `GraphRecordedNode::fanin_count` on the recording side (the reduction works + on a local per-node view or a copy of the ranges — see Implementation). +- Boundary/external dependency handling: producers outside the recording + window were already dropped at record time; nothing to reduce against. + +## Implementation sketch + +In `graph_build_definition`, between the counting loop and the layout calls: + +```cpp +// Returns the reduced edge list as (producer, consumer) pairs in row order, +// or empty when the recording has no redundant shortcut edge. +std::vector graph_reduce_transitive_edges(const GraphRecording &recording); +``` + +- The fill loop then reads the reduced rows instead of + `recording.internal_fanins[fanin_offset + f]` directly. Simplest shape: the + helper returns a **new** flat producer array plus per-node offsets, and the + existing loop consumes those; when the helper finds nothing to drop it + returns the identity projection so the loop is unchanged in shape. +- `total_fanins` for the layout pass is taken from the reduced count. +- Log one `LOG_DEBUG` line with nodes/edges before→after when the drop count + is nonzero (gated to the existing debug channel — cold path, not per-submit). + +Mirrored identically in `src/a5/.../orchestrator_core/orchestrator.cpp` +(the two files are byte-identical today; the edit applies verbatim). + +## Testing + +1. **Unit test** (`tests/ut/cpp/common/test_hbg_graph_reduction.cpp`, linked + like `test_hbg_graph_cache`): construct small recordings by hand — + - the diamond `A→B→C` + `A→C`: `A→C` is dropped, `A→B`, `B→C` kept; + - a 3-hop chain + shortcut `A→B→C→D` + `A→D`: `A→D` dropped by the + *transitive* path, proving arbitrary depth (the case TMR could not do); + - two independent producers to one consumer: both kept; + - a diamond where the shortcut is the *only* path from one producer: + kept (no false drop); + - duplicate edges / self-edge guards (self-edges are rejected earlier by + `producer >= i`). +2. **bind_graph_topology invariants**: the reduced image must still pass the + existing CSR validation (offsets monotone, indices < consumer, edge_count + consistent both sides, root_count unchanged). The unit test asserts this by + running `bind_graph_topology` on the packed image — the same gate the + device runs. +3. **dep_gen differential**: an ST run with `--enable-dep-gen` before/after + must produce byte-identical `deps.json` (reduction does not touch the + recording). This is the guard that the recording/Definition boundary stays + clean. +4. **Scene tests**: `test-all-sim` for both arches (a2a3sim, a5sim) — the + reduced Definitions must replay every existing graph-carrying scene test + identically. + +## Measurement plan + +After landing: `/benchmark -r host_build_graph` (a2a3). HBG's Device column +covers the scheduler dispatch window where `graph_first_unmet_producer` / +`drain_graph_wake_list` run. Expectation on qwen3-class graphs: modest +improvement on the completion-path scans, zero change in host phases +(Definition build is cached). If no example improves beyond noise, record the +null result in `docs/investigations/` per the house rule — the mechanism is +still right (removing work the device does per wake), and the entry documents +by how much. + +## Risks + +- **Reduced image changes Definition hashes**: intentional, benign (see above). +- **Dense fanin rows near `PTO2_MAX_FANIN`**: reduction only shrinks rows; + nothing can overflow a cap by being reduced. +- **Predicate nodes**: dispatch predicates gate execution, not edges; a + predicated-off node's edges still exist in the CSR and still reduce like any + other node's. The predicate evaluation order is untouched. +- **`sync_start` cohorts**: cohort membership is built from slot states at + execution time, not from the CSR; unaffected. + +## Alternatives considered + +- **Reduce at record time (drop in `add_fanin`)**: wrong place — the recording + must stay as-constructed for dep_gen, and the reducibility question is a + whole-graph property, not knowable per-submit (the same reason TMR is 1-hop). +- **Reduce on the device at bind time**: pays the walk on the AICPU at every + first-execution of a Definition and complicates the verifier; the host + already owns the image build. +- **1-hop only (mirror TMR)**: strictly dominated on the host — the exact + algorithm is the same code shape and catches strictly more redundant edges. diff --git a/src/a2a3/runtime/host_build_graph/runtime/orchestrator_core/orchestrator.cpp b/src/a2a3/runtime/host_build_graph/runtime/orchestrator_core/orchestrator.cpp index 7d53656521..267e5ead4a 100644 --- a/src/a2a3/runtime/host_build_graph/runtime/orchestrator_core/orchestrator.cpp +++ b/src/a2a3/runtime/host_build_graph/runtime/orchestrator_core/orchestrator.cpp @@ -805,6 +805,99 @@ T *graph_image_section(std::vector *image, uint32_t offset) { return offset == 0 ? nullptr : reinterpret_cast(image->data() + offset); } +// Exact transitive reduction of the recorded DAG, as a projection for the +// packed image only: the recording itself (and therefore deps.json) keeps the +// as-constructed edge set. An edge p->i is a transitive shortcut when p is +// already an ancestor of i through another producer of i — i.e. p sits in the +// ancestor closure that another kept edge of row i contributes. HBG edges +// carry ordering only (node slots are never reclaimed mid-run), so +// reachability is the whole behavior contract and dropping shortcuts +// preserves it. Nodes arrive in topological order (the CSR fill rejects +// producer >= consumer), so one forward pass finalizes each producer's +// ancestor closure before any later row consumes it. +// +// Returns false only on allocation failure, leaving the recording untouched; +// callers then fall back to packing the unreduced edge set. +struct GraphReducedEdges { + std::vector rows; // (node_count + 1) offsets into producers + std::vector producers; +}; + +bool graph_reduce_transitive_edges(const GraphRecording &recording, GraphReducedEdges *out) { + const size_t node_count = recording.node_count; + if (node_count == 0 || node_count > GRAPH_MAX_NODES) return false; + + const size_t words_per_row = (node_count + 63) / 64; + // Two bitmaps per node: `ancestors` holds the node's ancestor closure over + // the reduced graph (kept edges only), `keep` the direct producers that + // survive. Separate because the closure contains transitive ancestors that + // are not direct edges. 1024-node cap = 2 x 128 KiB scratch, sized to the + // actual row count below that. + std::vector ancestors_storage; + std::vector keep_storage; + try { + ancestors_storage.assign(words_per_row * node_count, 0); + keep_storage.assign(words_per_row * node_count, 0); + out->rows.assign(node_count + 1, 0); + } catch (const std::bad_alloc &) { + return false; + } + auto ancestors = [words_per_row, &ancestors_storage](size_t i) { + return ancestors_storage.data() + i * words_per_row; + }; + auto keep_row = [words_per_row, &keep_storage](size_t i) { + return keep_storage.data() + i * words_per_row; + }; + + // Single forward pass. When row i is processed, every producer p < i has + // its final ancestor closure (all its producers are earlier), so the + // shortcut test reads exact reachability over the reduced graph. Within a + // row, entries are visited in reverse recording order so the deepest + // producer is kept first and its closure (which contains the row's + // transitive shortcuts) marks them for the drop test; an edge whose + // producer is already in the row's accumulated closure is dropped. Kept + // producers are appended after the row's decisions, in recording order, to + // keep each row contiguous. + try { + out->producers.reserve(recording.internal_fanins.size()); + } catch (const std::bad_alloc &) { + return false; + } + for (size_t i = 0; i < node_count; ++i) { + const GraphRecordedNode &node = recording.nodes[i]; + if (node.fanin_offset > recording.internal_fanins.size() || + node.fanin_count > recording.internal_fanins.size() - node.fanin_offset) { + return false; + } + for (uint32_t f = node.fanin_count; f-- > 0;) { + const size_t producer = recording.internal_fanins[node.fanin_offset + f]; + if (producer >= node_count || producer == i) return false; + // record_submit dedups a row's producers, so a repeat cannot occur; + // the keep bit would be idempotent anyway. + + if (ancestors(i)[producer / 64] & (1ULL << (producer % 64))) { + continue; // shortcut: p already reachable via a kept edge of this row + } + keep_row(i)[producer / 64] |= 1ULL << (producer % 64); + const uint64_t *producer_ancestors = ancestors(producer); + for (size_t w = 0; w < words_per_row; ++w) + ancestors(i)[w] |= producer_ancestors[w]; + ancestors(i)[producer / 64] |= 1ULL << (producer % 64); + } + // Emit kept producers in recording order so a row's entries keep their + // relative order from the recording. + for (uint32_t f = 0; f < node.fanin_count; ++f) { + const size_t producer = recording.internal_fanins[node.fanin_offset + f]; + if (keep_row(i)[producer / 64] & (1ULL << (producer % 64))) { + out->producers.push_back(static_cast(producer)); + } + } + out->rows[i + 1] = static_cast(out->producers.size()); + } + out->rows[0] = 0; + return true; +} + bool graph_build_definition(const GraphRecording &recording, std::vector *image) { if (image == nullptr || recording.unsupported || recording.node_count == 0 || recording.node_count > GRAPH_MAX_NODES || recording.boundary_tensors().empty() || @@ -816,7 +909,7 @@ bool graph_build_definition(const GraphRecording &recording, std::vector UINT32_MAX - total_tensors || source.scalar_count > UINT32_MAX - total_scalars || - source.fanin_count > UINT32_MAX - total_fanins || + source.fanin_count > UINT32_MAX - raw_fanins || source.tensor_source_offset > recording.tensor_sources.size() || source.tensors.size() > recording.tensor_sources.size() - source.tensor_source_offset || source.scalar_offset > recording.scalars.size() || @@ -837,12 +930,19 @@ bool graph_build_definition(const GraphRecording &recording, std::vector= 0 ? 1 : 0; } if (predicate_count > UINT16_MAX) return false; + // Reduce before any count or layout consumes the edge set: every array the + // image carries (fanin/fanout CSR, edge_count) must describe the same + // reduced graph or bind_graph_topology rejects the image on the device. + GraphReducedEdges reduced{}; + bool have_reduced = graph_reduce_transitive_edges(recording, &reduced); + const size_t total_fanins = have_reduced ? reduced.producers.size() : raw_fanins; + GraphDefinition definition{}; definition.full_key = recording.full_key; definition.task_count = static_cast(recording.node_count); @@ -921,11 +1021,20 @@ bool graph_build_definition(const GraphRecording &recording, std::vector(i); - for (uint32_t f = 0; f < source.fanin_count; ++f) { - const size_t producer = recording.internal_fanins[source.fanin_offset + f]; - if (producer >= i) return false; - fanin_indices[fanin_cursor++] = static_cast(producer); - fanout_offsets[producer + 1]++; + if (have_reduced) { + for (uint32_t e = reduced.rows[i]; e < reduced.rows[i + 1]; ++e) { + const uint16_t producer = reduced.producers[e]; + if (producer >= i) return false; + fanin_indices[fanin_cursor++] = producer; + fanout_offsets[producer + 1]++; + } + } else { + for (uint32_t f = 0; f < source.fanin_count; ++f) { + const size_t producer = recording.internal_fanins[source.fanin_offset + f]; + if (producer >= i) return false; + fanin_indices[fanin_cursor++] = static_cast(producer); + fanout_offsets[producer + 1]++; + } } fanin_offsets[i + 1] = static_cast(fanin_cursor); diff --git a/src/a5/runtime/host_build_graph/runtime/orchestrator_core/orchestrator.cpp b/src/a5/runtime/host_build_graph/runtime/orchestrator_core/orchestrator.cpp index 7d53656521..267e5ead4a 100644 --- a/src/a5/runtime/host_build_graph/runtime/orchestrator_core/orchestrator.cpp +++ b/src/a5/runtime/host_build_graph/runtime/orchestrator_core/orchestrator.cpp @@ -805,6 +805,99 @@ T *graph_image_section(std::vector *image, uint32_t offset) { return offset == 0 ? nullptr : reinterpret_cast(image->data() + offset); } +// Exact transitive reduction of the recorded DAG, as a projection for the +// packed image only: the recording itself (and therefore deps.json) keeps the +// as-constructed edge set. An edge p->i is a transitive shortcut when p is +// already an ancestor of i through another producer of i — i.e. p sits in the +// ancestor closure that another kept edge of row i contributes. HBG edges +// carry ordering only (node slots are never reclaimed mid-run), so +// reachability is the whole behavior contract and dropping shortcuts +// preserves it. Nodes arrive in topological order (the CSR fill rejects +// producer >= consumer), so one forward pass finalizes each producer's +// ancestor closure before any later row consumes it. +// +// Returns false only on allocation failure, leaving the recording untouched; +// callers then fall back to packing the unreduced edge set. +struct GraphReducedEdges { + std::vector rows; // (node_count + 1) offsets into producers + std::vector producers; +}; + +bool graph_reduce_transitive_edges(const GraphRecording &recording, GraphReducedEdges *out) { + const size_t node_count = recording.node_count; + if (node_count == 0 || node_count > GRAPH_MAX_NODES) return false; + + const size_t words_per_row = (node_count + 63) / 64; + // Two bitmaps per node: `ancestors` holds the node's ancestor closure over + // the reduced graph (kept edges only), `keep` the direct producers that + // survive. Separate because the closure contains transitive ancestors that + // are not direct edges. 1024-node cap = 2 x 128 KiB scratch, sized to the + // actual row count below that. + std::vector ancestors_storage; + std::vector keep_storage; + try { + ancestors_storage.assign(words_per_row * node_count, 0); + keep_storage.assign(words_per_row * node_count, 0); + out->rows.assign(node_count + 1, 0); + } catch (const std::bad_alloc &) { + return false; + } + auto ancestors = [words_per_row, &ancestors_storage](size_t i) { + return ancestors_storage.data() + i * words_per_row; + }; + auto keep_row = [words_per_row, &keep_storage](size_t i) { + return keep_storage.data() + i * words_per_row; + }; + + // Single forward pass. When row i is processed, every producer p < i has + // its final ancestor closure (all its producers are earlier), so the + // shortcut test reads exact reachability over the reduced graph. Within a + // row, entries are visited in reverse recording order so the deepest + // producer is kept first and its closure (which contains the row's + // transitive shortcuts) marks them for the drop test; an edge whose + // producer is already in the row's accumulated closure is dropped. Kept + // producers are appended after the row's decisions, in recording order, to + // keep each row contiguous. + try { + out->producers.reserve(recording.internal_fanins.size()); + } catch (const std::bad_alloc &) { + return false; + } + for (size_t i = 0; i < node_count; ++i) { + const GraphRecordedNode &node = recording.nodes[i]; + if (node.fanin_offset > recording.internal_fanins.size() || + node.fanin_count > recording.internal_fanins.size() - node.fanin_offset) { + return false; + } + for (uint32_t f = node.fanin_count; f-- > 0;) { + const size_t producer = recording.internal_fanins[node.fanin_offset + f]; + if (producer >= node_count || producer == i) return false; + // record_submit dedups a row's producers, so a repeat cannot occur; + // the keep bit would be idempotent anyway. + + if (ancestors(i)[producer / 64] & (1ULL << (producer % 64))) { + continue; // shortcut: p already reachable via a kept edge of this row + } + keep_row(i)[producer / 64] |= 1ULL << (producer % 64); + const uint64_t *producer_ancestors = ancestors(producer); + for (size_t w = 0; w < words_per_row; ++w) + ancestors(i)[w] |= producer_ancestors[w]; + ancestors(i)[producer / 64] |= 1ULL << (producer % 64); + } + // Emit kept producers in recording order so a row's entries keep their + // relative order from the recording. + for (uint32_t f = 0; f < node.fanin_count; ++f) { + const size_t producer = recording.internal_fanins[node.fanin_offset + f]; + if (keep_row(i)[producer / 64] & (1ULL << (producer % 64))) { + out->producers.push_back(static_cast(producer)); + } + } + out->rows[i + 1] = static_cast(out->producers.size()); + } + out->rows[0] = 0; + return true; +} + bool graph_build_definition(const GraphRecording &recording, std::vector *image) { if (image == nullptr || recording.unsupported || recording.node_count == 0 || recording.node_count > GRAPH_MAX_NODES || recording.boundary_tensors().empty() || @@ -816,7 +909,7 @@ bool graph_build_definition(const GraphRecording &recording, std::vector UINT32_MAX - total_tensors || source.scalar_count > UINT32_MAX - total_scalars || - source.fanin_count > UINT32_MAX - total_fanins || + source.fanin_count > UINT32_MAX - raw_fanins || source.tensor_source_offset > recording.tensor_sources.size() || source.tensors.size() > recording.tensor_sources.size() - source.tensor_source_offset || source.scalar_offset > recording.scalars.size() || @@ -837,12 +930,19 @@ bool graph_build_definition(const GraphRecording &recording, std::vector= 0 ? 1 : 0; } if (predicate_count > UINT16_MAX) return false; + // Reduce before any count or layout consumes the edge set: every array the + // image carries (fanin/fanout CSR, edge_count) must describe the same + // reduced graph or bind_graph_topology rejects the image on the device. + GraphReducedEdges reduced{}; + bool have_reduced = graph_reduce_transitive_edges(recording, &reduced); + const size_t total_fanins = have_reduced ? reduced.producers.size() : raw_fanins; + GraphDefinition definition{}; definition.full_key = recording.full_key; definition.task_count = static_cast(recording.node_count); @@ -921,11 +1021,20 @@ bool graph_build_definition(const GraphRecording &recording, std::vector(i); - for (uint32_t f = 0; f < source.fanin_count; ++f) { - const size_t producer = recording.internal_fanins[source.fanin_offset + f]; - if (producer >= i) return false; - fanin_indices[fanin_cursor++] = static_cast(producer); - fanout_offsets[producer + 1]++; + if (have_reduced) { + for (uint32_t e = reduced.rows[i]; e < reduced.rows[i + 1]; ++e) { + const uint16_t producer = reduced.producers[e]; + if (producer >= i) return false; + fanin_indices[fanin_cursor++] = producer; + fanout_offsets[producer + 1]++; + } + } else { + for (uint32_t f = 0; f < source.fanin_count; ++f) { + const size_t producer = recording.internal_fanins[source.fanin_offset + f]; + if (producer >= i) return false; + fanin_indices[fanin_cursor++] = static_cast(producer); + fanout_offsets[producer + 1]++; + } } fanin_offsets[i + 1] = static_cast(fanin_cursor); diff --git a/tests/ut/cpp/CMakeLists.txt b/tests/ut/cpp/CMakeLists.txt index 7d341fb9b1..93984f8cc8 100644 --- a/tests/ut/cpp/CMakeLists.txt +++ b/tests/ut/cpp/CMakeLists.txt @@ -937,6 +937,26 @@ target_sources(test_a5_hbg_graph_submit_failure PRIVATE ${A5_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_reduction common/test_hbg_graph_reduction.cpp) +target_sources(test_hbg_graph_reduction PRIVATE + ${HBG_RUNTIME_DIR}/orchestrator_core/orchestrator.cpp + ${HBG_RUNTIME_DIR}/orchestrator_core/ring_buffer.cpp + ${HBG_RUNTIME_DIR}/shared/shared_memory.cpp + ${HBG_RUNTIME_DIR}/shared/tensormap.cpp + ${HBG_RUNTIME_DIR}/shared/runtime_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_reduction common/test_hbg_graph_reduction.cpp) +target_sources(test_a5_hbg_graph_reduction PRIVATE + ${A5_HBG_RUNTIME_DIR}/orchestrator_core/orchestrator.cpp + ${A5_HBG_RUNTIME_DIR}/orchestrator_core/ring_buffer.cpp + ${A5_HBG_RUNTIME_DIR}/shared/shared_memory.cpp + ${A5_HBG_RUNTIME_DIR}/shared/tensormap.cpp + ${A5_HBG_RUNTIME_DIR}/shared/runtime_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_graph_async_submit common/test_hbg_graph_async_submit.cpp) add_a5_hbg_runtime_test(test_a5_hbg_slot_claim common/test_hbg_slot_claim.cpp) target_sources(test_a5_hbg_slot_claim PRIVATE diff --git a/tests/ut/cpp/common/test_hbg_graph_reduction.cpp b/tests/ut/cpp/common/test_hbg_graph_reduction.cpp new file mode 100644 index 0000000000..6bce5201b6 --- /dev/null +++ b/tests/ut/cpp/common/test_hbg_graph_reduction.cpp @@ -0,0 +1,233 @@ +/* + * 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. + * ----------------------------------------------------------------------------------------------------------- + */ + +// Exact transitive reduction of the packed Definition's fanin CSR. Each test +// records a small DAG through the real orchestrator path (graph_begin / +// graph_prepare / submit / graph_end), takes the packed image out of the host +// Definition list, and asserts the reduced edge set plus the invariants +// bind_graph_topology enforces on the device. + +#include + +#include +#include +#include +#include + +#include "graph_execution.h" +#include "graph_host_state.h" +#include "orchestrator.h" +#include "scheduler/scheduler.h" +#include "shared_memory.h" +#include "utils/device_arena.h" + +namespace { + +class HbgGraphReductionTest : public ::testing::Test { +protected: + static constexpr size_t HEAP_BYTES = 256 * 1024; + + DeviceArena sm_arena; + DeviceArena runtime_arena; + std::vector gm_heap{}; + PTO2SharedMemoryHandle *sm_handle = nullptr; + PTO2SchedulerLayout sched_layout{}; + PTO2SchedulerState sched{}; + PTO2OrchestratorState orch{}; + GraphHostStatePtr graph_state{}; + + void SetUp() override { + sm_handle = PTO2SharedMemoryHandle::create_and_init_default(sm_arena); + ASSERT_NE(sm_handle, nullptr); + gm_heap.resize(HEAP_BYTES); + + sched_layout = PTO2SchedulerState::reserve_layout(runtime_arena); + ASSERT_NE(runtime_arena.commit(), nullptr); + + ASSERT_TRUE(sched.init_data_from_layout(sched_layout, runtime_arena, sm_handle->sm_base)); + sched.wire_arena_pointers(sched_layout, runtime_arena); + ASSERT_TRUE(orch.init(sm_handle->sm_base, gm_heap.data(), HEAP_BYTES, PTO2_TASK_WINDOW_SIZE, &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(); + sched.destroy(); + runtime_arena.release(); + sm_arena.release(); + } + + // Records `producer_count` dummy tasks (node 0 .. producer_count-1), then + // one consumer task consuming tensors produced by `consumer_inputs` of the + // producers. Dependencies come out of creator retention: the consumer + // takes producer outputs as INPUT tensors, which records one fanin edge per + // distinct producer. + // + // The chain edges between the producers themselves are laid in by giving + // each producer p (p > 0) the previous producer's output as an input, so + // producer rows read [p-1]; the consumer's row reads the chosen subset. + // A diamond needs the subset to include both a middle hop and its own + // ancestor. + const GraphDefinition * + record_chain_and_consumer(uint32_t producer_count, const std::vector &consumer_inputs, uint64_t key) { + std::array storage{}; + uint32_t shape[] = {static_cast(storage.size())}; + ChipTensor boundary = make_tensor_external(storage.data(), shape, 1); + GraphTaskArgs boundary_args; + boundary_args.add_input(boundary); + + orch.begin_scope(); + const GraphScopeResult begin = orch.graph_begin(key, boundary_args, key ^ 0x5a5a); + EXPECT_TRUE(begin.recording); + orch.graph_prepare(begin.recording_handle, boundary_args); + + // Each producer reads the previous producer's output: chain 0<-1<-2... + std::vector outputs; + for (uint32_t p = 0; p < producer_count; ++p) { + CoreTaskArgs args; + if (p == 0) { + args.add_input(boundary); + } else { + args.add_input(outputs[p - 1].get_ref(0)); + } + TensorCreateInfo out(shape, 1, DataType::UINT32); + args.add_output(out); + const auto submitted = orch.submit_dummy_task(args); + EXPECT_TRUE(submitted.task_id().is_valid()); + outputs.push_back(submitted); + } + + CoreTaskArgs consumer_args; + for (uint32_t input : consumer_inputs) { + consumer_args.add_input(outputs[input].get_ref(0)); + } + TensorCreateInfo consumer_out(shape, 1, DataType::UINT32); + consumer_args.add_output(consumer_out); + EXPECT_TRUE(orch.submit_dummy_task(consumer_args).task_id().is_valid()); + + EXPECT_TRUE(orch.graph_end()); + orch.end_scope(); + + // Each fixture test records exactly one graph, so the definition list + // holds the one entry just built. full_key is a callable-hash mix the + // caller never sees directly. + const GraphHostDefinitionList definitions = graph_host_definitions(*graph_state); + if (definitions.entries.size() != 1u) return nullptr; + return reinterpret_cast(definitions.entries[0].data); + } + + static std::vector fanin_row(const GraphDefinition &definition, uint32_t consumer) { + const auto *offsets = reinterpret_cast( + reinterpret_cast(&definition) + definition.off_fanin_offsets + ); + const auto *indices = reinterpret_cast( + reinterpret_cast(&definition) + definition.off_fanin_indices + ); + return std::vector(indices + offsets[consumer], indices + offsets[consumer + 1]); + } +}; + +// Diamond: producers 0 -> 1 -> 2 (chain), consumer reads 0 and 2. The direct +// 0 -> consumer edge is implied by 0 -> 1 -> 2 -> consumer, so only 2 must +// remain in the consumer's row. +TEST_F(HbgGraphReductionTest, DiamondShortcutEdgeIsDropped) { + const GraphDefinition *definition = record_chain_and_consumer(3, {0, 2}, 0xd1a); + ASSERT_NE(definition, nullptr); + ASSERT_EQ(definition->task_count, 4u); + + const std::vector row = fanin_row(*definition, 3); + ASSERT_EQ(row.size(), 1u); + EXPECT_EQ(row[0], 2u); +} + +// Same shape but the consumer reads 1 and 2: 1 -> 2 -> consumer implies the +// 1 -> consumer shortcut, so only the chain tail remains. +TEST_F(HbgGraphReductionTest, DeeperDiamondStillReducesToOneHop) { + const GraphDefinition *definition = record_chain_and_consumer(3, {1, 2}, 0xd2); + ASSERT_NE(definition, nullptr); + const std::vector row = fanin_row(*definition, 3); + ASSERT_EQ(row.size(), 1u); + EXPECT_EQ(row[0], 2u); +} + +// Chain 0 -> 1 -> 2 -> 3, consumer reads 0 and 3: the shortcut spans three +// hops. The 1-hop TMR reducer cannot see this; the host reduction can. +TEST_F(HbgGraphReductionTest, ThreeHopShortcutIsDropped) { + const GraphDefinition *definition = record_chain_and_consumer(4, {0, 3}, 0xd3); + ASSERT_NE(definition, nullptr); + ASSERT_EQ(definition->task_count, 5u); + + const std::vector row = fanin_row(*definition, 4); + ASSERT_EQ(row.size(), 1u); + EXPECT_EQ(row[0], 3u); +} + +// A consumer of a single chain tail has nothing redundant: its one edge stays. +TEST_F(HbgGraphReductionTest, IndependentProducersAreKept) { + const GraphDefinition *definition = record_chain_and_consumer(4, {3}, 0xd4a); + ASSERT_NE(definition, nullptr); + EXPECT_EQ(fanin_row(*definition, 4).size(), 1u); +} + +// The reduced image must still satisfy the CSR invariants bind_graph_topology +// enforces on the device: monotone offsets, edge_count consistency on both the +// fanin and fanout sides, indices strictly below their consumer, and roots +// exactly the zero-length rows. +TEST_F(HbgGraphReductionTest, ReducedImageHoldsCsrInvariants) { + const GraphDefinition *definition = record_chain_and_consumer(4, {0, 3}, 0xd5); + ASSERT_NE(definition, nullptr); + + const auto *bytes = reinterpret_cast(definition); + const auto *fanin_offsets = reinterpret_cast(bytes + definition->off_fanin_offsets); + const auto *fanin_indices = reinterpret_cast(bytes + definition->off_fanin_indices); + const auto *fanout_offsets = reinterpret_cast(bytes + definition->off_fanout_offsets); + + ASSERT_EQ(fanin_offsets[0], 0u); + ASSERT_EQ(fanin_offsets[definition->task_count], definition->edge_count); + ASSERT_EQ(fanout_offsets[0], 0u); + ASSERT_EQ(fanout_offsets[definition->task_count], definition->edge_count); + + uint32_t observed_roots = 0; + std::vector fanout_counts(definition->task_count + 1, 0); + for (uint32_t consumer = 0; consumer < definition->task_count; ++consumer) { + const uint32_t begin = fanin_offsets[consumer]; + const uint32_t end = fanin_offsets[consumer + 1]; + ASSERT_LE(begin, end); + ASSERT_LE(end, definition->edge_count); + if (begin == end) observed_roots++; + for (uint32_t edge = begin; edge < end; ++edge) { + ASSERT_LT(fanin_indices[edge], consumer); + fanout_counts[fanin_indices[edge]]++; + } + } + EXPECT_EQ(observed_roots, definition->root_count); + // The fanout side is derived data rebuilt from the reduced edges, so its + // per-producer prefix sums must match the fanin-side counts exactly. + for (uint32_t producer = 0; producer < definition->task_count; ++producer) { + EXPECT_EQ(fanout_offsets[producer + 1] - fanout_offsets[producer], fanout_counts[producer]) + << "producer " << producer; + } +} + +// Roots (zero-fanin nodes) are unaffected by reduction: the chain head keeps +// its empty row and root_count stays 1. +TEST_F(HbgGraphReductionTest, RootCountUnchanged) { + const GraphDefinition *definition = record_chain_and_consumer(4, {0, 3}, 0xd6); + ASSERT_NE(definition, nullptr); + EXPECT_EQ(definition->root_count, 1u); + EXPECT_TRUE(fanin_row(*definition, 0).empty()); +} + +} // namespace