Deprecate proclist and single in favor of scope

Also changes behavior such that proclist and single override scope when
set, to prevent issues with mixing proclist/single with scope.

Author: jpsamaroo

src/sch/Sch.jl

@@ -9,7 +9,7 @@ import Random: randperm
 import Base: @invokelatest
 
 import ..Dagger
-import ..Dagger: Context, Processor, Thunk, WeakThunk, ThunkFuture, ThunkFailedException, Chunk, WeakChunk, OSProc, DefaultScope
+import ..Dagger: Context, Processor, Thunk, WeakThunk, ThunkFuture, ThunkFailedException, Chunk, WeakChunk, OSProc, AnyScope, DefaultScope
 import ..Dagger: order, dependents, noffspring, istask, inputs, unwrap_weak_checked, affinity, tochunk, timespan_start, timespan_finish, procs, move, chunktype, processor, default_enabled, get_processors, get_parent, execute!, rmprocs!, addprocs!, thunk_processor, constrain, cputhreadtime
 
 const OneToMany = Dict{Thunk, Set{Thunk}}
@@ -143,13 +143,13 @@ end
 Stores DAG-global options to be passed to the Dagger.Sch scheduler.
 
 # Arguments
-- `single::Int=0`: Force all work onto worker with specified id. `0` disables
-this option.
-- `proclist=nothing`: Force scheduler to use one or more processors that are
-instances/subtypes of a contained type. Alternatively, a function can be
-supplied, and the function will be called with a processor as the sole
-argument and should return a `Bool` result to indicate whether or not to use
-the given processor. `nothing` enables all default processors.
+- `single::Int=0`: (Deprecated) Force all work onto worker with specified id.
+`0` disables this option.
+- `proclist=nothing`: (Deprecated) Force scheduler to use one or more
+processors that are instances/subtypes of a contained type. Alternatively, a
+function can be supplied, and the function will be called with a processor as
+the sole argument and should return a `Bool` result to indicate whether or not
+to use the given processor. `nothing` enables all default processors.
 - `allow_errors::Bool=true`: Allow thunks to error without affecting
 non-dependent thunks.
 - `checkpoint=nothing`: If not `nothing`, uses the provided function to save
@@ -176,11 +176,11 @@ end
 Stores Thunk-local options to be passed to the Dagger.Sch scheduler.
 
 # Arguments
-- `single::Int=0`: Force thunk onto worker with specified id. `0` disables this
-option.
-- `proclist=nothing`: Force thunk to use one or more processors that are
-instances/subtypes of a contained type. Alternatively, a function can be
-supplied, and the function will be called with a processor as the sole
+- `single::Int=0`: (Deprecated) Force thunk onto worker with specified id. `0`
+disables this option.
+- `proclist=nothing`: (Deprecated) Force thunk to use one or more processors
+that are instances/subtypes of a contained type. Alternatively, a function can
+be supplied, and the function will be called with a processor as the sole
 argument and should return a `Bool` result to indicate whether or not to use
 the given processor. `nothing` enables all default processors.
 - `time_util::Dict{Type,Any}=Dict{Type,Any}()`: Indicates the maximum expected
@@ -634,7 +634,12 @@ function schedule!(ctx, state, procs=procs_to_use(ctx))
         scope = if task.f isa Chunk
             task.f.scope
         else
-            DefaultScope()
+            if task.options.proclist !== nothing
+                # proclist overrides scope selection
+                AnyScope()
+            else
+                DefaultScope()
+            end
         end
         for input in task.inputs
             input = unwrap_weak_checked(input)
@@ -682,7 +687,8 @@ function schedule!(ctx, state, procs=procs_to_use(ctx))
 
         for proc in local_procs
             gproc = get_parent(proc)
-            if can_use_proc(task, gproc, proc, opts, scope)
+            can_use, scope = can_use_proc(task, gproc, proc, opts, scope)
+            if can_use
                 has_cap, est_time_util, est_alloc_util = has_capacity(state, proc, gproc.pid, opts.time_util, opts.alloc_util, sig)
                 if has_cap
                     # Schedule task onto proc
@@ -706,7 +712,8 @@ function schedule!(ctx, state, procs=procs_to_use(ctx))
         cap, extra_util = nothing, nothing
         procs_found = false
         # N.B. if we only have one processor, we need to select it now
-        if can_use_proc(task, entry.gproc, entry.proc, opts, scope)
+        can_use, scope = can_use_proc(task, entry.gproc, entry.proc, opts, scope)
+        if can_use
             has_cap, est_time_util, est_alloc_util = has_capacity(state, entry.proc, entry.gproc.pid, opts.time_util, opts.alloc_util, sig)
             if has_cap
                 selected_entry = entry
@@ -730,7 +737,8 @@ function schedule!(ctx, state, procs=procs_to_use(ctx))
                 @goto pop_task
             end
 
-            if can_use_proc(task, entry.gproc, entry.proc, opts, scope)
+            can_use, scope = can_use_proc(task, entry.gproc, entry.proc, opts, scope)
+            if can_use
                 has_cap, est_time_util, est_alloc_util = has_capacity(state, entry.proc, entry.gproc.pid, opts.time_util, opts.alloc_util, sig)
                 if has_cap
                     # Select this processor

src/sch/eager.jl

@@ -18,12 +18,12 @@ function init_eager()
     ctx = eager_context()
     @async try
         sopts = SchedulerOptions(;allow_errors=true)
-        topts = ThunkOptions(;single=1)
+        scope = Dagger.ExactScope(Dagger.ThreadProc(1, 1))
         atexit() do
             EAGER_FORCE_KILL[] = true
             close(EAGER_THUNK_CHAN)
         end
-        Dagger.compute(ctx, Dagger.delayed(eager_thunk; options=topts)(); options=sopts)
+        Dagger.compute(ctx, Dagger.delayed(eager_thunk; scope)(); options=sopts)
     catch err
         iob = IOContext(IOBuffer(), :color=>true)
         println(iob, "Error in eager scheduler:")

src/sch/util.jl

@@ -263,43 +263,55 @@ end
 
 function can_use_proc(task, gproc, proc, opts, scope)
     # Check against proclist
-    if opts.proclist === nothing
-        if !default_enabled(proc)
-            @debug "Rejected $proc: !default_enabled(proc)"
-            return false
-        end
-    elseif opts.proclist isa Function
-        if !Base.invokelatest(opts.proclist, proc)
-            @debug "Rejected $proc: proclist(proc) == false"
-            return false
+    if opts.proclist !== nothing
+        @warn "The `proclist` option is deprecated, please use scopes instead\nSee https://juliaparallel.org/Dagger.jl/stable/scopes/ for details" maxlog=1
+        if opts.proclist isa Function
+            if !Base.invokelatest(opts.proclist, proc)
+                @debug "[$(task.id)] Rejected $proc: proclist(proc) == false"
+                return false, scope
+            end
+            scope = constrain(scope, Dagger.ExactScope(proc))
+        elseif opts.proclist isa Vector
+            if !(typeof(proc) in opts.proclist)
+                @debug "[$(task.id)] Rejected $proc: !(typeof(proc) in proclist)"
+                return false, scope
+            end
+            scope = constrain(scope,
+                              Dagger.UnionScope(map(Dagger.ProcessorTypeScope, opts.proclist)))
+        else
+            throw(SchedulingException("proclist must be a Function, Vector, or nothing"))
         end
-    elseif opts.proclist isa Vector
-        if !(typeof(proc) in opts.proclist)
-            @debug "Rejected $proc: !(typeof(proc) in proclist)"
-            return false
+        if scope isa Dagger.InvalidScope
+            @debug "[$(task.id)] Rejected $proc: Not contained in task scope ($scope)"
+            return false, scope
         end
-    else
-        throw(SchedulingException("proclist must be a Function, Vector, or nothing"))
     end
 
     # Check against single
     if opts.single !== nothing
+        @warn "The `single` option is deprecated, please use scopes instead\nSee https://juliaparallel.org/Dagger.jl/stable/scopes/ for details" maxlog=1
         if gproc.pid != opts.single
             @debug "[$(task.id)] Rejected $proc: gproc.pid ($(gproc.pid)) != single ($(opts.single))"
-            return false
+            return false, scope
+        end
+        scope = constrain(scope, Dagger.ProcessScope(opts.single))
+        if scope isa Dagger.InvalidScope
+            @debug "[$(task.id)] Rejected $proc: Not contained in task scope ($scope)"
+            return false, scope
         end
     end
 
-    # Check scope
+    # Check against scope
     proc_scope = Dagger.ExactScope(proc)
     if constrain(scope, proc_scope) isa Dagger.InvalidScope
         @debug "[$(task.id)] Rejected $proc: Not contained in task scope ($scope)"
-        return false
+        return false, scope
     end
 
-    @debug "[$(task.id)] Accepted $proc"
+    @label accept
 
-    return true
+    @debug "[$(task.id)] Accepted $proc"
+    return true, scope
 end
 
 function has_capacity(state, p, gp, time_util, alloc_util, sig)

test/fakeproc.jl

@@ -1,7 +1,7 @@
 @everywhere begin
 
 using Dagger
-import Dagger: ThreadProc
+import Dagger: OSProc, ThreadProc
 
 struct FakeProc <: Dagger.Processor
     owner::Int
@@ -18,6 +18,7 @@ fakesum(xs...) = FakeVal(sum(map(y->y.x, xs)))
 Dagger.iscompatible_func(proc::FakeProc, opts, f) = true
 Dagger.iscompatible_arg(proc::FakeProc, opts, ::Type{<:Integer}) = true
 Dagger.iscompatible_arg(proc::FakeProc, opts, ::Type{<:FakeVal}) = true
+Dagger.move(from_proc::OSProc, to_proc::FakeProc, x::Integer) = FakeVal(x)
 Dagger.move(from_proc::ThreadProc, to_proc::FakeProc, x::Integer) = FakeVal(x)
 Dagger.execute!(proc::FakeProc, func, args...) = FakeVal(42+func(args...).x)
 

test/processors.jl

@@ -37,9 +37,9 @@ end
     end
     @testset "Processor exhaustion" begin
         opts = ThunkOptions(proclist=[OptOutProc])
-        @test_throws_unwrap Dagger.ThunkFailedException ex isa Dagger.Sch.SchedulingException ex.reason="No processors available, try making proclist more liberal" collect(delayed(sum; options=opts)([1,2,3]))
+        @test_throws_unwrap Dagger.ThunkFailedException ex isa Dagger.Sch.SchedulingException ex.reason="No processors available, try widening scope" collect(delayed(sum; options=opts)([1,2,3]))
         opts = ThunkOptions(proclist=(proc)->false)
-        @test_throws_unwrap Dagger.ThunkFailedException ex isa Dagger.Sch.SchedulingException ex.reason="No processors available, try making proclist more liberal" collect(delayed(sum; options=opts)([1,2,3]))
+        @test_throws_unwrap Dagger.ThunkFailedException ex isa Dagger.Sch.SchedulingException ex.reason="No processors available, try widening scope" collect(delayed(sum; options=opts)([1,2,3]))
         opts = ThunkOptions(proclist=nothing)
         @test collect(delayed(sum; options=opts)([1,2,3])) == 6
     end