From 35b1aaa599c5de07d6f0b8f0904c12c34b114dca Mon Sep 17 00:00:00 2001 From: Julian P Samaroo Date: Mon, 12 Oct 2026 14:38:51 -0700 Subject: [PATCH 1/4] Document dependency constraints behind Datadeps tree serialization --- AGENTS.md | 42 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/AGENTS.md b/AGENTS.md index 0385d5384..6f38e7e21 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -686,3 +686,45 @@ lesson. running; MPICH had placed each rank in its own process group. After a timeout, inspect rank PIDs/PGIDs and stop those groups too. Killing only the wrapper's group is not sufficient for this launcher. + +67. **A topological planning order can serialize independent execution.** + Same-region `DTask` arguments are fetched by `populate_task_info!`, so the + hierarchical sequential planner flushes submission before visiting a value + consumer. DFS topological order interleaves a reduction tree's leaves and + consumers, making each fetch stall planning before the remaining leaves + are submitted. The scalar tree in issue #759 is safe to execute with all + leaves concurrent: flat submission order submits every leaf before the + first consumer fetch. On one MPI rank with four threads, changing only the + sequential planner's traversal to submission order restored concurrency + from two to four and reduced a warmed run from 0.71 s to 0.32 s, preserving + the checked result. MPI requires rank-identical planning order, not DFS; + retain producer-before-consumer ordering, flushes, and result aliasing + analysis when changing the traversal. This does not justify running MPI + planning itself concurrently or skipping fetches for mutable results. + +68. **Submission order is not necessarily breadth-first tree order.** MPI + `collect(::DArray)` builds a Datadeps concatenation tree through + `treereduce_nd`, whose `treereduce` helper recursively emits one subtree + before the next. With 16 vector tiles, both submission order and DFS visit + concatenation levels `[1, 1, 2, 1, 1, 2, 3, ...]`, although eight bottom + concatenations are independent. Merely preserving submission order fixes + issue #759's explicitly level-ordered tree, but does not expose all those + independent gather tasks before the first consumer fetch. A broader fix + needs to prioritize independent producers while preserving dependencies; + inspect how an algorithm emits its tasks before assuming a traversal + change helps it. Buffer-based factorizations avoid this particular + same-region result-fetch stall. + +69. **A placement plan is not proof that a launch reordering is safe.** + The hierarchical prescan records same-region value edges but skips the + unresolved result's aliasing. For `t = @spawn identity(In(A))`, followed + by `@spawn sum(In(t))` and `@spawn fill!(Out(A), 0)`, the initial DAG has + edges 1→2 and 1→3 but lacks the required 2→3 edge: the result aliases A. + A general ready-frontier traversal must conservatively preserve those + possible conflicts until aliases are resolved. At + `aot-schedulers-rebased` commit 04a8ec2, `dag_add_task!` explicitly stops + at same-region result consumers, AOT schedules store placements only, + and EFT release times use original task indices. An ordering extension + therefore needs symbolic value dependencies, alias-safety rules, and + actual launch positions in the cost model; simply re-enabling the + commented `DTaskDAGID` handling or swapping DFS for BFS is insufficient. From 8767df0797adf08e39e0b846a990b41bb4d8388a Mon Sep 17 00:00:00 2001 From: Julian P Samaroo Date: Mon, 12 Oct 2026 15:24:55 -0700 Subject: [PATCH 2/4] Expose independent producers before hierarchical result fetches DFS visits a reduction's consumers before the remaining independent leaves have been submitted. A deterministic FIFO topological frontier exposes those leaves first, including recursively emitted DArray gather trees. Keep DFS for buffer-only regions and retain runtime result fetching, alias analysis, and batch flushes. Unresolved mutable results can alias earlier buffers, so preserve possible RAW/WAR/WAW conflicts until they are resolved. Only isbits result metadata proves no mutable alias; Any remains conservative. Explicit syncdep IDs also constrain submission and must be in the planning DAG. Reuse typed scratch for indegrees, unknown accesses, and syncdep UID lookup. Memory: after 10 warmups and GC.gc(), minimum of 7 Base.gc_num() deltas on a 127-vertex tree: DFS 13 allocations / 5,352 bytes; FIFO 2 / 1,120. In the quiet 2-rank x 2-thread tree benchmark, baseline minima were 11,271/11,273 allocations and 456,160/456,208 bytes per rank; final FIFO minima were 11,197/11,202 and 457,376/459,024 (less than 1% byte overhead). Allocation-suite bounds remain unchanged. Validated full Distributed Datadeps (2,166 assertions), allocations (24), and MPI CPU (656 per rank), plus 27 focused checks on 1 rank x 1 thread. The AOT-compatible port is checked at aot-schedulers-rebased 04a8ec2, including release-position gaps and Greedy/IG/SA replay consistency. --- AGENTS.md | 17 ++++ src/datadeps/hierarchical.jl | 163 ++++++++++++++++++++++++++++++---- test/datadeps.jl | 2 + test/datadeps/launch_order.jl | 139 +++++++++++++++++++++++++++++ test/mpi/cpu.jl | 1 + 5 files changed, 303 insertions(+), 19 deletions(-) create mode 100644 test/datadeps/launch_order.jl diff --git a/AGENTS.md b/AGENTS.md index 6f38e7e21..4fef660c5 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -728,3 +728,20 @@ lesson. therefore needs symbolic value dependencies, alias-safety rules, and actual launch positions in the cost model; simply re-enabling the commented `DTaskDAGID` handling or swapping DFS for BFS is insufficient. + +70. **Explicit syncdeps constrain submission as well as execution.** The + eager submitter resolves a `ThunkSyncdep` ID through its thunk dictionary + when submitting the consumer. A FIFO frontier that includes value and + alias edges but omits explicit syncdeps can submit that consumer before + its synchronization predecessor exists. Include same-region syncdep IDs + in the planning DAG, even when the consumer never fetches that result. + Reuse the UID-to-vertex scratch dictionary and skip it entirely when no + task has explicit syncdeps. + +71. **A returned array need not alias its argument's original storage.** An + `identity(In(A))` task placed away from A receives a Datadeps replica; + its returned array aliases that replica, not necessarily the original A. + A regression test that writes through the returned result and expects a + later read of A to see the write must keep the producer on A's owner. + Derive that scope from the acceleration-aware wrapped chunk's memory + space so the test exercises actual aliasing under Distributed and MPI. diff --git a/src/datadeps/hierarchical.jl b/src/datadeps/hierarchical.jl index b195a879e..0e4632133 100644 --- a/src/datadeps/hierarchical.jl +++ b/src/datadeps/hierarchical.jl @@ -412,6 +412,10 @@ struct HierarchicalTaskMeta # arguments (e.g. `In(t1)`). These cannot be `fetch`'d during the # pre-scan because they are not launched yet; they become hard DAG edges. value_deps::Vector{Int} + # An unresolved mutable result can overlap any argument in the region. + # These effects supplement the known spans until distribute_task! resolves it. + unresolved_read::Bool + unresolved_write::Bool end ### Cross-partition chunk ownership ### @@ -731,6 +735,8 @@ function _collect_aliased_args_range!(task_metas::Vector{HierarchicalTaskMeta}, all_deps = HierarchicalTaskInfo[] value_deps = Int[] + unresolved_read = false + unresolved_write = false first_chunk = nothing task_may_alias = false task_inplace = false @@ -747,19 +753,22 @@ function _collect_aliased_args_range!(task_metas::Vector{HierarchicalTaskMeta}, # dependency, which `build_dependency_dag` turns into a hard edge; # aliasing of its result is picked up in `distribute_task!`. # - # N.B. An unlaunched producer from *outside* the region (`pred_idx == - # 0`, reachable when the region is nested inside a `spawn_bulk` that - # has not flushed) is skipped entirely: no value dep and no aliasing - # entry. `distribute_task!` still fetches and tracks it, so the task - # itself is correct, but it contributes no DAG edges -- so if another - # in-region task aliases the same underlying data, the dependency - # between them is missed. Handling this needs the pre-scan to be able - # to wait on out-of-region producers. + # Preserve possible conflicts with other arguments without fetching + # the result. Only an isbits return type proves it cannot alias: + # abstract types (including Any) and arrays must stay conservative. + # This also covers an unlaunched out-of-region producer, although + # submitting that producer is still its enclosing queue's job. if arg_pre_unwrap isa DTask && !istaskstarted(arg_pre_unwrap) pred_idx = get(task_to_idx, arg_pre_unwrap, 0) if pred_idx != 0 && pred_idx != task_idx push!(value_deps, pred_idx) end + if !isbitstype(arg_pre_unwrap.metadata.return_type) + for (_, readdep, writedep) in deps + unresolved_read |= readdep + unresolved_write |= writedep + end + end continue end @@ -795,7 +804,8 @@ function _collect_aliased_args_range!(task_metas::Vector{HierarchicalTaskMeta}, end task_metas[task_idx] = HierarchicalTaskMeta( - pair, first_chunk, task_may_alias, task_inplace, all_deps, value_deps + pair, first_chunk, task_may_alias, task_inplace, all_deps, value_deps, + unresolved_read, unresolved_write ) end end @@ -1076,9 +1086,132 @@ function build_dependency_dag(task_metas::Vector{HierarchicalTaskMeta}, write_num += 1 end + add_syncdep_edges!(dag, task_metas) + add_unresolved_alias_edges!(dag, task_metas) + return dag +end + +# Explicit synchronization also constrains submission: the eager scheduler +# resolves its IDs when a task is submitted, so its predecessors must exist. +function add_syncdep_edges!(dag::SimpleDiGraph, task_metas::Vector{HierarchicalTaskMeta}) + has_syncdeps = false + for meta in task_metas + deps = meta.pair.spec.options.syncdeps + has_syncdeps |= deps !== nothing && !isempty(deps) + end + has_syncdeps || return dag + uid_to_idx = @reusable_dict :hierarchical_syncdep_vertices UInt Int UInt(0) 0 32 + for (v, meta) in enumerate(task_metas) + uid_to_idx[meta.pair.task.uid] = v + end + for (v, meta) in enumerate(task_metas) + deps = meta.pair.spec.options.syncdeps + deps === nothing && continue + for dep in deps + dep.id === nothing && continue + pred = get(uid_to_idx, UInt(dep.id.id), 0) + pred != 0 && pred != v && add_edge!(dag, pred, v) + end + end + return dag +end + +# Unknown result spans obey the same RAW/WAR/WAW rules as known spans, but +# potentially overlap every mutable argument. Read-only branches remain +# independent. An unknown writer starts a new epoch: it depends on every +# earlier access, so its successors need only that writer and later accesses. +function add_unresolved_alias_edges!(dag::SimpleDiGraph, + task_metas::Vector{HierarchicalTaskMeta}) + any_unknown = false + for meta in task_metas + any_unknown |= meta.unresolved_read | meta.unresolved_write + end + any_unknown || return dag + + readers = @reusable_vector :unresolved_alias_readers Int 0 32 + writers = @reusable_vector :unresolved_alias_writers Int 0 32 + unknown_readers = @reusable_vector :unresolved_alias_unknown_readers Int 0 32 + unknown_writer = 0 + for (v, meta) in enumerate(task_metas) + reads = false + writes = false + for dep in meta.deps + reads |= dep.readdep + writes |= dep.writedep + end + if unknown_writer != 0 && + (reads || writes || meta.unresolved_read || meta.unresolved_write) + add_edge!(dag, unknown_writer, v) + end + if meta.unresolved_write + for pred in readers + add_edge!(dag, pred, v) + end + for pred in writers + add_edge!(dag, pred, v) + end + for pred in unknown_readers + add_edge!(dag, pred, v) + end + empty!(readers) + empty!(writers) + empty!(unknown_readers) + unknown_writer = v + else + if meta.unresolved_read + for pred in writers + add_edge!(dag, pred, v) + end + end + if writes + for pred in unknown_readers + add_edge!(dag, pred, v) + end + end + reads && push!(readers, v) + writes && push!(writers, v) + meta.unresolved_read && push!(unknown_readers, v) + end + end return dag end +""" + datadeps_launch_order(dag, has_value_deps) -> Vector{Int} + +Keep the existing DFS traversal for buffer-only regions. For value-dependent +regions, use a FIFO topological frontier: independent producers are prepared +before newly unlocked consumers can block planning by fetching their results. +This is a submission order, not an execution barrier between graph levels. +The DAG must include conservative edges for unresolved mutable result aliases. +Vertex and neighbor order are deterministic so MPI ranks replay the same plan. +""" +function datadeps_launch_order(dag::SimpleDiGraph, has_value_deps::Bool) + if !has_value_deps + return topological_sort_by_dfs(dag) + end + n = nv(dag) + remaining = @reusable_vector :datadeps_launch_indegrees Int 0 32 + resize!(remaining, n) + order = Int[] + sizehint!(order, n) + for v in vertices(dag) + remaining[v] = length(inneighbors(dag, v)) + remaining[v] == 0 && push!(order, v) + end + head = 1 + while head <= length(order) + v = order[head] + for succ in outneighbors(dag, v) + remaining[succ] -= 1 + remaining[succ] == 0 && push!(order, succ) + end + head += 1 + end + length(order) == n || throw(ArgumentError("Datadeps dependency graph contains a cycle")) + return order +end + """ Target number of tasks per partition when partitioning across the local processors of a single owner. @@ -1368,11 +1501,7 @@ function schedule_partition_full!(queue::DataDepsTaskQueue, proc_to_scope_lfu = BasicLFUCache{Processor,AbstractScope}(1024) vert_set = Set{Int}(partition_verts) - topo = try - topological_sort_by_dfs(dag) - catch - collect(vertices(dag)) - end + topo = datadeps_launch_order(dag, !isempty(value_dep_verts)) ordered_verts = filter(v -> v in vert_set, topo) # N.B. Each partition gets its own fresh scheduler shard via `similar` @@ -1544,11 +1673,7 @@ function schedule_partitions_sequential!(queue::DataDepsTaskQueue, temp_queues[pid] = DataDepsTaskQueue(submit_queue; scheduler=similar(queue.scheduler)) end - topo = try - topological_sort_by_dfs(dag) - catch - collect(vertices(dag)) - end + topo = datadeps_launch_order(dag, !isempty(value_dep_verts)) # Copy tasks spawned from within `distribute_task!` go through the same queue # as the tasks they serve, so that they are held (or not) alongside them. diff --git a/test/datadeps.jl b/test/datadeps.jl index a7d020350..63deecae7 100644 --- a/test/datadeps.jl +++ b/test/datadeps.jl @@ -2,6 +2,8 @@ import Dagger: ChunkView, Chunk, AbstractAliasing, MemorySpace, ArgumentWrapper import Dagger: aliasing, memory_space using LinearAlgebra, Graphs +include("datadeps/launch_order.jl") + @testset "Hierarchical timing gates" begin # Exercise both the identity slot and the general move_rewrap path. Timing # must remain available when enabled, independently of scheduler logging. diff --git a/test/datadeps/launch_order.jl b/test/datadeps/launch_order.jl new file mode 100644 index 000000000..fc9c3ebb4 --- /dev/null +++ b/test/datadeps/launch_order.jl @@ -0,0 +1,139 @@ +# Shared by the Distributed and MPI suites. Capture tasks without launching +# them so ordering checks do not depend on execution speed or thread count. +function capture_launch_graph(f) + queue = Dagger.DataDepsTaskQueue(Dagger.DefaultTaskQueue(); + scheduler=Dagger.RoundRobinScheduler()) + Dagger.with_options(f; task_queue=queue) + metas, args = Dagger.collect_aliased_args(queue.seen_tasks) + _, overlaps, ainfos = Dagger.build_aliasing_parallel(args) + dag = Dagger.build_dependency_dag(metas, ainfos, overlaps) + return metas, dag +end + +@testset "Datadeps launch ordering" begin + @testset "Independent leaves precede value consumers" begin + metas, dag = capture_launch_graph() do + p = [Dagger.@spawn identity(i) for i in 1:16] + Dagger.treereduce((x,y) -> Dagger.@spawn(+(x,y)), p) + end + order = @inferred Dagger.datadeps_launch_order(dag, true) + @test order[1:16] == collect(1:16) + positions = invperm(order) + @test all(positions[v] < positions[s] for v in Dagger.vertices(dag) + for s in Dagger.outneighbors(dag, v)) + @test Dagger.datadeps_launch_order(dag, false) == + Dagger.topological_sort_by_dfs(dag) + end + + @testset "Recursive gather exposes every independent bottom concatenation" begin + chunks = [Dagger.datadeps_arg_wrap([i]) for i in 1:16] + metas, dag = capture_launch_graph() do + Dagger.treereduce((x,y) -> Dagger.@spawn(vcat(Dagger.In(x), Dagger.In(y))), chunks) + end + roots = [v for v in Dagger.vertices(dag) if isempty(Dagger.inneighbors(dag, v))] + @test length(roots) == 8 + order = Dagger.datadeps_launch_order(dag, true) + @test order[1:8] == roots + end + + @testset "Unresolved aliases preserve RAW, WAR and WAW" begin + for modifier in (Dagger.In, Dagger.Out, Dagger.InOut) + A = ones(4) + metas, dag = capture_launch_graph() do + t = Dagger.@spawn identity(Dagger.In(A)) + Dagger.@spawn sum(modifier(t)) + Dagger.@spawn fill!(Dagger.Out(A), 0) + end + @test 3 in Dagger.outneighbors(dag, 2) + @test Dagger.datadeps_launch_order(dag, true) == [1, 2, 3] + end + A = ones(4) + metas, dag = capture_launch_graph() do + t = Dagger.@spawn identity(Dagger.In(A)) + Dagger.@spawn fill!(Dagger.Out(A), 2) + Dagger.@spawn sum(Dagger.In(t)) + end + @test 3 in Dagger.outneighbors(dag, 2) + + metas, dag = capture_launch_graph() do + t = Dagger.@spawn return_type=Any identity(Dagger.In(A)) + Dagger.@spawn sum(Dagger.In(t)) + Dagger.@spawn fill!(Dagger.Out(A), 0) + end + @test metas[2].unresolved_read + @test 3 in Dagger.outneighbors(dag, 2) + + # Array-typed results are unknown even when their producers look pure. + # An unknown writer must wait for earlier readers, and later readers + # and writers must wait for it. Consecutive unknown writes start new + # epochs without dropping any dependency. + metas, dag = capture_launch_graph() do + t = Dagger.@spawn identity(Dagger.In(A)) + Dagger.@spawn sum(Dagger.In(A)) + Dagger.@spawn fill!(Dagger.Out(t), 2) + Dagger.@spawn sum(Dagger.In(A)) + Dagger.@spawn fill!(Dagger.Out(t), 3) + Dagger.@spawn sum(Dagger.In(A)) + end + @test 3 in Dagger.outneighbors(dag, 2) + @test 4 in Dagger.outneighbors(dag, 3) + @test 5 in Dagger.outneighbors(dag, 4) + @test 6 in Dagger.outneighbors(dag, 5) + end + + @testset "Returned arrays and views retain program order" begin + for hierarchical in (false, true), producer in (identity, x -> view(x, 2:3)) + A = collect(1:4) + chunk = Dagger.datadeps_arg_wrap(A) + scope = Dagger.memory_space_scope(Dagger.memory_space(chunk)) + t = Dagger.with_options(; scope) do + Dagger.spawn_datadeps(; hierarchical) do + a = Dagger.@spawn producer(Dagger.In(A)) + r = Dagger.@spawn sum(Dagger.In(a)) + Dagger.@spawn fill!(Dagger.Out(A), 0) + r + end + end + @test fetch(t) == (producer === identity ? 10 : 5) + end + A = zeros(4) + # Returning a moved array can return a replica. Keep the producer + # on the original array's owner so this checks actual result aliasing. + chunk = Dagger.datadeps_arg_wrap(A) + scope = Dagger.memory_space_scope(Dagger.memory_space(chunk)) + tasks = Dagger.with_options(; scope) do + Dagger.spawn_datadeps() do + a = Dagger.@spawn identity(Dagger.In(A)) + Dagger.@spawn fill!(Dagger.Out(a), 7) + r = Dagger.@spawn sum(Dagger.In(A)) + Dagger.@spawn fill!(Dagger.Out(a), 8) + s = Dagger.@spawn sum(Dagger.In(A)) + (r, s) + end + end + @test fetch.(tasks) == (28, 32) + end + + @testset "Explicit synchronization constrains launch order" begin + metas, dag = capture_launch_graph() do + t = Dagger.@spawn identity(1) + s = Dagger.@spawn identity(t) + Dagger.@spawn syncdeps=Set([Dagger.ThunkSyncdep(s)]) identity(3) + end + @test 3 in Dagger.outneighbors(dag, 2) + @test Dagger.datadeps_launch_order(dag, true) == [1, 2, 3] + t = Dagger.spawn_datadeps() do + p = Dagger.@spawn identity(1) + s = Dagger.@spawn identity(p) + Dagger.@spawn syncdeps=Set([Dagger.ThunkSyncdep(s)]) identity(3) + end + @test fetch(t) == 3 + end + + @testset "Cycles are reported" begin + dag = Dagger.SimpleDiGraph(2) + Dagger.add_edge!(dag, 1, 2) + Dagger.add_edge!(dag, 2, 1) + @test_throws ArgumentError Dagger.datadeps_launch_order(dag, true) + end +end diff --git a/test/mpi/cpu.jl b/test/mpi/cpu.jl index 65c2697ea..b495ff736 100644 --- a/test/mpi/cpu.jl +++ b/test/mpi/cpu.jl @@ -54,6 +54,7 @@ const accel = Dagger.current_acceleration() include("broadcast.jl") include("receive.jl") +include(joinpath(@__DIR__, "..", "datadeps", "launch_order.jl")) mpi_procs() = sort(collect(Dagger.get_processors(MPIExt.MPIClusterProc(comm))); by=p->(p.rank, Dagger.short_name(p))) From c36fa382b5995242f5ea7e88540c475f3a22a77a Mon Sep 17 00:00:00 2001 From: Julian P Samaroo Date: Mon, 12 Oct 2026 15:25:09 -0700 Subject: [PATCH 3/4] Gather the MPI Cholesky fixture through its tiled parent Generic collect of an UpperTriangular DArray scalar-indexes the distributed storage, making the production-mode smoke spend minutes fetching individual elements. Collect the parent through its concatenation tree and re-wrap it, preserving the numerical assertion. The complete MPI CPU suite passes 656 assertions per rank. --- AGENTS.md | 8 ++++++++ test/mpi/cpu.jl | 4 +++- 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/AGENTS.md b/AGENTS.md index 4fef660c5..def4f91ab 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -745,3 +745,11 @@ lesson. later read of A to see the write must keep the producer on A's owner. Derive that scope from the acceleration-aware wrapped chunk's memory space so the test exercises actual aliasing under Distributed and MPI. + +72. **Gather a DArray before collecting a generic triangular wrapper.** + `collect(cholesky(DA).U)` dispatches to LinearAlgebra's generic triangular + copy, which scalar-indexes the DArray and fetches a tile for every element. + The MPI production-mode smoke consequently spent minutes gathering a + 256² factor. `UpperTriangular(collect(parent(cholesky(DA).U)))` gathers + through the DArray concatenation tree once and preserves the same + triangular values for the numerical assertion. diff --git a/test/mpi/cpu.jl b/test/mpi/cpu.jl index b495ff736..f2975fe3c 100644 --- a/test/mpi/cpu.jl +++ b/test/mpi/cpu.jl @@ -800,7 +800,9 @@ end local U for _ in 1:2 DA = DArray(A, Blocks(B, B)) - U = collect(cholesky(DA).U) + # Collect the tiled parent once; generic collect of the triangular + # wrapper fetches every scalar separately under MPI. + U = UpperTriangular(collect(parent(cholesky(DA).U))) end @test U ≈ cholesky(A).U finally From 41dde04751c293f7c682d3479442b7786889bc82 Mon Sep 17 00:00:00 2001 From: Julian P Samaroo Date: Mon, 12 Oct 2026 15:25:45 -0700 Subject: [PATCH 4/4] Measure value-tree launch stalls with rank-uniform MPI samples Provide one shared script for scalar reduction trees, tiny-task trees, tiled DArray gather, and a buffer-only control. Await fixtures, warm ten iterations, and collect seven samples with cooperative barriers and GC outside timing. Report maximum wall time across ranks and minimum GC counts/bytes separately per rank; optional untimed logging counts moves. Reject invalid case selections rather than emitting an empty comparison. Julia 1.12.4, two local MPI ranks with two threads each, uniformity checks off, baseline 02381759 versus the final implementation: scalar_tree: 355.591 -> 163.087 ms median (2.18x) fast_tree: 25.992 -> 14.964 ms median (1.74x) collect: 31.903 -> 24.714 ms median (1.29x) buffer_region: 8.243 -> 8.941 ms median (0.92x) Repeated buffer-control runs show no consistent improvement. Planned moves remain unchanged: zero for both trees and buffer control, 31 for the gather. Scalar-tree minimum allocations/bytes per rank: baseline: 11271 / 456160; 11273 / 456208 final: 11197 / 457376; 11202 / 459024 The traversal improves time without claiming lower whole-region bytes. Reproduce using the same script and Julia environment on both revisions: julia --project=/path/to/env test/mpi/run.jl 2 2 benchmark/launch_order.jl The full script ran on both revisions; final selector guard syntax checked. --- benchmark/launch_order.jl | 115 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 115 insertions(+) create mode 100644 benchmark/launch_order.jl diff --git a/benchmark/launch_order.jl b/benchmark/launch_order.jl new file mode 100644 index 000000000..c45736730 --- /dev/null +++ b/benchmark/launch_order.jl @@ -0,0 +1,115 @@ +# Compare this script across revisions using the same Julia environment. +# Use an environment containing Dagger and MPI. +# Example: julia --project=/path/to/env test/mpi/run.jl 2 2 benchmark/launch_order.jl +# Seven samples follow ten warmups; GC and the cooperative MPI barrier are +# outside the timed region. Wall times are the maximum across ranks, and GC +# counts are reported separately for each rank. Uniformity checks default off. +# DAGGER_LAUNCH_BENCH_CASE selects one case; DAGGER_LAUNCH_BENCH_SAMPLES must be odd. +# DAGGER_LAUNCH_BENCH_COPIES=true counts planned move! tasks in one untimed call. +using Dagger, MPI +Dagger.accelerate!(:mpi) +const uniformity_checks = parse(Bool, get(ENV, "DAGGER_LAUNCH_BENCH_CHECKS", "false")) +Dagger.check_uniformity!(uniformity_checks) +const samples = parse(Int, get(ENV, "DAGGER_LAUNCH_BENCH_SAMPLES", "7")) +samples > 0 && isodd(samples) || error("Sample count must be positive and odd") +const comm = MPI.COMM_WORLD +const rank = MPI.Comm_rank(comm) +function barrier() + req = MPI.Ibarrier(comm) + while !MPI.Test(req) + sleep(0.001) + end +end +function work(i, n) + s = 0.0 + for j in 1:n + x = (j - 0.5) / n + s += 4.0 / (1.0 + x*x) + end + i*s/n +end +function scalar_tree() + t = Dagger.spawn_datadeps() do + p = [Dagger.@spawn work(i, 1<<24) for i in 1:16] + m = length(p) + while m>1 + for k in 1:m÷2 + p[k] = Dagger.@spawn +(p[2k-1], p[2k]) + end + m÷=2 + end + p[1] + end + fetch(t) +end +function fast_tree() + t = Dagger.spawn_datadeps() do + p = [Dagger.@spawn identity(i) for i in 1:64] + Dagger.treereduce((x,y)->Dagger.@spawn(+(x,y)), p) + end + fetch(t) +end +noop!(A) = nothing +const buffers = [zeros(32) for _ in 1:32] +function buffer_region() + Dagger.spawn_datadeps() do + for A in buffers + Dagger.@spawn noop!(Dagger.InOut(A)) + end + end +end +const source = reshape(collect(Float64,1:512*512),512,512) +const fixture = Dagger.DArray(source, Dagger.Blocks(64,64)) +wait(fixture) +const cases = (("scalar_tree",scalar_tree,x->isapprox(x,136pi;rtol=1e-6)), + ("fast_tree",fast_tree,x->x==sum(1:64)), + ("collect",()->collect(fixture),x->x==source), + ("buffer_region",buffer_region,x->true)) +rank == 0 && println("Julia ", VERSION, ", ranks=", MPI.Comm_size(comm), + ", threads/rank=", Threads.nthreads(), + ", uniformity_checks=", uniformity_checks, ", samples=", samples) +selected = get(ENV, "DAGGER_LAUNCH_BENCH_CASE", "") +isempty(selected) || any(case -> first(case) == selected, cases) || + error("Unknown launch-order benchmark case: $selected") +for (name,f,check) in cases + isempty(selected) || selected == name || continue + for _ in 1:10 + @assert check(f()) + end + GC.gc(); barrier() + times = Float64[]; allocs=Int[]; bytes=Int[] + for _ in 1:samples + GC.gc(); barrier() + before = Base.gc_num() + start = time_ns() + result=f() + elapsed=(time_ns()-start)/1e9 + diff=Base.GC_Diff(Base.gc_num(),before) + @assert check(result) + push!(times,MPI.Allreduce(elapsed,max,comm)) + push!(allocs,Base.gc_alloc_count(diff)); push!(bytes,Int(diff.allocd)) + end + move_counts = nothing + if parse(Bool, get(ENV, "DAGGER_LAUNCH_BENCH_COPIES", "false")) + Dagger.enable_logging!(; tasknames=true, all_task_deps=false) + count = 0 + try + @assert check(f()) + logs = Dagger.fetch_logs!() + for worker_logs in values(logs), taskname in worker_logs[:tasknames] + if taskname isa String && occursin("move!", taskname) + count += 1 + end + end + finally + Dagger.disable_logging!() + end + move_counts = MPI.gather(count, comm) + end + all_allocs=MPI.gather(minimum(allocs),comm) + all_bytes=MPI.gather(minimum(bytes),comm) + if rank==0 + ordered=sort(times) + println(name," median_s=",ordered[cld(samples, 2)]," minimum_s=",first(ordered)," allocs_by_rank=",all_allocs," bytes_by_rank=",all_bytes," planned_moves_by_rank=",move_counts," samples=",times) + end +end