Merge pull request #374 from JuliaParallel/jps/processor-type-scope

Add ProcessorTypeScope, deprecate proclist and single

Author: jpsamaroo

docs/src/index.md

@@ -198,15 +198,17 @@ While Dagger generally "just works", sometimes one needs to exert some more
 fine-grained control over how the scheduler allocates work. There are two
 parallel mechanisms to achieve this: Scheduler options (from
 `Dagger.Sch.SchedulerOptions`) and Thunk options (from
-`Dagger.Sch.ThunkOptions`). These two options structs generally contain the
-same options, with the difference being that Scheduler options operate
+`Dagger.Sch.ThunkOptions`). These two options structs contain many shared
+options, with the difference being that Scheduler options operate
 globally across an entire DAG, and Thunk options operate on a thunk-by-thunk
-basis. Scheduler options can be constructed and passed to `collect()` or
-`compute()` as the keyword argument `options` for lazy API usage:
+basis.
+
+Scheduler options can be constructed and passed to `collect()` or `compute()`
+as the keyword argument `options` for lazy API usage:
 
 ```julia
 t = @par 1+2
-opts = Dagger.Sch.ThunkOptions(;single=1) # Execute on worker 1
+opts = Dagger.Sch.SchedulerOptions(;single=1) # Execute on worker 1
 
 compute(t; options=opts)
 
@@ -219,7 +221,6 @@ Thunk options can be passed to `@spawn/spawn`, `@par`, and `delayed` similarly:
 # Execute on worker 1
 
 Dagger.@spawn single=1 1+2
-
 Dagger.spawn(+, 1, 2; single=1)
 
 opts = Dagger.Sch.ThunkOptions(;single=1)

docs/src/processors.md

@@ -76,42 +76,13 @@ processor B. This mechanism uses Julia's Serialization library to serialize and
 deserialize data, so data must be serializable for this mechanism to work
 properly.
 
-### Future: Hierarchy Generic Path Move
-
-NOTE: This used to be the default move behavior, but was removed because it
-wasn't considered helpful, and there were not any processor implementations
-that made use of it.
-
-Movement of data between any two processors is decomposable into a sequence of
-"moves" between a child and its parent, termed a "generic path move". Movement
-of data may also take "shortcuts" between nodes in the tree which are not
-directly connected if enabled by libraries or the user, which may make use of
-IPC mechanisms to transfer data more directly and efficiently (such as
-Infiniband, GPU RDMA, NVLINK, etc.). All data is considered local to some
-processor, and may only be operated on by another processor by first doing an
-explicit move operation to that processor.
-
 ## Processor Selection
 
 By default, Dagger uses the CPU to process work, typically single-threaded per
 cluster node. However, Dagger allows access to a wider range of hardware and
 software acceleration techniques, such as multithreading and GPUs. These more
 advanced (but performant) accelerators are disabled by default, but can easily
-be enabled by using Scheduler/Thunk options in the `proclist` field. If
-`nothing`, all default processors will be used. If a vector of types, only the
-processor types contained in `options.proclist` will be used to compute all or
-a given thunk. If a function, it will be called for each processor (with the
-processor as the argument) until it returns `true`.
-
-```julia
-opts = Dagger.Sch.ThunkOptions(;proclist=nothing) # default behavior
-# OR
-opts = Dagger.Sch.ThunkOptions(;proclist=[DaggerGPU.CuArrayProc]) # only execute on CuArrayProc
-# OR
-opts = Dagger.Sch.ThunkOptions(;proclist=(proc)->(proc isa Dagger.ThreadProc && proc.tid == 3)) # only run on ThreadProc with thread ID 3
-
-t = Dagger.@par options=opts sum(X) # do sum(X) on the specified processor
-```
+be enabled by using scopes (see [Scopes](@ref) for details).
 
 ## Resource Control
 
@@ -137,7 +108,7 @@ sufficient resources become available by thunks completing execution.
 The [DaggerGPU.jl](https://github.com/JuliaGPU/DaggerGPU.jl) package can be
 imported to enable GPU acceleration for NVIDIA and AMD GPUs, when available.
 The processors provided by that package are not enabled by default, but may be
-enabled via `options.proclist` as usual.
+enabled via custom scopes ([Scopes](@ref)).
 
 ### Future: Network Devices and Topology
 

docs/src/scopes.md

@@ -15,16 +15,16 @@ considered valid.
 
 Let's take the example of a webcam handle generated by VideoIO.jl. This handle
 is a C pointer, and thus has process scope. We can open the handle on a given
-process, and set the scope of the resulting data to a `ProcessScope()`, which
-defaults to the current Julia process:
+process, and set the scope of the resulting data to be locked to the current
+process with `Dagger.scope` to construct a `ProcessScope`:
 
 ```julia
-using VideoIO
+using VideoIO, Distributed
 
 function get_handle()
     handle = VideoIO.opencamera()
     proc = Dagger.thunk_processor()
-    scope = ProcessScope()
+    scope = Dagger.scope(worker=myid()) # constructs a `ProcessScope`
     return Dagger.tochunk(handle, proc, scope)
 end
 
@@ -41,7 +41,7 @@ cam_frame = Dagger.@spawn read(cam_handle)
 
 The `cam_frame` task is executed within any processor on the same process that
 the `cam_handle` task was executed on. Of course, the resulting camera frame is
-*not* scoped to anywhere specific (denoted as `AnyScope()`), and thus
+*not* scoped to anywhere specific (denoted as `AnyScope`), and thus
 computations on it may execute anywhere.
 
 You may also encounter situations where you want to use a callable struct (such
@@ -53,13 +53,15 @@ using Flux
 m = Chain(...)
 # If `m` is only safe to transfer to and execute on this process,
 # we can set a `ProcessScope` on it:
-result = Dagger.@spawn scope=ProcessScope() m(rand(8,8))
+result = Dagger.@spawn scope=Dagger.scope(worker=myid()) m(rand(8,8))
 ```
 
 Setting a scope on the function treats it as a regular piece of data (like the
 arguments to the function), so it participates in the scoping rules described
 in the following sections all the same.
 
+[`Dagger.scope`](@ref)
+
 Now, let's try out some other kinds of scopes, starting with `NodeScope`. This
 scope encompasses the server that one or more Julia processes may be running
 on. Say we want to use memory mapping (mmap) to more efficiently send arrays
@@ -75,6 +77,7 @@ function generate()
     arr = Mmap.mmap(path, Matrix{Int}, (64,64))
     fill!(arr, 1)
     Mmap.sync!(arr)
+    # Note: Dagger.scope() does not yet support node scopes
     Dagger.tochunk(path, Dagger.thunk_processor(), NodeScope())
 end
 
@@ -87,14 +90,14 @@ a = Dagger.@spawn generate()
 @assert fetch(Dagger.@spawn consume(a)) == 64*64
 ```
 
-Whatever server `a` executed on, `b` will also execute on!
+Whatever server `a` executed on, `b` will also execute on it!
 
 Finally, we come to the "lowest" scope on the scope hierarchy, the
 `ExactScope`. This scope specifies one exact processor as the bounding scope,
-and is typically useful in certain limited cases. We won't provide an example
-here, because you don't usually need to ever use this scope, but if you already
-understand the `NodeScope` and `ProcessScope`, the `ExactScope` should be easy
-to figure out.
+and is typically useful in certain limited cases (such as data existing only on
+a specific GPU). We won't provide an example here, because you don't usually
+need to ever use this scope, but if you already understand the `NodeScope` and
+`ProcessScope`, the `ExactScope` should be easy to figure out.
 
 ## Union Scopes
 

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, AnyScope
+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
-            AnyScope()
+            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
@@ -693,7 +699,7 @@ function schedule!(ctx, state, procs=procs_to_use(ctx))
                 end
             end
         end
-        state.cache[task] = SchedulingException("No processors available, try making proclist more liberal")
+        state.cache[task] = SchedulingException("No processors available, try widening scope")
         state.errored[task] = true
         set_failed!(state, task)
         @goto pop_task
@@ -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
@@ -723,14 +730,15 @@ function schedule!(ctx, state, procs=procs_to_use(ctx))
                 if procs_found
                     push!(failed_scheduling, task)
                 else
-                    state.cache[task] = SchedulingException("No processors available, try making proclist more liberal")
+                    state.cache[task] = SchedulingException("No processors available, try widening scope")
                     state.errored[task] = true
                     set_failed!(state, task)
                 end
                 @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

@@ -14,7 +14,7 @@ function get_propagated_options(thunk)
     nt = NamedTuple()
     for key in thunk.propagates
         value = if key == :scope
-            isa(thunk.f, Chunk) ? thunk.f.scope : AnyScope()
+            isa(thunk.f, Chunk) ? thunk.f.scope : DefaultScope()
         elseif key == :processor
             isa(thunk.f, Chunk) ? thunk.f.processor : OSProc()
         elseif key in fieldnames(Thunk)
@@ -263,41 +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 "Rejected $proc: gproc.pid != single"
-            return false
+            @debug "[$(task.id)] Rejected $proc: gproc.pid ($(gproc.pid)) != single ($(opts.single))"
+            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 "Rejected $proc: Task scope ($scope) vs. processor scope ($proc_scope)"
-        return false
+        @debug "[$(task.id)] Rejected $proc: Not contained in task scope ($scope)"
+        return false, scope
     end
 
-    return true
+    @label accept
+
+    @debug "[$(task.id)] Accepted $proc"
+    return true, scope
 end
 
 function has_capacity(state, p, gp, time_util, alloc_util, sig)