Merge pull request #218 from JuliaParallel/jps/simple-osproc

Simplify OSProc

Author: jpsamaroo

src/Dagger.jl

@@ -65,10 +65,10 @@ function __init__()
     end
     @static if VERSION >= v"1.3.0-DEV.573"
         for tid in 1:Threads.nthreads()
-            push!(PROCESSOR_CALLBACKS, proc->ThreadProc(myid(), tid))
+            push!(PROCESSOR_CALLBACKS, ()->ThreadProc(myid(), tid))
         end
     else
-        push!(PROCESSOR_CALLBACKS, proc->ThreadProc(myid(), 1))
+        push!(PROCESSOR_CALLBACKS, ()->ThreadProc(myid(), 1))
     end
 end
 

src/processor.jl

@@ -87,88 +87,52 @@ capacity(proc, ::Type{T}) where T =
 """
     OSProc <: Processor
 
-Julia CPU (OS) process, identified by Distributed pid. Executes thunks when
-threads or other "children" processors are not available, and/or the user has
-not opted to use those processors.
+Julia CPU (OS) process, identified by Distributed pid. The logical parent of
+all processors on a given node, but otherwise does not participate in
+computations.
 """
 struct OSProc <: Processor
     pid::Int
-    attrs::Dict{Symbol,Any}
-    children::Vector{Processor}
-    queue::Vector{Processor}
-end
-const OSPROC_CACHE = Dict{Int,OSProc}()
-OSProc(pid::Int=myid()) = get!(OSPROC_CACHE, pid) do
-    remotecall_fetch(get_osproc, pid, pid)
+    function OSProc(pid::Int=myid())
+        get!(OSPROC_CACHE, pid) do
+            remotecall_fetch(get_proc_hierarchy, pid)
+        end
+        new(pid)
+    end
 end
-function get_osproc(pid::Int)
-    proc = OSProc(pid, Dict{Symbol,Any}(), Processor[], Processor[])
+const OSPROC_CACHE = Dict{Int,Vector{Processor}}()
+children(proc::OSProc) = OSPROC_CACHE[proc.pid]
+function get_proc_hierarchy()
+    children = Processor[]
     for cb in PROCESSOR_CALLBACKS
         try
-            child = Base.invokelatest(cb, proc)
-            child !== nothing && push!(proc.children, child)
+            child = Base.invokelatest(cb)
+            if (child isa Tuple) || (child isa Vector)
+                append!(children, child)
+            elseif child !== nothing
+                push!(children, child)
+            end
         catch err
             @error "Error in processor callback" exception=(err,catch_backtrace())
         end
     end
-    proc
+    children
 end
 function add_callback!(func)
     push!(Dagger.PROCESSOR_CALLBACKS, func)
     empty!(OSPROC_CACHE)
 end
 Base.:(==)(proc1::OSProc, proc2::OSProc) = proc1.pid == proc2.pid
 iscompatible(proc::OSProc, opts, f, args...) =
-    any(child->iscompatible(child, opts, f, args...), proc.children)
+    any(child->iscompatible(child, opts, f, args...), children(proc))
 iscompatible_func(proc::OSProc, opts, f) =
-    any(child->iscompatible_func(child, opts, f), proc.children)
+    any(child->iscompatible_func(child, opts, f), children(proc))
 iscompatible_arg(proc::OSProc, opts, args...) =
     any(child->
         all(arg->iscompatible_arg(child, opts, arg), args),
-    proc.children)
+    children(proc))
 get_processors(proc::OSProc) =
-    vcat((get_processors(child) for child in proc.children)...,)
-function choose_processor(options, f, Targs)
-    osproc = OSProc()
-    if isempty(osproc.queue)
-        for child in osproc.children
-            grandchildren = get_processors(child)
-            append!(osproc.queue, grandchildren)
-        end
-    end
-    @assert !isempty(osproc.queue)
-    for i in 1:length(osproc.queue)
-        proc = popfirst!(osproc.queue)
-        push!(osproc.queue, proc)
-        if !iscompatible(proc, options, f, Targs...)
-            continue
-        end
-        if options.proclist === nothing
-            default_enabled(proc) && return proc
-        elseif options.proclist isa Function
-            options.proclist(proc) && return proc
-        elseif any(p->proc isa p, options.proclist)
-            return proc
-        end
-    end
-    throw(ProcessorSelectionException(options.proclist, osproc.queue, f, Targs))
-end
-struct ProcessorSelectionException <: Exception
-    proclist
-    procsavail::Vector{Processor}
-    f
-    args
-end
-function Base.show(io::IO, pex::ProcessorSelectionException)
-    println(io, "(Worker $(myid())) Exhausted all available processor types!")
-    println(io, "  Proclist: $(pex.proclist)")
-    println(io, "  Procs Available: $(pex.procsavail)")
-    println(io, "  Function: $(pex.f)")
-    print(io, "  Arguments: $(pex.args)")
-end
-
-execute!(proc::OSProc, f, args...) = f(args...)
-default_enabled(proc::OSProc) = true
+    vcat((get_processors(child) for child in children(proc))...)
 
 """
     ThreadProc <: Processor

src/sch/Sch.jl

@@ -649,7 +649,7 @@ end
             # Under-subscribed, calculate extra utilization and execute thunk
             @debug "($(myid())) Using available $to_proc: $extra_util | $(real_util[])/$cap"
             extra_util = if extra_util isa MaxUtilization
-                count(c->typeof(c)===typeof(to_proc), from_proc.children)
+                count(c->typeof(c)===typeof(to_proc), children(from_proc))
             else
                 extra_util
             end

test/processors.jl

@@ -32,7 +32,6 @@ end
         end
     end
     @testset "Opt-in/Opt-out" begin
-        @test Dagger.default_enabled(OSProc()) == true
         @test Dagger.default_enabled(ThreadProc(1,1)) == true
         @test Dagger.default_enabled(OptOutProc()) == false
     end
@@ -54,12 +53,12 @@ end
     end
     @testset "Add callback in same world" begin
         function addcb()
-            @everywhere Dagger.add_callback!(cb)
-            cb = eval(Dagger, :(proc->PathProc(myid())))
-            opts = ThunkOptions(proclist=[PathProc])
-            collect(delayed(identity; options=opts)(1.0))
-            @everywhere pop!(Dagger.PROCESSOR_CALLBACKS)
+            cb = @eval ()->FakeProc(myid())
+            @everywhere Dagger.add_callback!($cb)
+            @test any(x->x isa FakeProc, Dagger.children(OSProc()))
+            @everywhere pop!(Dagger.PROCESSOR_CALLBACKS); empty!(Dagger.OSPROC_CACHE)
         end
+        addcb()
     end
 
     @testset "Modify workers in Context" begin

test/runtests.jl

@@ -8,12 +8,12 @@ include("util.jl")
 include("fakeproc.jl")
 
 include("thunk.jl")
-include("domain.jl")
-include("array.jl")
 include("scheduler.jl")
 include("processors.jl")
 include("ui.jl")
 include("checkpoint.jl")
+include("domain.jl")
+include("array.jl")
 try # TODO: Fault tolerance is sometimes unreliable
 include("fault-tolerance.jl")
 catch

test/scheduler.jl

@@ -150,7 +150,7 @@ end
                 @test collect(Context(), a) == 1
             end
         end
-        @everywhere Dagger.add_callback!(proc->FakeProc())
+        @everywhere Dagger.add_callback!(()->FakeProc())
         @testset "proclist FakeProc" begin
             @test Dagger.iscompatible_arg(FakeProc(), nothing, Int) == true
             @test Dagger.iscompatible_arg(FakeProc(), nothing, FakeVal) == true