Merge pull request #640 from JuliaParallel/jps/constrain-perf

scopes/Sch: Add optimized check for proc in scope

Author: jpsamaroo

src/sch/Sch.jl

@@ -1101,8 +1101,7 @@ function start_processor_runner!(istate::ProcessorInternalState, uid::UInt64, re
                         end
                         task, occupancy = peek(queue)
                         scope = task.scope
-                        if !isa(constrain(scope, Dagger.ExactScope(to_proc)),
-                                InvalidScope) &&
+                        if Dagger.proc_in_scope(to_proc, scope)
                            typemax(UInt32) - proc_occupancy_cached >= occupancy
                             # Compatible, steal this task
                             return dequeue_pair!(queue)

src/sch/util.jl

@@ -447,8 +447,7 @@ function can_use_proc(state, task, gproc, proc, opts, scope)
     end
 
     # Check against scope
-    proc_scope = Dagger.ExactScope(proc)
-    if constrain(scope, proc_scope) isa Dagger.InvalidScope
+    if !Dagger.proc_in_scope(proc, scope)
         @dagdebug task :scope "Rejected $proc: Not contained in task scope ($scope)"
         return false, scope
     end

src/scopes.jl

@@ -4,8 +4,11 @@ abstract type AbstractScope end
 
 "Widest scope that contains all processors."
 struct AnyScope <: AbstractScope end
+proc_in_scope(::Processor, ::AnyScope) = true
 
 abstract type AbstractScopeTaint end
+proc_in_scope(proc::Processor, scope::AbstractScope) =
+    !isa(constrain(scope, ExactScope(proc)), InvalidScope)
 
 "Taints a scope for later evaluation."
 struct TaintScope <: AbstractScope
@@ -44,6 +47,8 @@ UnionScope(scopes...) = UnionScope((scopes...,))
 UnionScope(scopes::Vector{<:AbstractScope}) = UnionScope((scopes...,))
 UnionScope(s::AbstractScope) = UnionScope((s,))
 UnionScope() = UnionScope(())
+proc_in_scope(proc::Processor, scope::UnionScope) =
+    any(subscope->proc_in_scope(proc, subscope), scope.scopes)
 
 function Base.:(==)(us1::UnionScope, us2::UnionScope)
     if length(us1.scopes) != length(us2.scopes)
@@ -78,6 +83,8 @@ function ProcessScope(wid::Integer)
 end
 ProcessScope(p::OSProc) = ProcessScope(p.pid)
 ProcessScope() = ProcessScope(myid())
+proc_in_scope(proc::Processor, scope::ProcessScope) =
+    root_worker_id(proc) == scope.wid
 
 struct ProcessorTypeTaint{T} <: AbstractScopeTaint end
 
@@ -92,12 +99,14 @@ struct ExactScope <: AbstractScope
     processor::Processor
 end
 ExactScope(proc) = ExactScope(ProcessScope(get_parent(proc).pid), proc)
+proc_in_scope(proc::Processor, scope::ExactScope) = proc == scope.processor
 
 "Indicates that the applied scopes `x` and `y` are incompatible."
 struct InvalidScope <: AbstractScope
     x::AbstractScope
     y::AbstractScope
 end
+proc_in_scope(::Processor, ::InvalidScope) = false
 
 # Show methods