diff --git a/docs/make.jl b/docs/make.jl index 8f1f97f5..d99f0162 100644 --- a/docs/make.jl +++ b/docs/make.jl @@ -20,6 +20,7 @@ makedocs(; "Parallel Nested Loops" => "use-cases/parallel-nested-loops.md", ], "Task Spawning" => "task-spawning.md", + "Task Affinity" => "task-affinity.md", "Data Management" => "data-management.md", "Distributed Arrays" => "darray.md", "Streaming Tasks" => "streaming.md", diff --git a/docs/src/task-affinity.md b/docs/src/task-affinity.md new file mode 100644 index 00000000..9ee69c15 --- /dev/null +++ b/docs/src/task-affinity.md @@ -0,0 +1,127 @@ +# Task Affinity + +Dagger.jl's `@spawn` macro allows precise control over task execution and result accessibility using `scope`, `compute_scope`, and `result_scope`, which specify various chunk scopes of the task. + +For more information on how these scopes work, see [Scopes](scopes.md#Scopes). + +--- + +## Key Terms + +### Scope +`scope` defines the general set of locations where a Dagger task can execute. If `scope` is not explicitly set, the task runs within the `compute_scope`. If both `scope` and `compute_scope` both are unspecified, the task falls back to `DefaultScope()`, allowing it to run wherever execution is possible. Execution occurs on any worker within the defined scope. + +**Example:** +```julia +g = Dagger.@spawn scope=Dagger.scope(worker=3) f(x,y) +``` +Task `g` executes only on worker 3. Its result can be accessed by any worker. + +--- + +### Compute Scope +Like `scope`,`compute_scope` also specifies where a Dagger task can execute. The key difference is if both `compute_scope` and `scope` are provided, `compute_scope` takes precedence over `scope` for execution placement. If neither is specified, the they default to `DefaultScope()`. + +**Example:** +```julia +g1 = Dagger.@spawn scope=Dagger.scope(worker=2,thread=3) compute_scope=Dagger.scope((worker=1, thread=2), (worker=3, thread=1)) f(x,y) +g2 = Dagger.@spawn compute_scope=Dagger.scope((worker=1, thread=2), (worker=3, thread=1)) f(x,y) +``` +Tasks `g1` and `g2` execute on either thread 2 of worker 1, or thread 1 of worker 3. The `scope` argument to `g1` is ignored. Their result can be accessed by any worker. + +--- + +### Result Scope + +The result_scope limits the workers from which a task's result can be accessed. This is crucial for managing data locality and minimizing transfers. If `result_scope` is not specified, it defaults to `AnyScope()`, meaning the result can be accessed by any worker. + +**Example:** +```julia +g = Dagger.@spawn result_scope=Dagger.scope(worker=3, threads=[1,3, 4]) f(x,y) +``` +The result of `g` is accessible only from threads 1, 3 and 4 of worker process 3. The task's execution may happen anywhere on threads 1, 3 and 4 of worker 3. + +--- + +## Interaction of `compute_scope` and `result_scope` + +When `scope`, `compute_scope`, and `result_scope` are all used, the scheduler executes the task on the intersection of the effective compute scope (which will be `compute_scope` if provided, otherwise `scope`) and the `result_scope`. If the intersection is empty then the scheduler throws a `Dagger.Sch.SchedulerException` error. + +**Example:** +```julia +g = Dagger.@spawn scope=Dagger.scope(worker=3,thread=2) compute_scope=Dagger.scope(worker=2) result_scope=Dagger.scope((worker=2, thread=2), (worker=4, thread=2)) f(x,y) +``` +The task `g` computes on thread 2 of worker 2 (as it's the intersection of compute and result scopes), and its result access is also restricted to thread 2 of worker 2. + +--- + +## Chunk Inputs to Tasks + +This section explains how `scope`, `compute_scope`, and `result_scope` affect tasks when a `Chunk` is the primary input to `@spawn` (e.g. created via `Dagger.tochunk(...)` or by calling `fetch(task; raw=true)` on a task). + +Assume `g` is some function, e.g. `g(x, y) = x * 2 + y * 3`, `chunk_proc` is the chunk's processor, and `chunk_scope` is its defined accessibility. + +When `Dagger.tochunk(...)` is directly spawned: +- The task executes on `chunk_proc`. +- The result is accessible only within `chunk_scope`. +- This behavior occurs irrespective of the `scope`, `compute_scope`, and `result_scope` values provided in the `@spawn` macro. +- Dagger validates that there is an intersection between the effective `compute_scope` (derived from `@spawn`'s `compute_scope` or `scope`) and the `result_scope`. If no intersection exists, the scheduler throws an exception. + +!!! info While `chunk_proc` is currently required when constructing a chunk, it is largely unused in actual scheduling logic. It exists primarily for backward compatibility and may be deprecated in the future. + +**Usage:** +```julia +h1 = Dagger.@spawn scope=Dagger.scope(worker=3) Dagger.tochunk(g(10, 11), chunk_proc, chunk_scope) +h2 = Dagger.@spawn compute_scope=Dagger.scope((worker=1, thread=2), (worker=3, thread=1)) Dagger.tochunk(g(20, 21), chunk_proc, chunk_scope) +h3 = Dagger.@spawn scope=Dagger.scope(worker=2,thread=3) compute_scope=Dagger.scope((worker=1, thread=2), (worker=3, thread=1)) Dagger.tochunk(g(30, 31), chunk_proc, chunk_scope) +h4 = Dagger.@spawn result_scope=Dagger.scope(worker=3) Dagger.tochunk(g(40, 41), chunk_proc, chunk_scope) +h5 = Dagger.@spawn scope=Dagger.scope(worker=3,thread=2) compute_scope=Dagger.ProcessScope(2) result_scope=Dagger.scope(worker=2,threads=[2,3]) Dagger.tochunk(g(50, 51), chunk_proc, chunk_scope) +``` +In all these cases (`h1` through `h5`), the tasks get executed on processor `chunk_proc` of chunk, and its result is accessible only within `chunk_scope`. + +--- + +## Function with Chunk Arguments as Tasks + +This section details behavior when `scope`, `compute_scope`, and `result_scope` are used with tasks where a function is the input, and its arguments include `Chunk`s. + +Assume `g(x, y) = x * 2 + y * 3` is a function, and `arg = Dagger.tochunk(g(1, 2), arg_proc, arg_scope)` is a chunk argument, where `arg_proc` is the chunk's processor and `arg_scope` is its defined scope. + +### Scope +If `arg_scope` and `scope` do not intersect, the scheduler throws an exception. Execution occurs on the intersection of `scope` and `arg_scope`. + +```julia +h = Dagger.@spawn scope=Dagger.scope(worker=3) g(arg, 11) +``` +Task `h` executes on any worker within the intersection of `scope` and `arg_scope`. The result is accessible from any worker. + +--- + +### Compute scope and Chunk argument scopes interaction +If `arg_scope` and `compute_scope` do not intersect, the scheduler throws an exception. Otherwise, execution happens on the intersection of the effective compute scope (which will be `compute_scope` if provided, otherwise `scope`) and `arg_scope`. `result_scope` defaults to `AnyScope()`. + +```julia +h1 = Dagger.@spawn compute_scope=Dagger.scope((worker=1, thread=2), (worker=3, thread=1)) g(arg, 11) +h2 = Dagger.@spawn scope=Dagger.scope(worker=2,thread=3) compute_scope=Dagger.scope((worker=1, thread=2), (worker=3, thread=1)) g(arg, 21) +``` +Tasks `h1` and `h2` execute on any worker within the intersection of the `compute_scope` and `arg_scope`. `scope` is ignored if `compute_scope` is specified. The result is stored and accessible from anywhere. + +--- + +### Result scope and Chunk argument scopes interaction +If only `result_scope` is specified, computation happens on any worker within `arg_scope`, and the result is only accessible from `result_scope`. + +```julia +h = Dagger.@spawn result_scope=Dagger.scope(worker=3) g(arg, 11) +``` +Task `h` executes on any worker within `arg_scope`. The result is accessible from `result_scope`. + +--- + +### Compute, result, and chunk argument scopes interaction +When `scope`, `compute_scope`, and `result_scope` are all used, the scheduler executes the task on the intersection of `arg_scope`, the effective compute scope (which is `compute_scope` if provided, otherwise `scope`), and `result_scope`. If no intersection exists, the scheduler throws an exception. + +```julia +h = Dagger.@spawn scope=Dagger.scope(worker=3,thread=2) compute_scope=Dagger.ProcessScope(2) result_scope=Dagger.scope((worker=2, thread=2), (worker=4, thread=2)) g(arg, 31) +``` +Task `h` computes on thread 2 of worker 2 (as it's the intersection of `arg`, `compute`, and `result` scopes), and its result access is also restricted to thread 2 of worker 2. diff --git a/src/sch/Sch.jl b/src/sch/Sch.jl index b894f452..0335fe3e 100644 --- a/src/sch/Sch.jl +++ b/src/sch/Sch.jl @@ -14,7 +14,7 @@ import Random: randperm import Base: @invokelatest import ..Dagger -import ..Dagger: Context, Processor, Thunk, WeakThunk, ThunkFuture, DTaskFailedException, Chunk, WeakChunk, OSProc, AnyScope, DefaultScope, LockedObject +import ..Dagger: Context, Processor, Thunk, WeakThunk, ThunkFuture, DTaskFailedException, Chunk, WeakChunk, OSProc, AnyScope, DefaultScope, InvalidScope, LockedObject import ..Dagger: order, dependents, noffspring, istask, inputs, unwrap_weak_checked, affinity, tochunk, timespan_start, timespan_finish, procs, move, chunktype, processor, get_processors, get_parent, execute!, rmprocs!, task_processor, constrain, cputhreadtime import ..Dagger: @dagdebug, @safe_lock_spin1 import DataStructures: PriorityQueue, enqueue!, dequeue_pair!, peek @@ -726,16 +726,25 @@ function schedule!(ctx, state, procs=procs_to_use(ctx)) sig = signature(state, task) # Calculate scope - scope = if task.f isa Chunk - task.f.scope - else - if task.options.proclist !== nothing - # proclist overrides scope selection - AnyScope() - else - DefaultScope() + scope = constrain(task.compute_scope, task.result_scope) + if scope isa InvalidScope + ex = SchedulingException("compute_scope and result_scope are not compatible: $(scope.x), $(scope.y)") + state.cache[task] = ex + state.errored[task] = true + set_failed!(state, task) + @goto pop_task + end + if task.f isa Chunk + scope = constrain(scope, task.f.scope) + if scope isa InvalidScope + ex = SchedulingException("Current scope and function Chunk Scope are not compatible: $(scope.x), $(scope.y)") + state.cache[task] = ex + state.errored[task] = true + set_failed!(state, task) + @goto pop_task end end + for (_,input) in task.inputs input = unwrap_weak_checked(input) chunk = if istask(input) @@ -747,8 +756,8 @@ function schedule!(ctx, state, procs=procs_to_use(ctx)) end chunk isa Chunk || continue scope = constrain(scope, chunk.scope) - if scope isa Dagger.InvalidScope - ex = SchedulingException("Scopes are not compatible: $(scope.x), $(scope.y)") + if scope isa InvalidScope + ex = SchedulingException("Current scope and argument Chunk scope are not compatible: $(scope.x), $(scope.y)") state.cache[task] = ex state.errored[task] = true set_failed!(state, task) @@ -1086,7 +1095,7 @@ function fire_tasks!(ctx, thunks::Vector{<:Tuple}, (gproc, proc), state) thunk.get_result, thunk.persist, thunk.cache, thunk.meta, options, propagated, ids, positions, (log_sink=ctx.log_sink, profile=ctx.profile), - sch_handle, state.uid]) + sch_handle, state.uid, thunk.result_scope]) end # N.B. We don't batch these because we might get a deserialization # error due to something not being defined on the worker, and then we don't @@ -1305,7 +1314,7 @@ function start_processor_runner!(istate::ProcessorInternalState, uid::UInt64, re task = task_spec[] scope = task[5] if !isa(constrain(scope, Dagger.ExactScope(to_proc)), - Dagger.InvalidScope) && + InvalidScope) && typemax(UInt32) - proc_occupancy_cached >= occupancy # Compatible, steal this task return dequeue_pair!(queue) @@ -1488,7 +1497,7 @@ function do_task(to_proc, task_desc) scope, Tf, data, send_result, persist, cache, meta, options, propagated, ids, positions, - ctx_vars, sch_handle, sch_uid = task_desc + ctx_vars, sch_handle, sch_uid, result_scope = task_desc ctx = Context(Processor[]; log_sink=ctx_vars.log_sink, profile=ctx_vars.profile) from_proc = OSProc() @@ -1696,7 +1705,7 @@ function do_task(to_proc, task_desc) # Construct result # TODO: We should cache this locally - send_result || meta ? res : tochunk(res, to_proc; device, persist, cache=persist ? true : cache, + send_result || meta ? res : tochunk(res, to_proc, result_scope; device, persist, cache=persist ? true : cache, tag=options.storage_root_tag, leaf_tag=something(options.storage_leaf_tag, MemPool.Tag()), retain=options.storage_retain) diff --git a/src/thunk.jl b/src/thunk.jl index b24806f8..659ee9a2 100644 --- a/src/thunk.jl +++ b/src/thunk.jl @@ -73,6 +73,8 @@ mutable struct Thunk eager_ref::Union{DRef,Nothing} options::Any # stores scheduler-specific options propagates::Tuple # which options we'll propagate + compute_scope::AbstractScope + result_scope::AbstractScope function Thunk(f, xs...; syncdeps=nothing, id::Int=next_id(), @@ -84,16 +86,14 @@ mutable struct Thunk affinity=nothing, eager_ref=nothing, processor=nothing, - scope=nothing, + scope=DefaultScope(), + compute_scope=scope, + result_scope=AnyScope(), options=nothing, propagates=(), kwargs... ) - if !isa(f, Chunk) && (!isnothing(processor) || !isnothing(scope)) - f = tochunk(f, - something(processor, OSProc()), - something(scope, DefaultScope())) - end + xs = Base.mapany(identity, xs) syncdeps_set = Set{Any}(filterany(is_task_or_chunk, Base.mapany(last, xs))) if syncdeps !== nothing @@ -105,11 +105,11 @@ mutable struct Thunk if options !== nothing @assert isempty(kwargs) new(f, xs, syncdeps_set, id, get_result, meta, persist, cache, - cache_ref, affinity, eager_ref, options, propagates) + cache_ref, affinity, eager_ref, options, propagates, compute_scope, result_scope) else new(f, xs, syncdeps_set, id, get_result, meta, persist, cache, cache_ref, affinity, eager_ref, Sch.ThunkOptions(;kwargs...), - propagates) + propagates, compute_scope, result_scope) end end end @@ -476,15 +476,6 @@ function spawn(f, args...; kwargs...) args = args[2:end] end - # Wrap f in a Chunk if necessary - processor = haskey(options, :processor) ? options.processor : nothing - scope = haskey(options, :scope) ? options.scope : nothing - if !isnothing(processor) || !isnothing(scope) - f = tochunk(f, - something(processor, get_options(:processor, OSProc())), - something(scope, get_options(:scope, DefaultScope()))) - end - # Process the args and kwargs into Pair form args_kwargs = args_kwargs_to_pairs(args, kwargs) diff --git a/test/runtests.jl b/test/runtests.jl index 79ba890d..b1282ecc 100644 --- a/test/runtests.jl +++ b/test/runtests.jl @@ -9,6 +9,7 @@ tests = [ ("Options", "options.jl"), ("Mutation", "mutation.jl"), ("Task Queues", "task-queues.jl"), + ("Task Affinity", "task-affinity.jl"), ("Datadeps", "datadeps.jl"), ("Streaming", "streaming.jl"), ("Domain Utilities", "domain.jl"), diff --git a/test/task-affinity.jl b/test/task-affinity.jl new file mode 100644 index 00000000..7a74d095 --- /dev/null +++ b/test/task-affinity.jl @@ -0,0 +1,569 @@ +@testset "Task affinity" begin + + get_compute_scope(x::DTask) = try + Dagger.Sch._find_thunk(x).compute_scope + catch + Dagger.InvalidScope + end + + get_result_scope(x::DTask) = try + fetch(x; raw=true).scope + catch + Dagger.InvalidScope + end + + get_execution_scope(x::DTask) = try + chunk = fetch(x; raw=true) + Dagger.ExactScope(chunk.processor) + catch + Dagger.InvalidScope + end + + function intersect_scopes(scope1::Dagger.AbstractScope, scopes::Dagger.AbstractScope...) + for s in scopes + scope1 = Dagger.constrain(scope1, s) + scope1 isa Dagger.InvalidScope && return (scope1,) + end + return scope1.scopes + end + + availprocs = collect(Dagger.all_processors()) + availscopes = shuffle!(Dagger.ExactScope.(availprocs)) + numscopes = length(availscopes) + + master_proc = Dagger.ThreadProc(1, 1) + master_scope = Dagger.ExactScope(master_proc) + + @testset "Function: scope, compute_scope and result_scope" begin + + @everywhere f(x) = x + 1 + + @testset "scope" begin + scope_only = Dagger.UnionScope(rand(availscopes, rand(1:length(availscopes)))) + + task1 = Dagger.@spawn scope=scope_only f(10); fetch(task1) + @test get_compute_scope(task1) == scope_only + @test get_result_scope(task1) == Dagger.AnyScope() + + execution_scope1 = get_execution_scope(task1) + @test execution_scope1 in intersect_scopes(execution_scope1,scope_only) + end + + @testset "compute_scope" begin + compute_scope_only = Dagger.UnionScope(rand(availscopes, rand(1:length(availscopes)))) + scope = Dagger.UnionScope(rand(availscopes, rand(1:length(availscopes)))) + + task1 = Dagger.@spawn compute_scope=compute_scope_only f(10); fetch(task1) + task2 = Dagger.@spawn scope=scope compute_scope=compute_scope_only f(20); fetch(task2) + + @test get_compute_scope(task1) == get_compute_scope(task2) == compute_scope_only + @test get_result_scope(task1) == get_result_scope(task2) == Dagger.AnyScope() + + execution_scope1 = get_execution_scope(task1) + execution_scope2 = get_execution_scope(task2) + @test execution_scope1 in intersect_scopes(execution_scope1, compute_scope_only) && + execution_scope2 in intersect_scopes(execution_scope2, compute_scope_only) + end + + @testset "result_scope" begin + result_scope_only = Dagger.UnionScope(rand(availscopes, rand(1:length(availscopes)))) + + task1 = Dagger.@spawn result_scope=result_scope_only f(10); fetch(task1) + + @test get_compute_scope(task1) == Dagger.DefaultScope() + @test get_result_scope(task1) == result_scope_only + end + + @testset "compute_scope and result_scope with intersection" begin + if numscopes >= 3 + n = cld(numscopes, 3) + + scope_a = availscopes[1:n] + scope_b = availscopes[n+1:2n] + scope_c = availscopes[2n+1:end] + + compute_scope_intersect = Dagger.UnionScope(scope_a..., scope_b...) + scope_intersect = compute_scope_intersect + scope_rand = Dagger.UnionScope(rand(availscopes, rand(1:length(availscopes)))) + result_scope_intersect = Dagger.UnionScope(scope_b..., scope_c...) + + task1 = Dagger.@spawn compute_scope=compute_scope_intersect result_scope=result_scope_intersect f(10); fetch(task1) + task2 = Dagger.@spawn scope=scope_intersect result_scope=result_scope_intersect f(20); fetch(task2) + task3 = Dagger.@spawn compute_scope=compute_scope_intersect scope=scope_rand result_scope=result_scope_intersect f(30); fetch(task3) + + @test get_compute_scope(task1) == get_compute_scope(task2) == get_compute_scope(task3) == compute_scope_intersect + @test get_result_scope(task1) == get_result_scope(task2) == get_result_scope(task3) == result_scope_intersect + + execution_scope1 = get_execution_scope(task1) + execution_scope2 = get_execution_scope(task2) + execution_scope3 = get_execution_scope(task3) + @test execution_scope1 in intersect_scopes(execution_scope1, compute_scope_intersect, result_scope_intersect) && + execution_scope2 in intersect_scopes(execution_scope2, compute_scope_intersect, result_scope_intersect) && + execution_scope3 in intersect_scopes(execution_scope3, compute_scope_intersect, result_scope_intersect) + end + end + + @testset "compute_scope and result_scope without intersection" begin + if length(availscopes) >= 2 + n = cld(numscopes, 2) + + scope_a = availscopes[1:n] + scope_b = availscopes[n+1:end] + + compute_scope_no_intersect = Dagger.UnionScope(scope_a...) + scope_no_intersect = Dagger.UnionScope(scope_a...) + scope_rand = Dagger.UnionScope(rand(availscopes, rand(1:length(availscopes)))) + result_scope_no_intersect = Dagger.UnionScope(scope_b...) + + task1 = Dagger.@spawn compute_scope=compute_scope_no_intersect result_scope=result_scope_no_intersect f(10); wait(task1) + task2 = Dagger.@spawn scope=scope_no_intersect result_scope=result_scope_no_intersect f(20); wait(task2) + task3 = Dagger.@spawn compute_scope=compute_scope_no_intersect scope=scope_rand result_scope=result_scope_no_intersect f(30); wait(task3) + + @test get_compute_scope(task1) == get_compute_scope(task2) == get_compute_scope(task3) == compute_scope_no_intersect + @test get_result_scope(task1) == get_result_scope(task2) == get_result_scope(task3) == Dagger.InvalidScope + + @test get_execution_scope(task1) == get_execution_scope(task2) == get_execution_scope(task3) == Dagger.InvalidScope + end + end + + end + + @testset "Chunk function: scope, compute_scope and result_scope" begin + + @everywhere g(x, y) = x * 2 + y * 3 + + availscopes = shuffle!(Dagger.ExactScope.(collect(Dagger.all_processors()))) + n = cld(numscopes, 2) + + chunk_scope = Dagger.UnionScope(rand(availscopes, rand(1:numscopes))) + chunk_proc = rand(availprocs) + + @testset "scope" begin + scope_only = Dagger.UnionScope(rand(availscopes, rand(1:numscopes))) + + task1 = Dagger.@spawn scope=scope_only Dagger.tochunk(g(10, 11), chunk_proc, chunk_scope); fetch(task1) + + @test get_compute_scope(task1) == scope_only + @test get_result_scope(task1) == chunk_scope + + execution_scope1 = get_execution_scope(task1) + + @test execution_scope1 == Dagger.ExactScope(chunk_proc) + end + + @testset "compute_scope" begin + compute_scope_only = Dagger.UnionScope(rand(availscopes, rand(1:numscopes))) + scope = Dagger.UnionScope(rand(availscopes, rand(1:numscopes))) + + task1 = Dagger.@spawn compute_scope=compute_scope_only Dagger.tochunk(g(10, 11), chunk_proc, chunk_scope); fetch(task1) + task2 = Dagger.@spawn scope=scope compute_scope=compute_scope_only Dagger.tochunk(g(20, 21), chunk_proc, chunk_scope); fetch(task2) + + @test get_compute_scope(task1) == get_compute_scope(task2) == compute_scope_only + @test get_result_scope(task1) == get_result_scope(task2) == chunk_scope + + execution_scope1 = get_execution_scope(task1) + execution_scope2 = get_execution_scope(task2) + @test execution_scope1 == execution_scope2 == Dagger.ExactScope(chunk_proc) + end + + @testset "result_scope" begin + result_scope_only = Dagger.UnionScope(rand(availscopes, rand(1:numscopes))) + + task1 = Dagger.@spawn result_scope=result_scope_only Dagger.tochunk(g(10, 11), chunk_proc, chunk_scope); fetch(task1) + + @test get_compute_scope(task1) == Dagger.DefaultScope() + @test get_result_scope(task1) == chunk_scope + + execution_scope1 = get_execution_scope(task1) + @test execution_scope1 == Dagger.ExactScope(chunk_proc) + end + + @testset "compute_scope and result_scope with intersection" begin + if length(availscopes) >= 3 + n = cld(numscopes, 3) + + shuffle!(availscopes) + scope_a = availscopes[1:n] + scope_b = availscopes[n+1:2n] + scope_c = availscopes[2n+1:end] + + compute_scope_intersect = Dagger.UnionScope(scope_a..., scope_b...) + scope_intersect = compute_scope_intersect + scope_rand = Dagger.UnionScope(rand(availscopes, rand(1:length(availscopes)))) + result_scope_intersect = Dagger.UnionScope(scope_b..., scope_c...) + + task1 = Dagger.@spawn compute_scope=compute_scope_intersect result_scope=result_scope_intersect Dagger.tochunk(g(10, 11), chunk_proc, chunk_scope); fetch(task1 ) + task2 = Dagger.@spawn scope=scope_intersect result_scope=result_scope_intersect Dagger.tochunk(g(20, 21), chunk_proc, chunk_scope); fetch(task2 ) + task3 = Dagger.@spawn compute_scope=compute_scope_intersect scope=scope_rand result_scope=result_scope_intersect Dagger.tochunk(g(30, 31), chunk_proc, chunk_scope); fetch(task3 ) + + @test get_compute_scope(task1) == get_compute_scope(task2) == get_compute_scope(task3) == compute_scope_intersect + @test get_result_scope(task1) == get_result_scope(task2) == get_result_scope(task3) == chunk_scope + + execution_scope1 = get_execution_scope(task1) + execution_scope2 = get_execution_scope(task2) + execution_scope3 = get_execution_scope(task3) + @test execution_scope1 == execution_scope2 == execution_scope3 == Dagger.ExactScope(chunk_proc) + end + end + + @testset "compute_scope and result_scope without intersection" begin + if length(availscopes) >= 2 + n = cld(length(availscopes), 2) + + shuffle!(availscopes) + scope_a = availscopes[1:n] + scope_b = availscopes[n+1:end] + + compute_scope_no_intersect = Dagger.UnionScope(scope_a...) + scope_no_intersect = Dagger.UnionScope(scope_a...) + scope_rand = Dagger.UnionScope(rand(availscopes, rand(1:length(availscopes)))) + result_scope_no_intersect = Dagger.UnionScope(scope_b...) + + task1 = Dagger.@spawn compute_scope=compute_scope_no_intersect result_scope=result_scope_no_intersect Dagger.tochunk(g(10, 11), chunk_proc, chunk_scope); wait(task1 ) + task2 = Dagger.@spawn scope=scope_no_intersect result_scope=result_scope_no_intersect Dagger.tochunk(g(20, 21), chunk_proc, chunk_scope); wait(task2 ) + task3 = Dagger.@spawn compute_scope=compute_scope_no_intersect scope=scope_rand result_scope=result_scope_no_intersect Dagger.tochunk(g(30, 31), chunk_proc, chunk_scope); wait(task3 ) + + @test get_compute_scope(task1) == get_compute_scope(task2) == get_compute_scope(task3) == compute_scope_no_intersect + @test get_result_scope(task1) == get_result_scope(task2) == get_result_scope(task3) == Dagger.InvalidScope + + execution_scope1 = get_execution_scope(task1) + execution_scope2 = get_execution_scope(task2) + execution_scope3 = get_execution_scope(task3) + @test get_execution_scope(task1) == get_execution_scope(task2) == get_execution_scope(task3) == Dagger.InvalidScope + end + end + + end + + @testset "Chunk arguments: scope, compute_scope and result_scope with non-intersection of chunk arg and scope" begin + + @everywhere g(x, y) = x * 2 + y * 3 + + availscopes = shuffle!(Dagger.ExactScope.(collect(Dagger.all_processors()))) + n = cld(numscopes, 2) + scope_a = availscopes[1:n] + scope_b = availscopes[n+1:end] + + arg_scope = Dagger.UnionScope(scope_a...) + arg_proc = rand(availprocs) + arg = Dagger.tochunk(g(1, 2), arg_proc, arg_scope) + + @testset "scope" begin + scope_only = Dagger.UnionScope(rand(scope_b, rand(1:length(scope_b)))) + + task11 = Dagger.@spawn scope=scope_only g(arg, 11); wait(task11) + + @test get_compute_scope(task11) == scope_only + @test get_result_scope(task11) == Dagger.InvalidScope + + execution_scope11 = get_execution_scope(task11) + + @test execution_scope11 == Dagger.InvalidScope + end + + @testset "compute_scope" begin + compute_scope_only = Dagger.UnionScope(rand(scope_b, rand(1:length(scope_b)))) + scope = Dagger.UnionScope(rand(scope_b, rand(1:length(scope_b)))) + + task11 = Dagger.@spawn compute_scope=compute_scope_only g(arg, 11); wait(task11) + task21 = Dagger.@spawn scope=scope compute_scope=compute_scope_only g(arg, 21); wait(task21) + + @test get_compute_scope(task11) == get_compute_scope(task21) == compute_scope_only + @test get_result_scope(task11) == get_result_scope(task21) == Dagger.InvalidScope + + execution_scope11 = get_execution_scope(task11) + execution_scope21 = get_execution_scope(task21) + @test execution_scope11 == execution_scope21 == Dagger.InvalidScope + end + + @testset "result_scope" begin + result_scope_only = Dagger.UnionScope(rand(scope_b, rand(1:length(scope_b)))) + + task11 = Dagger.@spawn result_scope=result_scope_only g(arg, 11); wait(task11) + + @test get_compute_scope(task11) == Dagger.DefaultScope() + @test get_result_scope(task11) == Dagger.InvalidScope + + execution_scope11 = get_execution_scope(task11) + @test execution_scope11 == Dagger.InvalidScope + end + + @testset "compute_scope and result_scope with intersection" begin + if length(scope_b) >= 3 + n = cld(length(scope_b), 3) + + scope_ba = scope_b[1:n] + scope_bb = scope_b[n+1:2n] + scope_bc = scope_b[2n+1:end] + + compute_scope_intersect = Dagger.UnionScope(scope_ba..., scope_bb...) + scope_intersect = compute_scope_intersect + scope_rand = Dagger.UnionScope(rand(availscopes, rand(1:length(availscopes)))) + result_scope_intersect = Dagger.UnionScope(scope_bb..., scope_bc...) + + task11 = Dagger.@spawn compute_scope=compute_scope_intersect result_scope=result_scope_intersect g(arg, 11); wait(task11 ) + task21 = Dagger.@spawn scope=scope_intersect result_scope=result_scope_intersect g(arg, 21); wait(task21 ) + task31 = Dagger.@spawn compute_scope=compute_scope_intersect scope=scope_rand result_scope=result_scope_intersect g(arg, 31); wait(task31 ) + + @test get_compute_scope(task11) == get_compute_scope(task21) == get_compute_scope(task31) == compute_scope_intersect + @test get_result_scope(task11) == get_result_scope(task21) == get_result_scope(task31) == Dagger.InvalidScope + + execution_scope11 = get_execution_scope(task11) + execution_scope21 = get_execution_scope(task21) + execution_scope31 = get_execution_scope(task31) + @test execution_scope11 == execution_scope21 == execution_scope31 == Dagger.InvalidScope + end + end + + @testset "compute_scope and result_scope without intersection" begin + if length(scope_b) >= 2 + n = cld(length(scope_b), 2) + + scope_ba = scope_b[1:n] + scope_bb = scope_b[n+1:end] + + compute_scope_no_intersect = Dagger.UnionScope(scope_ba...) + scope_no_intersect = Dagger.UnionScope(scope_ba...) + scope_rand = Dagger.UnionScope(rand(availscopes, rand(1:length(availscopes)))) + result_scope_no_intersect = Dagger.UnionScope(scope_bb...) + + task11 = Dagger.@spawn compute_scope=compute_scope_no_intersect result_scope=result_scope_no_intersect g(arg, 11); ; wait(task11 ) + task21 = Dagger.@spawn scope=scope_no_intersect result_scope=result_scope_no_intersect g(arg, 21); ; wait(task21 ) + task31 = Dagger.@spawn compute_scope=compute_scope_no_intersect scope=scope_rand result_scope=result_scope_no_intersect g(arg, 31); wait(task31 ) + + @test get_compute_scope(task11) == get_compute_scope(task21) == get_compute_scope(task31) == compute_scope_no_intersect + @test get_result_scope(task11) == get_result_scope(task21) == get_result_scope(task31) == Dagger.InvalidScope + + execution_scope11 = get_execution_scope(task11) + execution_scope21 = get_execution_scope(task21) + execution_scope31 = get_execution_scope(task31) + @test get_execution_scope(task11) == get_execution_scope(task21) == get_execution_scope(task31) == Dagger.InvalidScope + end + end + + end + + @testset "Chunk arguments: scope, compute_scope and result_scope with intersection of chunk arg and scope" begin + + @everywhere g(x, y) = x * 2 + y * 3 + + availscopes = shuffle!(Dagger.ExactScope.(collect(Dagger.all_processors()))) + n = cld(numscopes, 3) + scope_a = availscopes[1:n] + scope_b = availscopes[n+1:2n] + scope_c = availscopes[2n+1:end] + + arg_scope = Dagger.UnionScope(scope_a..., scope_b...) + arg_proc = rand(availprocs) + arg = Dagger.tochunk(g(1, 2), arg_proc, arg_scope) + + @testset "scope" begin + scope_only = Dagger.UnionScope(rand(scope_b, rand(1:length(scope_b)))..., rand(scope_c, rand(1:length(scope_c)))...) + + task11 = Dagger.@spawn scope=scope_only g(arg, 11); fetch(task11) + + @test get_compute_scope(task11) == scope_only + @test get_result_scope(task11) == Dagger.AnyScope() + + execution_scope11 = get_execution_scope(task11) + + @test execution_scope11 in intersect_scopes(execution_scope11, scope_only, arg_scope) + end + + @testset "compute_scope" begin + compute_scope_only = Dagger.UnionScope(rand(scope_b, rand(1:length(scope_b)))..., rand(scope_c, rand(1:length(scope_c)))...) + scope = Dagger.UnionScope(rand(scope_b, rand(1:length(scope_b)))..., rand(scope_c, rand(1:length(scope_c)))...) + + task11 = Dagger.@spawn compute_scope=compute_scope_only g(arg, 11); fetch(task11) + task21 = Dagger.@spawn scope=scope compute_scope=compute_scope_only g(arg, 21); fetch(task21) + + @test get_compute_scope(task11) == get_compute_scope(task21) == compute_scope_only + @test get_result_scope(task11) == get_result_scope(task21) == Dagger.AnyScope() + + execution_scope11 = get_execution_scope(task11) + execution_scope21 = get_execution_scope(task21) + @test execution_scope11 in intersect_scopes(execution_scope11, compute_scope_only, arg_scope) && + execution_scope11 in intersect_scopes(execution_scope11, compute_scope_only, arg_scope) + end + + @testset "result_scope" begin + result_scope_only = Dagger.UnionScope(rand(scope_b, rand(1:length(scope_b)))..., rand(scope_c, rand(1:length(scope_c)))...) + + task11 = Dagger.@spawn result_scope=result_scope_only g(arg, 11); fetch(task11) + + @test get_compute_scope(task11) == Dagger.DefaultScope() + @test get_result_scope(task11) == result_scope_only + + execution_scope11 = get_execution_scope(task11) + @test execution_scope11 in intersect_scopes(execution_scope11, result_scope_only, arg_scope) + end + + @testset "compute_scope and result_scope with intersection" begin + scope_bc = [scope_b...,scope_c...] + if length(scope_bc) >= 3 + n = cld(length(scope_bc), 3) + + scope_bca = scope_bc[1:n] + scope_bcb = scope_bc[n+1:2n] + scope_bcc = scope_bc[2n+1:end] + + compute_scope_intersect = Dagger.UnionScope(scope_bca..., scope_bcb...) + scope_intersect = compute_scope_intersect + scope_rand = Dagger.UnionScope(rand(availscopes, rand(1:length(availscopes)))) + result_scope_intersect = Dagger.UnionScope(scope_bcb..., scope_bcc...) + + task11 = Dagger.@spawn compute_scope=compute_scope_intersect result_scope=result_scope_intersect g(arg, 11); fetch(task11 ) + task21 = Dagger.@spawn scope=scope_intersect result_scope=result_scope_intersect g(arg, 21); fetch(task21 ) + task31 = Dagger.@spawn compute_scope=compute_scope_intersect scope=scope_rand result_scope=result_scope_intersect g(arg, 31); fetch(task31 ) + + @test get_compute_scope(task11) == get_compute_scope(task21) == get_compute_scope(task31) == compute_scope_intersect + @test get_result_scope(task11) == get_result_scope(task21) == get_result_scope(task31) == result_scope_intersect + + execution_scope11 = get_execution_scope(task11) + execution_scope21 = get_execution_scope(task21) + execution_scope31 = get_execution_scope(task31) + @test execution_scope11 in intersect_scopes(execution_scope11, compute_scope_intersect, result_scope_intersect, arg_scope) && + execution_scope21 in intersect_scopes(execution_scope21, scope_intersect, result_scope_intersect, arg_scope) && + execution_scope31 in intersect_scopes(execution_scope31, compute_scope_intersect, result_scope_intersect, arg_scope) + end + end + + @testset "compute_scope and result_scope without intersection" begin + scope_bc = [scope_b...,scope_c...] + if length(scope_bc) >= 2 + n = cld(length(scope_bc), 2) + + scope_bca = scope_bc[1:n] + scope_bcb = scope_bc[n+1:end] + + compute_scope_no_intersect = Dagger.UnionScope(scope_bca...) + scope_no_intersect = Dagger.UnionScope(scope_bca...) + scope_rand = Dagger.UnionScope(rand(availscopes, rand(1:length(availscopes)))) + result_scope_no_intersect = Dagger.UnionScope(scope_bcb...) + + task11 = Dagger.@spawn compute_scope=compute_scope_no_intersect result_scope=result_scope_no_intersect g(arg, 11); wait(task11 ) + task21 = Dagger.@spawn scope=scope_no_intersect result_scope=result_scope_no_intersect g(arg, 21); wait(task21 ) + task31 = Dagger.@spawn compute_scope=compute_scope_no_intersect scope=scope_rand result_scope=result_scope_no_intersect g(arg, 31); wait(task31 ) + + @test get_compute_scope(task11) == get_compute_scope(task21) == get_compute_scope(task31) == compute_scope_no_intersect + @test get_result_scope(task11) == get_result_scope(task21) == get_result_scope(task31) == Dagger.InvalidScope + + execution_scope11 = get_execution_scope(task11) + execution_scope21 = get_execution_scope(task21) + execution_scope31 = get_execution_scope(task31) + @test get_execution_scope(task11) == get_execution_scope(task21) == get_execution_scope(task31) == Dagger.InvalidScope + end + end + + end + + # @testset "Chunk function with Chunk arguments: scope, compute_scope and result_scope with non-intersection of chunk arg and chunk scope" begin + + # @everywhere g(x, y) = x * 2 + y * 3 + + # availscopes = shuffle!(Dagger.ExactScope.(collect(Dagger.all_processors()))) + + # chunk_scope = Dagger.UnionScope(rand(availscopes, rand(1:numscopes))) + # chunk_proc = rand(chunk_scope.scopes).processor + # arg_scope = Dagger.UnionScope(rand(availscopes, rand(1:numscopes))) + # arg_proc = rand(arg_scope.scopes).processor + # arg = Dagger.tochunk(g(1, 2), arg_proc, arg_scope) + + # @testset "scope" begin + # scope_only = Dagger.UnionScope(rand(availscopes, rand(1:numscopes))) + + # task11 = Dagger.@spawn scope=scope_only arg -> Dagger.tochunk(g(arg, 11), chunk_proc, chunk_scope); fetch(task11) + + # @test get_compute_scope(task11) == scope_only + # @test get_result_scope(task11) == chunk_scope + + # execution_scope11 = get_execution_scope(task11) + + # @test execution_scope11 == Dagger.ExactScope(chunk_proc) + # end + + # @testset "compute_scope" begin + # compute_scope_only = Dagger.UnionScope(rand(availscopes, rand(1:numscopes))) + # scope = Dagger.UnionScope(rand(availscopes, rand(1:numscopes))) + + # task11 = Dagger.@spawn compute_scope=compute_scope_only arg -> Dagger.tochunk(g(arg, 11), chunk_proc, chunk_scope); fetch(task11) + # task21 = Dagger.@spawn scope=scope compute_scope=compute_scope_only arg -> Dagger.tochunk(g(arg, 21), chunk_proc, chunk_scope); fetch(task21) + + # @test get_compute_scope(task11) == get_compute_scope(task21) == compute_scope_only + # @test get_result_scope(task11) == get_result_scope(task21) == chunk_scope + + # execution_scope11 = get_execution_scope(task11) + # execution_scope21 = get_execution_scope(task21) + # @test execution_scope11 == execution_scope21 == Dagger.ExactScope(chunk_proc) + # end + + # @testset "result_scope" begin + # result_scope_only = Dagger.UnionScope(rand(availscopes, rand(1:numscopes))) + + # task11 = Dagger.@spawn result_scope=result_scope_only arg -> Dagger.tochunk(g(arg, 11), chunk_proc, chunk_scope); fetch(task11) + + # @test get_compute_scope(task11) == Dagger.DefaultScope() + # @test get_result_scope(task11) == chunk_scope + + # execution_scope11 = get_execution_scope(task11) + # @test execution_scope11 == Dagger.ExactScope(chunk_proc) + # end + + # @testset "compute_scope and result_scope with intersection" begin + # if numscopes >= 3 + # n = cld(numscopes, 3) + + # shuffle!(availscopes) + # scope_a = availscopes[1:n] + # scope_b = availscopes[n+1:2n] + # scope_c = availscopes[2n+1:end] + + # compute_scope_intersect = Dagger.UnionScope(scope_a..., scope_b...) + # scope_intersect = compute_scope_intersect + # scope_rand = Dagger.UnionScope(rand(availscopes, rand(1:length(availscopes)))) + # result_scope_intersect = Dagger.UnionScope(scope_b..., scope_c...) + + # task11 = Dagger.@spawn compute_scope=compute_scope_intersect result_scope=result_scope_intersect arg -> Dagger.tochunk(g(arg, 11), chunk_proc, chunk_scope); wait(task11 ) + # task21 = Dagger.@spawn scope=scope_intersect result_scope=result_scope_intersect arg -> Dagger.tochunk(g(arg, 21), chunk_proc, chunk_scope); wait(task21 ) + # task31 = Dagger.@spawn compute_scope=compute_scope_intersect scope=scope_rand result_scope=result_scope_intersect arg -> Dagger.tochunk(g(arg, 31), chunk_proc, chunk_scope); wait(task31 ) + + # @test get_compute_scope(task11) == get_compute_scope(task21) == get_compute_scope(task31) == compute_scope_intersect + # @test get_result_scope(task11) == get_result_scope(task21) == get_result_scope(task31) == chunk_scope + + # execution_scope11 = get_execution_scope(task11) + # execution_scope21 = get_execution_scope(task21) + # execution_scope31 = get_execution_scope(task31) + # @test execution_scope11 == execution_scope21 == execution_scope31 == Dagger.ExactScope(chunk_proc) + # end + # end + + # @testset "compute_scope and result_scope without intersection" begin + # if numscopes >= 2 + # n = cld(numscopes, 2) + + # shuffle!(availscopes) + # scope_a = availscopes[1:n] + # scope_b = availscopes[n+1:end] + + # compute_scope_no_intersect = Dagger.UnionScope(scope_a...) + # scope_no_intersect = Dagger.UnionScope(scope_a...) + # scope_rand = Dagger.UnionScope(rand(availscopes, rand(1:length(availscopes)))) + # result_scope_no_intersect = Dagger.UnionScope(scope_b...) + + # task11 = Dagger.@spawn compute_scope=compute_scope_no_intersect result_scope=result_scope_no_intersect arg -> Dagger.tochunk(g(arg, 11), chunk_proc, chunk_scope); wait(task11 ) + # task21 = Dagger.@spawn scope=scope_no_intersect result_scope=result_scope_no_intersect arg -> Dagger.tochunk(g(arg, 21), chunk_proc, chunk_scope); wait(task21 ) + # task31 = Dagger.@spawn compute_scope=compute_scope_no_intersect scope=scope_rand result_scope=result_scope_no_intersect arg -> Dagger.tochunk(g(arg, 31), chunk_proc, chunk_scope); wait(task31 ) + + # @test get_compute_scope(task11) == get_compute_scope(task21) == get_compute_scope(task31) == compute_scope_no_intersect + # @test get_result_scope(task11) == get_result_scope(task21) == get_result_scope(task31) == Dagger.InvalidScope + + # execution_scope11 = get_execution_scope(task11) + # execution_scope21 = get_execution_scope(task21) + # execution_scope31 = get_execution_scope(task31) + # @test get_execution_scope(task11) == get_execution_scope(task21) == get_execution_scope(task31) == Dagger.InvalidScope + # end + # end + + # end + +end \ No newline at end of file