From 281d25a1aae7bd7f1a1ec05e82b07aeffcbf9803 Mon Sep 17 00:00:00 2001 From: ChaoZheng109 Date: Fri, 28 Aug 2026 19:28:38 -0700 Subject: [PATCH] Add: transitive reduction of redundant host_build_graph fanin edges A fanin edge P -> C carries readiness only. When another producer Q of C already reaches P, the chain P -> ... -> Q -> C orders C behind P on its own and the direct edge decides nothing: the device still scans it in classify_fanin_state and still moves C between wake lists for it. Both paths that build an edge now drop such edges, at the resolution each can afford. The global submit path publishes one 64-bit ancestor word per task, indexed by task local id. A submit folds its producers' words -- each shifted by that producer's distance, which is distance addition -- into its own, then drops every producer the fold covers. Two host_build_graph properties keep this to a single word of state: a task id is its slot index, handed out by a forward-only bump allocator and never reclaimed, so it doubles as the global submission order and a distance is a subtraction; and a producer's word, published by its own submit, is never rewritten, so reading it needs no proof that the slot still holds the task that wrote it. FANIN_REACH_WINDOW bounds the proof at 64 ids, and a producer further back keeps its edge. A recorded Graph body reduces into the Definition's own fanin CSR, and carries an exact closure rather than a window: a body is capped at MAX_IN_GRAPH_TASKS, so a row is a fixed 128 B and the array 128 KiB of recorder scratch, and a Graph is recorded once against however many replays read the shortened CSR. A producer arbitrarily far back in the body is therefore reduced too. The two compose without either knowing about the other -- a body is ordered against everything before the Graph by the outer Graph task's own fanin, and a producer outside the recording window contributes no in-body edge at all. Because each ancestor row is already a closure, a chain of any length collapses in one pass rather than one hop at a time. Reduction rewrites readiness only. A global producer's buffer lifetime rides last_consumer_local_id, raised when the edge was appended and never lowered here, so a producer whose edge is dropped still waits for that consumer to retire before the host may overwrite it. A body's buffers come out of the Graph's own heap and are released when the Graph completes, never per task, so a dropped in-body edge holds no lifetime either. Every dropped producer is reached from another producer with a strictly larger id, so following the cover relation up terminates at one nothing covers: a task with producers always keeps an edge. Both paths assert that, because emptying a fanin would make the device treat the task as a root and dispatch it against unfinished producers -- a data race rather than a hang. A SIMPLER_DFX build reports the edges built and dropped once per orchestration; each Definition logs its shipped and reduced edge counts at DEBUG as it is laid out. --- .../host_build_graph/docs/RUNTIME_LOGIC.md | 64 ++- .../host_build_graph/docs/RUNTIME_LOGIC.md | 64 ++- src/common/host_build_graph/orchestrator.h | 13 + src/common/host_build_graph/runtime_types.h | 9 + .../host_build_graph/shared/orchestrator.cpp | 203 +++++++++ .../host_build_graph/shared/runtime_init.cpp | 11 + tests/ut/cpp/CMakeLists.txt | 12 + .../cpp/common/test_hbg_fanin_reduction.cpp | 405 ++++++++++++++++++ 8 files changed, 775 insertions(+), 6 deletions(-) create mode 100644 tests/ut/cpp/common/test_hbg_fanin_reduction.cpp 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 a5e7a8cd3c..ced0ba6263 100644 --- a/src/a2a3/runtime/host_build_graph/docs/RUNTIME_LOGIC.md +++ b/src/a2a3/runtime/host_build_graph/docs/RUNTIME_LOGIC.md @@ -113,7 +113,8 @@ a boundary from which region happens to be reserved first — **Why the orchestrator is not in the arena at all.** hbg has no device-side orchestrator, so nothing on the device reads its state: not the `fanin_seen_epoch` -table, not the scope arrays, not the TensorMap (~9.3 MB between them). It is +or `fanin_reach` tables, not the scope arrays, not the TensorMap (~9.4 MB between +them). It is therefore a plain host object that owns those arrays — `OrchestratorState::init` allocates them — and `RuntimeContext` reaches it through a pointer that `bind` drops before the copied zone is uploaded, so no host address crosses the boundary. A @@ -275,12 +276,69 @@ TensorMap maps tensor regions to producer task IDs. For every task: 1. INPUT/INOUT regions look up overlapping producers. 2. Explicit and discovered producers are deduplicated into the payload's fanin region. -3. OUTPUT/INOUT regions register the new task as producer. -4. Each producer tracks its highest consumer local ID for completion metadata. +3. Transitive reduction drops the producers another producer already reaches. +4. OUTPUT/INOUT regions register the new task as producer. +5. Each producer tracks its highest consumer local ID for completion metadata. There is no fanout adjacency or dependency pool. A per-slot completion flag is the readiness truth on device. +#### Bounded transitive reduction + +Step 3 removes edges the rest of the fanin already orders. When a consumer names +both `P` and `Q`, and `Q` is itself reachable from `P`, the chain +`P -> ... -> Q -> consumer` orders the consumer behind `P` on its own, so the +direct `P` edge decides nothing. Dropping it shortens the region the boot scan +and every wake-list reclassification walk, at the cost of two words of work per +submit on the host. + +Each task publishes one 64-bit word of ancestors, indexed by task local id: bit +`i` of task `t` is set when the task `i + 1` ids before `t` reaches it. A submit +folds its producers' words -- shifted by each producer's distance, which is +distance addition -- into its own, then drops every producer whose bit the fold +produced. Because a producer's word is already its own closure, a chain of any +length inside the window collapses in one pass, not one hop at a time. + +Two host_build_graph properties keep this to a single word of state. A task id +is its slot index, handed out by a forward-only bump allocator and never +reclaimed, so it doubles as the global submission order and a distance is a +subtraction; and a producer's word, published by its own submit, is never +rewritten, so reading it needs no proof that the slot still holds the task that +wrote it. `FANIN_REACH_WINDOW` bounds how far back a proof can reach: a producer +further back keeps its edge, since neither it nor its ancestors fit the word. + +Reduction rewrites readiness only. Buffer lifetime rides +`last_consumer_local_id`, raised when the edge was appended and never lowered, so +a producer whose edge is dropped still waits for that consumer to retire before +the host may overwrite it. A `SIMPLER_DFX` build reports the edges built and the +edges dropped once per orchestration. + +#### Inside a recorded Graph body + +A Graph body's edges live in the Definition's own fanin CSR, not in a task table, +and they are reduced where they are recorded -- once, before any replay reads +them. `graph_reduce_recorded_fanin` runs on each recorded task as its producers +are settled, and shortens the CSR the same way. + +The resolution differs, and the difference is the point. The global path runs on +every submit against a table with no fixed bound, so one word of ancestors is +what it can afford. A body is capped at `MAX_IN_GRAPH_TASKS`, so a row is a fixed +128 B and the whole array 128 KiB of recorder scratch; and the fold is paid once +against however many times that Graph is replayed. The recording path therefore +carries the **exact** closure and has no window: a producer arbitrarily far back +in the body is still reduced. + +The two compose without either knowing about the other. A body's tasks are +ordered against everything before the Graph by the outer Graph task's own fanin, +which the global path reduced; a producer outside the recording window contributes +no in-body edge at all. So reducing inside a body cannot change how the body is +ordered against the rest of the run. + +Recorded bodies need no retention argument: a body's buffers come out of the +Graph's own heap and are released when the Graph completes, never per task, so a +dropped edge holds no lifetime. Each Definition logs its shipped and reduced edge +counts at DEBUG as it is laid out. + ## 6. Boot Classification and Wake Lists Submit does not push tasks into ready queues. After the graph arrives on device, diff --git a/src/a5/runtime/host_build_graph/docs/RUNTIME_LOGIC.md b/src/a5/runtime/host_build_graph/docs/RUNTIME_LOGIC.md index a5e7a8cd3c..ced0ba6263 100644 --- a/src/a5/runtime/host_build_graph/docs/RUNTIME_LOGIC.md +++ b/src/a5/runtime/host_build_graph/docs/RUNTIME_LOGIC.md @@ -113,7 +113,8 @@ a boundary from which region happens to be reserved first — **Why the orchestrator is not in the arena at all.** hbg has no device-side orchestrator, so nothing on the device reads its state: not the `fanin_seen_epoch` -table, not the scope arrays, not the TensorMap (~9.3 MB between them). It is +or `fanin_reach` tables, not the scope arrays, not the TensorMap (~9.4 MB between +them). It is therefore a plain host object that owns those arrays — `OrchestratorState::init` allocates them — and `RuntimeContext` reaches it through a pointer that `bind` drops before the copied zone is uploaded, so no host address crosses the boundary. A @@ -275,12 +276,69 @@ TensorMap maps tensor regions to producer task IDs. For every task: 1. INPUT/INOUT regions look up overlapping producers. 2. Explicit and discovered producers are deduplicated into the payload's fanin region. -3. OUTPUT/INOUT regions register the new task as producer. -4. Each producer tracks its highest consumer local ID for completion metadata. +3. Transitive reduction drops the producers another producer already reaches. +4. OUTPUT/INOUT regions register the new task as producer. +5. Each producer tracks its highest consumer local ID for completion metadata. There is no fanout adjacency or dependency pool. A per-slot completion flag is the readiness truth on device. +#### Bounded transitive reduction + +Step 3 removes edges the rest of the fanin already orders. When a consumer names +both `P` and `Q`, and `Q` is itself reachable from `P`, the chain +`P -> ... -> Q -> consumer` orders the consumer behind `P` on its own, so the +direct `P` edge decides nothing. Dropping it shortens the region the boot scan +and every wake-list reclassification walk, at the cost of two words of work per +submit on the host. + +Each task publishes one 64-bit word of ancestors, indexed by task local id: bit +`i` of task `t` is set when the task `i + 1` ids before `t` reaches it. A submit +folds its producers' words -- shifted by each producer's distance, which is +distance addition -- into its own, then drops every producer whose bit the fold +produced. Because a producer's word is already its own closure, a chain of any +length inside the window collapses in one pass, not one hop at a time. + +Two host_build_graph properties keep this to a single word of state. A task id +is its slot index, handed out by a forward-only bump allocator and never +reclaimed, so it doubles as the global submission order and a distance is a +subtraction; and a producer's word, published by its own submit, is never +rewritten, so reading it needs no proof that the slot still holds the task that +wrote it. `FANIN_REACH_WINDOW` bounds how far back a proof can reach: a producer +further back keeps its edge, since neither it nor its ancestors fit the word. + +Reduction rewrites readiness only. Buffer lifetime rides +`last_consumer_local_id`, raised when the edge was appended and never lowered, so +a producer whose edge is dropped still waits for that consumer to retire before +the host may overwrite it. A `SIMPLER_DFX` build reports the edges built and the +edges dropped once per orchestration. + +#### Inside a recorded Graph body + +A Graph body's edges live in the Definition's own fanin CSR, not in a task table, +and they are reduced where they are recorded -- once, before any replay reads +them. `graph_reduce_recorded_fanin` runs on each recorded task as its producers +are settled, and shortens the CSR the same way. + +The resolution differs, and the difference is the point. The global path runs on +every submit against a table with no fixed bound, so one word of ancestors is +what it can afford. A body is capped at `MAX_IN_GRAPH_TASKS`, so a row is a fixed +128 B and the whole array 128 KiB of recorder scratch; and the fold is paid once +against however many times that Graph is replayed. The recording path therefore +carries the **exact** closure and has no window: a producer arbitrarily far back +in the body is still reduced. + +The two compose without either knowing about the other. A body's tasks are +ordered against everything before the Graph by the outer Graph task's own fanin, +which the global path reduced; a producer outside the recording window contributes +no in-body edge at all. So reducing inside a body cannot change how the body is +ordered against the rest of the run. + +Recorded bodies need no retention argument: a body's buffers come out of the +Graph's own heap and are released when the Graph completes, never per task, so a +dropped edge holds no lifetime. Each Definition logs its shipped and reduced edge +counts at DEBUG as it is laid out. + ## 6. Boot Classification and Wake Lists Submit does not push tasks into ready queues. After the graph arrives on device, diff --git a/src/common/host_build_graph/orchestrator.h b/src/common/host_build_graph/orchestrator.h index 65872c8460..02abd5b1f7 100644 --- a/src/common/host_build_graph/orchestrator.h +++ b/src/common/host_build_graph/orchestrator.h @@ -66,6 +66,14 @@ struct OrchestratorState { std::unique_ptr fanin_seen_epoch; uint32_t fanin_seen_current_epoch{1}; + // Frozen dependency-ancestor reachability, one word per task, indexed by local + // id. Bit i of entry `t` is set when the task with local id `t - i - 1` reaches + // `t` through a chain of fanin edges. Published once by t's own submit and never + // rewritten: a task id is also its slot index and hbg reclaims neither, so an + // entry describes the same task for the whole run. Orchestrator scratch like + // fanin_seen_epoch — host-only, and nothing on the device reads it. + std::unique_ptr fanin_reach; + // === TENSOR MAP (Private) === ChipTensorMap tensor_map; // Producer lookup @@ -124,6 +132,11 @@ struct OrchestratorState { int64_t tasks_submitted; int64_t buffers_allocated; int64_t bytes_allocated; + // Fanin edges transitive reduction dropped, and the edges that reached it. Both + // count the whole run and are reported once by mark_done, which is how a change + // to the window or to a workload's shape is read off a run. + int64_t fanin_edges_seen; + int64_t fanin_edges_reduced; #endif bool in_manual_scope() const { return scope_stack_top >= manual_begin_depth; } diff --git a/src/common/host_build_graph/runtime_types.h b/src/common/host_build_graph/runtime_types.h index b01e3e17ba..13b5f9541d 100644 --- a/src/common/host_build_graph/runtime_types.h +++ b/src/common/host_build_graph/runtime_types.h @@ -143,6 +143,15 @@ inline constexpr uint64_t READY_QUEUE_CAPACITY_LIMIT = 32768; // fanin of any workload (paged_attention is the densest). #define CHIP_MAX_FANIN 128 +// How far back transitive reduction can prove one fanin edge redundant, measured +// in task local ids. One native word, so a task's whole ancestor set is a single +// uint64 and the shift-merge in reduce_redundant_fanin is one instruction; the +// pass relies on that width, since a shift by FANIN_REACH_WINDOW would be +// undefined and every ancestor of a producer that far back is out of the window +// anyway. Raising it means widening the entry to several words and shifting +// across them, not editing this number. +inline constexpr int32_t FANIN_REACH_WINDOW = 64; + // Alignment of every per-task region inside an argument pool. Each region starts // and ends on a cache line so TaskPayload::init's round-up scalar memcpy stays // inside the task's own region — see its comment. simpler::hbg::Tensor is already 2 cache diff --git a/src/common/host_build_graph/shared/orchestrator.cpp b/src/common/host_build_graph/shared/orchestrator.cpp index 6b12e117af..f87a6606b5 100644 --- a/src/common/host_build_graph/shared/orchestrator.cpp +++ b/src/common/host_build_graph/shared/orchestrator.cpp @@ -402,6 +402,9 @@ struct GraphBoundary { std::vector types; }; +// Words per row of GraphRecording::task_reach — one bit per task a body may hold. +inline constexpr size_t GRAPH_REACH_WORDS = (MAX_IN_GRAPH_TASKS + 63) / 64; + // Storage for one recorded body, owned by the recorder thread and reset per // recording rather than allocated per recording — see recorder_recording(). struct GraphRecording { @@ -437,6 +440,22 @@ struct GraphRecording { std::vector scalar_sources; std::vector internal_fanins; std::vector output_ranges; + // Exact ancestor closure per recorded task, GRAPH_REACH_WORDS words each, indexed + // by task index within the body. Bit j of task i's row is set when task j reaches + // task i through the body's internal edges. + // + // This is a full bitset rather than the one-word window the global submit path + // carries, and the two limits are what make that affordable here: a body holds at + // most MAX_IN_GRAPH_TASKS tasks, so a row is a fixed 128 B and the whole array + // 128 KiB next to this recording's 4 MB tensor pool; and a Graph is recorded once + // and then replayed, so the fold is paid once for every replay that reads the + // reduced edge set. No window means no edge is kept merely because its producer + // sat too far back. + std::vector task_reach; + // Edges this body's reduction removed, reported against the shipped edge count + // when the Definition is laid out. A Graph is recorded once, so this is where a + // body's redundancy is read off a run. + size_t reduced_edges{0}; // Indexed by RecordedInGraphTask::predicate_index; only predicated tasks // contribute an entry. std::vector predicates; @@ -774,6 +793,9 @@ bool graph_recording_reserve_storage(GraphRecording &recording) { recording.scalar_sources.reserve(kInGraphTaskCap * static_cast(CORE_MAX_SCALAR_ARGS)); recording.output_ranges.reserve(kInGraphTaskCap); recording.predicates.reserve(kInGraphTaskCap); + // Sized rather than reserved: a row is read by index, and every row a body can + // reach must exist before the first task is recorded. + recording.task_reach.assign(kInGraphTaskCap * GRAPH_REACH_WORDS, 0); return true; } @@ -836,6 +858,11 @@ bool graph_recording_reset(GraphRecording &recording, const GraphInflightRecordi recording.internal_fanins.clear(); recording.output_ranges.clear(); recording.predicates.clear(); + // Cleared whole rather than per task, so a row left by the previous body cannot be + // read as this one's: a task that bails out mid-record never writes its own row, + // and a later task of the same body would otherwise fold a stale ancestor set. + std::fill(recording.task_reach.begin(), recording.task_reach.end(), uint64_t{0}); + recording.reduced_edges = 0; return true; } @@ -1028,6 +1055,11 @@ std::optional graph_layout_definition(const GraphRecording &rec return std::nullopt; } definition.total_bytes = static_cast(image_bytes); + LOG_DEBUG( + "[GraphExecution] Definition key=%#llx: %u tasks, %u edges shipped, %zu reduced away", + static_cast(definition.full_key), definition.task_count, definition.edge_count, + recording.reduced_edges + ); return definition; } @@ -1347,6 +1379,86 @@ static bool append_fanin_or_fail( return true; } +// Bounded transitive reduction of one task's fanin, run once per submit on the +// fully-appended edge list and before the count is published. +// +// Pass 1 folds two words over the edges. `direct` carries one bit per producer, at +// its distance d = self - producer; `via` is the union of each producer's own +// ancestor word shifted by that same d, which is distance addition — an ancestor +// a hops behind producer p is a + d hops behind this task. Their union is this +// task's ancestor set, published for its own consumers to walk. Pass 2 then drops +// every producer whose bit appears in `via`: some other producer already reaches +// it, so the direct edge orders nothing the chain does not, and the reduced edge +// list has the same reachability closure as the full one. +// +// Two properties of host_build_graph make the walk trivially sound, and both are +// why this is one word of state rather than the pinning protocol a ring runtime +// needs. A task id is its slot index, handed out by a forward-only bump allocator +// and never reclaimed, so it doubles as the global submission order — the distance +// is a subtraction, with no sequence number to carry. And a producer's entry was +// published by its own submit and is never rewritten, so reading it needs no proof +// that the slot still holds the task that wrote it. +// +// Dropping an edge does not shorten any buffer's lifetime. Retention rides +// last_consumer_local_id, which append_fanin_or_fail already raised to this task +// when the edge was appended and which nothing here lowers: a producer whose edge +// is dropped still waits for this task before the host may overwrite it. +// +// A producer further back than FANIN_REACH_WINDOW keeps its edge and contributes +// nothing — it and all its ancestors are unrepresentable in the window. At exactly +// FANIN_REACH_WINDOW only the direct bit is set: the shift would be undefined, and +// every ancestor of that producer already lies outside the window. +static void reduce_redundant_fanin(OrchestratorState *orch, TaskId self_task_id, int32_t *fanin_slots, int32_t &count) { + const int32_t self = static_cast(simpler::hbg::task_local_id(self_task_id)); + uint64_t *reach = orch->fanin_reach.get(); + + uint64_t direct = 0; + uint64_t via = 0; + for (int32_t i = 0; i < count; i++) { + // A producer is always submitted before its consumer, so the distance is + // positive; a non-positive one would name this task or an unsubmitted slot, + // and is skipped rather than indexed. + const int32_t d = self - fanin_slots[i]; + debug_assert(d > 0 && "a fanin producer is submitted before its consumer"); + if (d <= 0 || d > FANIN_REACH_WINDOW) continue; + direct |= uint64_t{1} << (d - 1); + if (d < FANIN_REACH_WINDOW) via |= reach[fanin_slots[i]] << d; + } + // A via bit lands at index (i + d) for ancestor bit i >= 0 of a producer at + // distance d >= 1, so index 0 is unreachable: the immediately preceding task can + // never be proven redundant. A set bit 0 means the shift-merge has drifted (the + // classic form shifts by d - 1) and the pass is about to drop an edge nothing + // covers. + always_assert((via & uint64_t{1}) == 0 && "via bit 0 set: a distance-1 producer is not reducible"); + reach[self] = direct | via; + +#if SIMPLER_DFX + orch->fanin_edges_seen += count; +#endif + if (via == 0) return; + + // Compact in place, preserving order: classify_fanin_state scans the region from + // the back for the latest-submitted unmet producer, so the surviving edges must + // stay in the order they were appended. + int32_t kept = 0; + for (int32_t i = 0; i < count; i++) { + const int32_t d = self - fanin_slots[i]; + if (d > 0 && d <= FANIN_REACH_WINDOW && (via & (uint64_t{1} << (d - 1))) != 0) continue; + fanin_slots[kept++] = fanin_slots[i]; + } + // Every dropped producer is reachable from some other producer, and that covering + // producer's local id is strictly larger, so following the cover relation up + // terminates at one that nothing covers. A task with producers therefore always + // keeps at least one edge. Emptying the region instead would make the device's + // boot scan classify this task as a root and dispatch it against its unfinished + // producers — a data race, not a hang, so it is worth catching here. + always_assert(kept > 0 && "reduction emptied a non-empty fanin: no producer survived as a maximal element"); +#if SIMPLER_DFX + orch->fanin_edges_reduced += count - kept; +#endif + count = kept; +} + struct PreparedTask { TaskId task_id = TaskId::invalid(); TaskAllocResult alloc_result = {-1, nullptr, nullptr}; @@ -1736,6 +1848,11 @@ static TaskOutputTensors submit_task_common( // The initial scan happens before the scheduler dispatch loop starts. Fanin is // a flat array of position-independent integers, so it crosses to the device // unchanged. + // + // Reduction runs first, on the complete edge list and before anything reads the + // count: it both publishes this task's ancestor word for later submits and + // settles which edges the region actually holds. + reduce_redundant_fanin(orch, task_id, fanin_slots, payload.fanin_count); // The region's length is settled, so the cursor closes it at the real count. The // equality holds only while nothing between the bind and here bound another fanin // region, which is what makes the deferred advance safe. @@ -2008,6 +2125,10 @@ bool graph_submit_outer( return false; } register_task_outputs(boundary_inputs, task_id, orch->tensor_map, orch->in_manual_scope()); + // An outer Graph task takes reduction on the same terms as an ordinary one: it + // completes only once its whole replayed body has, so it composes with the + // ancestor walk exactly as a single task does. + reduce_redundant_fanin(orch, task_id, fanin_slots, payload.fanin_count); // The region's length is settled, so the cursor closes it at the real count. The // equality holds only while nothing between the bind and here bound another fanin // region, which is what makes the deferred advance safe. @@ -2077,6 +2198,74 @@ bool graph_finalize_pending_submissions(OrchestratorState *orch, GraphHostState return true; } +// Exact transitive reduction of one recorded task's internal fanin, run once the +// task's producers are all appended and before its fanin_count is taken. +// +// Same idea as the global submit path's reduction, at full resolution. `via` is the +// union of the producers' own ancestor rows: every task reachable through one of +// them, and therefore already ordered before this task by the chain. A producer +// whose bit `via` carries adds no ordering of its own and is dropped. The published +// row is `via` plus a bit for every producer — kept or dropped, since dropping an +// edge does not stop the producer being an ancestor. +// +// The full bitset is what a recorded body buys over the global path's one-word +// window: a producer arbitrarily far back in the body is still covered, and because +// each row is already a closure, a chain of any length collapses in this one pass. +// A Graph is recorded once and replayed thereafter, so the fold is amortized over +// every replay that reads the shortened CSR. +// +// This rewrites readiness only. A body's buffers come out of the Graph's own heap +// and are released when the Graph completes, not per task, so there is no lifetime +// an edge could have been holding. +// +// An over-cap task index, or a recording whose storage never stood up, has already +// been marked unsupported and its Definition will be refused; either is left alone +// rather than indexed past the row array. +void graph_reduce_recorded_fanin(GraphRecording &recording, size_t task_index, size_t fanin_offset) { + if (task_index >= MAX_IN_GRAPH_TASKS || + recording.task_reach.size() < static_cast(MAX_IN_GRAPH_TASKS) * GRAPH_REACH_WORDS) { + return; + } + const size_t end = recording.internal_fanins.size(); + if (end == fanin_offset) { + return; // a body root: nothing to fold, and its row is already clear + } + + uint64_t via[GRAPH_REACH_WORDS] = {}; + uint64_t direct[GRAPH_REACH_WORDS] = {}; + for (size_t i = fanin_offset; i < end; ++i) { + const uint64_t *producer_row = &recording.task_reach[recording.internal_fanins[i] * GRAPH_REACH_WORDS]; + for (size_t w = 0; w < GRAPH_REACH_WORDS; ++w) { + via[w] |= producer_row[w]; + } + } + + // Compact in place, preserving order: the surviving producers keep the order the + // recording appended them in, which is the order materialize writes to the CSR. + size_t kept = fanin_offset; + for (size_t i = fanin_offset; i < end; ++i) { + const size_t producer = recording.internal_fanins[i]; + direct[producer / 64] |= uint64_t{1} << (producer % 64); + if ((via[producer / 64] >> (producer % 64) & uint64_t{1}) != 0) { + continue; + } + recording.internal_fanins[kept++] = producer; + } + // Each dropped producer is reached from another producer whose index is strictly + // larger, so following the cover relation up terminates at one nothing covers: a + // task with producers always keeps an edge. Emptying the list instead would make + // materialize record this task as a body root and replay it against unfinished + // producers. + always_assert(kept > fanin_offset && "reduction emptied a recorded task's fanin"); + recording.reduced_edges += end - kept; + recording.internal_fanins.resize(kept); + + uint64_t *self = &recording.task_reach[task_index * GRAPH_REACH_WORDS]; + for (size_t w = 0; w < GRAPH_REACH_WORDS; ++w) { + self[w] = via[w] | direct[w]; + } +} + // Record one in-graph task while recording, without consuming a task-table // slot. Builds the task's metadata and materialized outputs exactly as // submit_task_common would, but assigns output buffers from the bit-63 virtual @@ -2352,6 +2541,9 @@ TaskOutputTensors graph_record_submit_in_graph_task( add_fanin(static_cast(dep_index)); } + // Runs on the complete producer list and before the count is taken, so the count + // and the CSR materialize writes both describe the reduced edge set. + graph_reduce_recorded_fanin(recording, task_index, task.fanin_offset); task.fanin_count = static_cast(recording.internal_fanins.size() - task.fanin_offset); if (task.record_packed_base != 0 && task.total_output_size != 0 && task.total_output_size <= UINTPTR_MAX - task.record_packed_base) { @@ -2872,6 +3064,10 @@ TaskOutputTensors OrchestratorState::alloc_tensors(const CoreTaskArgs &args) { outputs.set_task_id(prepared.task_id); payload.init(args, outputs, prepared.alloc_result, layout); payload.fanin_count = 0; // hidden-alloc tasks have no producer dependencies + // With no producers there is nothing to reduce, but the slot is still a producer + // for later submits, so its ancestor word has to say so: empty, not whatever the + // allocation left there. + orch->fanin_reach[simpler::hbg::task_local_id(prepared.task_id)] = 0; CYCLE_COUNT_LAP(g_orch_args_cycle); if (prepared.slot_state != nullptr) { @@ -2928,6 +3124,13 @@ void OrchestratorState::mark_done() { int32_t total_tasks = orch->task_allocator.active_count(); if (total_tasks > 0) { LOG_DEBUG("=== [Orchestrator] total_tasks=%d ===", total_tasks); +#if SIMPLER_DFX + LOG_DEBUG( + "=== [Orchestrator] fanin edges: %lld built, %lld reduced (window=%d) ===", + static_cast(orch->fanin_edges_seen), static_cast(orch->fanin_edges_reduced), + FANIN_REACH_WINDOW + ); +#endif } orch->sm_header->orchestrator_done.store(1, std::memory_order_release); orch->scope_stack_top = -1; diff --git a/src/common/host_build_graph/shared/runtime_init.cpp b/src/common/host_build_graph/shared/runtime_init.cpp index 6374d71847..91024cfb33 100644 --- a/src/common/host_build_graph/shared/runtime_init.cpp +++ b/src/common/host_build_graph/shared/runtime_init.cpp @@ -236,6 +236,17 @@ bool OrchestratorState::init( } memset(orch->fanin_seen_epoch.get(), 0, slots * sizeof(uint32_t)); + // One ancestor word per task slot: 8 B each, 128 KiB at the default 16,384-task + // table, linear in runtime_env.ring_task_window. Zeroed here so a slot claimed + // by a submit that fails before publishing its own entry reads as "no known + // ancestors", which keeps reduction conservative rather than wrong. + orch->fanin_reach.reset(new (std::nothrow) uint64_t[slots]); + if (orch->fanin_reach == nullptr) { + LOG_ERROR("Orchestrator scratch allocation failed (max_tasks=%" PRIu64 ")", max_tasks); + return false; + } + memset(orch->fanin_reach.get(), 0, slots * sizeof(uint64_t)); + if (!orch->tensor_map.init_default(static_cast(max_tasks))) { return false; } diff --git a/tests/ut/cpp/CMakeLists.txt b/tests/ut/cpp/CMakeLists.txt index f1c9bf3419..28e623da69 100644 --- a/tests/ut/cpp/CMakeLists.txt +++ b/tests/ut/cpp/CMakeLists.txt @@ -960,6 +960,13 @@ target_sources(test_hbg_graph_recording_bounds PRIVATE ${HBG_ORCH_SHARED_SOURCES} ${CMAKE_SOURCE_DIR}/../../../src/common/platform/shared/aicpu/args_dump_aicpu.cpp ) +# Reads the published fanin region back after real submits, so it links the same +# out-of-line orchestrator members. +add_a2a3_hbg_runtime_test(test_hbg_fanin_reduction common/test_hbg_fanin_reduction.cpp) +target_sources(test_hbg_fanin_reduction PRIVATE + ${HBG_ORCH_SHARED_SOURCES} + ${CMAKE_SOURCE_DIR}/../../../src/common/platform/shared/aicpu/args_dump_aicpu.cpp +) add_a5_hbg_runtime_test(test_a5_hbg_graph_recording_bounds common/test_hbg_graph_recording_bounds.cpp) target_sources(test_a5_hbg_graph_recording_bounds PRIVATE ${HBG_ORCH_SHARED_SOURCES} @@ -981,6 +988,11 @@ target_sources(test_a5_hbg_slot_claim PRIVATE ${HBG_ORCH_SHARED_SOURCES} ${CMAKE_SOURCE_DIR}/../../../src/common/platform/shared/aicpu/args_dump_aicpu.cpp ) +add_a5_hbg_runtime_test(test_a5_hbg_fanin_reduction common/test_hbg_fanin_reduction.cpp) +target_sources(test_a5_hbg_fanin_reduction PRIVATE + ${HBG_ORCH_SHARED_SOURCES} + ${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_ready_queue_seed common/test_hbg_ready_queue_seed.cpp) add_a5_hbg_runtime_test(test_a5_hbg_mailbox_init common/test_hbg_mailbox_init.cpp) diff --git a/tests/ut/cpp/common/test_hbg_fanin_reduction.cpp b/tests/ut/cpp/common/test_hbg_fanin_reduction.cpp new file mode 100644 index 0000000000..fcb687bbcf --- /dev/null +++ b/tests/ut/cpp/common/test_hbg_fanin_reduction.cpp @@ -0,0 +1,405 @@ +/* + * 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. + * ----------------------------------------------------------------------------------------------------------- + */ +/** + * Acceptance for the orchestrator's transitive reduction of fanin, on both paths + * that build a dependency edge. + * + * An edge P -> C carries readiness only. When another producer Q of C already + * reaches P, the chain P -> ... -> Q -> C orders C behind P by itself and the + * direct edge is redundant, so it is dropped: the device scans a shorter fanin + * region and moves the consumer between fewer wake lists. + * + * The two paths reduce at different resolutions, on purpose. The global submit + * path runs per task on an unbounded table, so it carries one word of ancestors + * and proves coverage within FANIN_REACH_WINDOW ids. A recorded Graph body is + * capped at MAX_IN_GRAPH_TASKS and recorded once before any number of replays, so + * it carries an exact closure and has no window at all. + * + * Every case drives the real path and reads the published edges back — out of + * shared memory for a global task, out of the committed Definition's CSR for a + * recorded one — so what is asserted is what the device would see. The properties + * that must hold across all of them: + * + * - reachability is preserved (a dropped producer is still ordered before the + * consumer through a surviving edge), so no task can start early; + * - retention is untouched — last_consumer_local_id still names the consumer of + * a dropped edge, which is what gates the host's overwrite of that buffer; + * - a global producer beyond the window keeps its edge, because neither it nor + * its ancestors are representable in one word. + */ + +#include + +#include +#include +#include +#include + +#include "graph_execution.h" +#include "graph_host_state.h" +#include "host_build_graph/orchestrator.h" +#include "host_build_graph/shared_memory.h" +#include "host_build_graph/task_id_encoding.h" +#include "utils/device_arena.h" + +class HbgFaninReductionTest : public ::testing::Test { +protected: + DeviceArena sm_arena; + DeviceArena runtime_arena; + SharedMemoryHandle *sm_handle = nullptr; + OrchestratorState orch{}; + SchedulerState sched{}; + SchedulerLayout sched_layout{}; + std::vector gm_heap; + + static constexpr size_t HEAP_BYTES = 64 * 1024; + + void SetUp() override { + sm_handle = SharedMemoryHandle::create_and_init_default(sm_arena); + ASSERT_NE(sm_handle, nullptr); + gm_heap.resize(HEAP_BYTES); + + sched_layout = SchedulerState::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); + sched.seed_queue_slots(); + ASSERT_TRUE(orch.init(sm_handle->sm_base, gm_heap.data(), HEAP_BYTES, CHIP_DEFAULT_GRAPH_TASKS, &sched)); + orch.begin_scope(); + } + + void TearDown() override { + orch.end_scope(); + sched.destroy(); + runtime_arena.release(); + sm_arena.release(); + } + + // A kernel-less task with exactly the given producers. Explicit dependencies + // are the only fanin source here, so each case states its DAG outright rather + // than through tensor aliasing. + TaskId submit(const std::vector &deps) { + CoreTaskArgs args; + if (!deps.empty()) { + args.set_dependencies(deps.data(), static_cast(deps.size())); + } + const TaskOutputTensors result = orch.submit_dummy_task(args); + EXPECT_TRUE(result.task_id().is_valid()); + return result.task_id(); + } + + static int32_t local(TaskId id) { return static_cast(simpler::hbg::task_local_id(id)); } + + // The fanin region as the device boot scan would read it. + std::vector fanin_of(TaskId id) const { + const TaskPayload &payload = sm_handle->header->tasks.task_payloads[local(id)]; + const int32_t *slots = payload.fanin_data(); + return std::vector(slots, slots + payload.fanin_count); + } + + int32_t last_consumer_of(TaskId id) const { + return sm_handle->header->tasks.get_slot_state_by_task_id(local(id)).last_consumer_local_id; + } +}; + +// The base case: C depends on A and on B, and B already depends on A. B alone +// orders C behind A, so the direct A -> C edge goes. +TEST_F(HbgFaninReductionTest, DiamondDropsTheCoveredEdge) { + const TaskId a = submit({}); + const TaskId b = submit({a}); + const TaskId c = submit({a, b}); + + EXPECT_EQ(fanin_of(c), std::vector{local(b)}); +} + +// Coverage is transitive, not one hop: the ancestor word a producer publishes is +// already its own closure, so a chain of any length inside the window collapses. +TEST_F(HbgFaninReductionTest, MultiHopChainCoversTheDirectEdge) { + const TaskId a = submit({}); + const TaskId b = submit({a}); + const TaskId c = submit({b}); + const TaskId d = submit({a, b, c}); + + EXPECT_EQ(fanin_of(d), std::vector{local(c)}); +} + +// Nothing is dropped without a proof: two producers with no path between them +// both gate the consumer and both survive, in the order they were appended. +TEST_F(HbgFaninReductionTest, IndependentProducersBothSurvive) { + const TaskId a = submit({}); + const TaskId b = submit({}); + const TaskId c = submit({a, b}); + + const std::vector expected{local(a), local(b)}; + EXPECT_EQ(fanin_of(c), expected); +} + +// A dropped edge is a readiness edge only. The producer's reclaim gate still +// names the consumer, so the host cannot overwrite that buffer until the consumer +// that reads it has retired. +TEST_F(HbgFaninReductionTest, DroppedEdgeKeepsTheProducerReclaimGate) { + const TaskId a = submit({}); + const TaskId b = submit({a}); + const TaskId c = submit({a, b}); + + ASSERT_EQ(fanin_of(c), std::vector{local(b)}); + EXPECT_EQ(last_consumer_of(a), local(c)); +} + +// Every producer covered by another still leaves one maximal element, so a task +// that had producers never becomes a root. C's three producers form a chain, and +// only its head survives. +TEST_F(HbgFaninReductionTest, ChainedCoverageStillLeavesOneEdge) { + const TaskId a = submit({}); + const TaskId b = submit({a}); + const TaskId c = submit({b}); + const TaskId d = submit({c}); + const TaskId e = submit({a, b, c, d}); + + EXPECT_EQ(fanin_of(e), std::vector{local(d)}); +} + +// Reduction is bounded by one word of ancestors. A producer further back than +// FANIN_REACH_WINDOW is unrepresentable, so its edge is kept even though a chain +// covers it — conservative, never wrong. +TEST_F(HbgFaninReductionTest, ProducerBeyondTheWindowKeepsItsEdge) { + const TaskId root = submit({}); + TaskId prev = root; + // Push root exactly FANIN_REACH_WINDOW + 1 ids back from the consumer, while + // keeping an unbroken chain from it. + for (int32_t i = 0; i < FANIN_REACH_WINDOW; ++i) { + prev = submit({prev}); + } + const TaskId consumer = submit({root, prev}); + + ASSERT_EQ(local(consumer) - local(root), FANIN_REACH_WINDOW + 1); + const std::vector expected{local(root), local(prev)}; + EXPECT_EQ(fanin_of(consumer), expected); +} + +// The same chain one task shorter puts the root at exactly the window edge, where +// it is representable and the chain does cover it. +TEST_F(HbgFaninReductionTest, ProducerAtTheWindowEdgeIsStillReduced) { + const TaskId root = submit({}); + TaskId prev = root; + for (int32_t i = 0; i < FANIN_REACH_WINDOW - 1; ++i) { + prev = submit({prev}); + } + const TaskId consumer = submit({root, prev}); + + ASSERT_EQ(local(consumer) - local(root), FANIN_REACH_WINDOW); + EXPECT_EQ(fanin_of(consumer), std::vector{local(prev)}); +} + +// Surviving edges keep their append order: classify_fanin_state scans the region +// back-to-front for the latest-submitted unmet producer, so compaction must not +// reorder what it leaves behind. +TEST_F(HbgFaninReductionTest, CompactionPreservesAppendOrder) { + const TaskId a = submit({}); + const TaskId b = submit({}); + const TaskId covered = submit({a}); + const TaskId c = submit({b, covered, a}); + + // `a` is covered by `covered`; `b` is independent. Both survivors stay in the + // order set_dependencies listed them. + const std::vector expected{local(b), local(covered)}; + EXPECT_EQ(fanin_of(c), expected); +} + +// --------------------------------------------------------------------------- +// Recorded Graph bodies +// --------------------------------------------------------------------------- + +// A recorded body reduces into the Definition's own fanin CSR, which is what every +// replay of that Graph reads. The recording is driven end to end and the committed +// Definition is read back, so these assert the shipped edge set rather than the +// recorder's scratch. +class HbgRecordedFaninReductionTest : public ::testing::Test { +protected: + DeviceArena sm_arena; + DeviceArena runtime_arena; + SharedMemoryHandle *sm_handle = nullptr; + OrchestratorState orch{}; + SchedulerState sched{}; + SchedulerLayout sched_layout{}; + GraphHostStatePtr graph_state; + GraphDefinitionArena arena{}; + std::vector gm_heap; + std::vector staging; + + static constexpr size_t HEAP_BYTES = 256 * 1024; + static constexpr size_t STAGING_BYTES = 512 * 1024; + + void SetUp() override { + sm_handle = SharedMemoryHandle::create_and_init_default(sm_arena); + ASSERT_NE(sm_handle, nullptr); + gm_heap.resize(HEAP_BYTES); + + sched_layout = SchedulerState::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); + sched.seed_queue_slots(); + ASSERT_TRUE(orch.init(sm_handle->sm_base, gm_heap.data(), HEAP_BYTES, CHIP_DEFAULT_GRAPH_TASKS, &sched)); + + staging.assign(STAGING_BYTES, std::byte{0}); + arena = GraphDefinitionArena{}; + arena.base = staging.data(); + arena.capacity = staging.size(); + arena.object_prefix_bytes = sizeof(GraphDefinitionHeader); + arena.object_align = GRAPH_DEFINITION_OBJECT_ALIGN; + graph_state = make_graph_host_state(arena); + ASSERT_NE(graph_state, nullptr); + orch.graph_host_state = graph_state.get(); + orch.begin_scope(); + } + + void TearDown() override { + orch.end_scope(); + orch.graph_host_state = nullptr; + graph_state.reset(); + sched.destroy(); + runtime_arena.release(); + sm_arena.release(); + } + + // Record one body under `graph_key`, driven by `body`, then commit so the + // Definition is built. The recorder is normally a worker thread; this plays + // both roles on one thread, as the other Graph tests do. + template + void record(uint64_t graph_key, const simpler::hbg::Tensor &boundary, Fn &&body) { + GraphTaskArgs boundary_args; + boundary_args.add_input(boundary); + const GraphScopeResult scope = orch.graph_begin(graph_key, boundary_args, 0x1736); + ASSERT_TRUE(scope.recording); + ASSERT_TRUE(orch.graph_prepare(scope.recording_handle, boundary_args)); + body(); + ASSERT_TRUE(orch.graph_end()); + orch.graph_commit(); + ASSERT_FALSE(orch.fatal); + } + + const GraphDefinition *only_definition() const { + const GraphHostDefinitionList definitions = graph_host_definitions(*graph_state); + if (definitions.entries.size() != 1) return nullptr; + const GraphHostDefinition &entry = definitions.entries[0]; + const std::byte *image = + entry.spill != nullptr ? entry.spill : arena.base + entry.object_offset + arena.object_prefix_bytes; + return reinterpret_cast(image); + } + + // One in-graph task's producers, straight out of the CSR a replay walks. + static std::vector csr_fanin_of(const GraphDefinition &definition, uint32_t task) { + const auto *offsets = + graph_definition_array(definition, definition.off_fanin_offsets, definition.task_count + 1); + const auto *indices = + graph_definition_array(definition, definition.off_fanin_indices, definition.edge_count); + if (offsets == nullptr || (definition.edge_count != 0 && indices == nullptr)) return {}; + return std::vector(indices + offsets[task], indices + offsets[task + 1]); + } + + // A kernel-less in-graph task depending on the given in-graph tasks. + static TaskId body_task(OrchestratorState &orch, const std::vector &deps) { + CoreTaskArgs args; + if (!deps.empty()) { + args.set_dependencies(deps.data(), static_cast(deps.size())); + } + const TaskOutputTensors result = orch.submit_dummy_task(args); + EXPECT_TRUE(result.task_id().is_valid()); + return result.task_id(); + } +}; + +// The diamond again, this time inside a body: the Definition ships one edge into +// the consumer, so every replay of this Graph walks the shorter CSR. +TEST_F(HbgRecordedFaninReductionTest, DiamondInsideABodyShipsTheReducedCsr) { + std::array storage{}; + uint32_t shape[] = {static_cast(storage.size())}; + const simpler::hbg::Tensor boundary = simpler::hbg::make_tensor_external(storage.data(), shape, 1); + + record(0xFA1EDA01, boundary, [&] { + const TaskId a = body_task(orch, {}); + const TaskId b = body_task(orch, {a}); + body_task(orch, {a, b}); + }); + + const GraphDefinition *definition = only_definition(); + ASSERT_NE(definition, nullptr); + ASSERT_EQ(definition->task_count, 3u); + EXPECT_EQ(csr_fanin_of(*definition, 2), std::vector{1}); + // Two edges recorded (a->b, b->c); the direct a->c is gone, so the CSR is not + // merely reordered — it is smaller. + EXPECT_EQ(definition->edge_count, 2u); +} + +// The recording path carries an exact closure, not a window, so a producer at any +// distance inside the body is reduced. A chain longer than FANIN_REACH_WINDOW would +// defeat the global path's one word and is covered here. +TEST_F(HbgRecordedFaninReductionTest, AChainLongerThanTheGlobalWindowIsStillReduced) { + std::array storage{}; + uint32_t shape[] = {static_cast(storage.size())}; + const simpler::hbg::Tensor boundary = simpler::hbg::make_tensor_external(storage.data(), shape, 1); + + constexpr int32_t CHAIN = FANIN_REACH_WINDOW + 8; + record(0xFA1EDA02, boundary, [&] { + const TaskId root = body_task(orch, {}); + TaskId prev = root; + for (int32_t i = 0; i < CHAIN; ++i) { + prev = body_task(orch, {prev}); + } + body_task(orch, {root, prev}); + }); + + const GraphDefinition *definition = only_definition(); + ASSERT_NE(definition, nullptr); + const uint32_t consumer = definition->task_count - 1; + ASSERT_EQ(definition->task_count, static_cast(CHAIN) + 2u); + // The root sits CHAIN + 1 tasks back — past what one word could represent — and + // the chain still covers it. + EXPECT_EQ(csr_fanin_of(*definition, consumer), std::vector{static_cast(consumer - 1)}); +} + +// Independent producers inside a body both survive, and in recording order. +TEST_F(HbgRecordedFaninReductionTest, IndependentBodyProducersBothSurvive) { + std::array storage{}; + uint32_t shape[] = {static_cast(storage.size())}; + const simpler::hbg::Tensor boundary = simpler::hbg::make_tensor_external(storage.data(), shape, 1); + + record(0xFA1EDA03, boundary, [&] { + const TaskId a = body_task(orch, {}); + const TaskId b = body_task(orch, {}); + body_task(orch, {a, b}); + }); + + const GraphDefinition *definition = only_definition(); + ASSERT_NE(definition, nullptr); + const std::vector expected{0, 1}; + EXPECT_EQ(csr_fanin_of(*definition, 2), expected); +} + +// A body root has no internal producers — its ordering comes from the outer shell's +// own fanin — and reduction leaves that untouched, so the Definition still names it +// a root. +TEST_F(HbgRecordedFaninReductionTest, BodyRootsAreLeftAlone) { + std::array storage{}; + uint32_t shape[] = {static_cast(storage.size())}; + const simpler::hbg::Tensor boundary = simpler::hbg::make_tensor_external(storage.data(), shape, 1); + + record(0xFA1EDA04, boundary, [&] { + const TaskId a = body_task(orch, {}); + body_task(orch, {a}); + }); + + const GraphDefinition *definition = only_definition(); + ASSERT_NE(definition, nullptr); + EXPECT_TRUE(csr_fanin_of(*definition, 0).empty()); + EXPECT_EQ(definition->root_count, 1u); +}