Expose independent value producers in hierarchical Datadeps planning

AGENTS.md

@@ -686,3 +686,70 @@ 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.
+
+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.
+
+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.

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