diff --git a/AGENTS.md b/AGENTS.md index 0385d5384..def4f91ab 100644 --- a/AGENTS.md +++ b/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. 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 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..f2975fe3c 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))) @@ -799,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