Add eager versions of addprocs!()/rmprocs!()

Author: JamesWrigley

src/context.jl

@@ -1,3 +1,5 @@
+export Context, addprocs!, rmprocs!
+
 """
     Context(xs::Vector{OSProc}) -> Context
     Context(xs::Vector{Int}) -> Context
@@ -62,35 +64,104 @@ procs(ctx::Context) = lock(ctx) do
 end
 
 """
-    addprocs!(ctx::Context, xs)
+    addprocs!(xs)
 
-Add new workers `xs` to `ctx`.
+Add new workers `xs` to the eager scheduler.
 
 Workers will typically be assigned new tasks in the next scheduling iteration
 if scheduling is ongoing.
 
 Workers can be either `Processor`s or the underlying process IDs as `Integer`s.
 """
-addprocs!(ctx::Context, xs::AbstractVector{<:Integer}) = addprocs!(ctx, map(OSProc, xs))
-function addprocs!(ctx::Context, xs::AbstractVector{<:OSProc})
+addprocs!(xs::AbstractVector{<:Integer}) = addprocs!(map(OSProc, xs))
+function addprocs!(xs::AbstractVector{<:OSProc})
+    ctx = Sch.eager_context()
+    state = Sch.EAGER_STATE[]
+
+    timespan_start(ctx, :addprocs!, nothing, nothing)
+
+    # Initialize new procs
+    for p in xs
+        Sch.init_proc(state, p, ctx.log_sink)
+
+        # Empty the processor cache list and force reschedule
+        lock(state.lock) do
+            state.procs_cache_list[] = nothing
+        end
+        put!(state.chan, Sch.RescheduleSignal())
+    end
+
     lock(ctx) do
         append!(ctx.procs, xs)
     end
     lock(ctx.proc_notify) do
         notify(ctx.proc_notify)
     end
+
+    timespan_finish(ctx, :addprocs!, nothing, nothing)
 end
 
 """
-    rmprocs!(ctx::Context, xs)
+    rmprocs!(xs)
 
-Remove the specified workers `xs` from `ctx`.
+Remove the specified workers `xs` from the eager scheduler.
 
 Workers will typically finish all their assigned tasks if scheduling is ongoing
 but will not be assigned new tasks after removal.
 
 Workers can be either `Processor`s or the underlying process IDs as `Integer`s.
 """
+rmprocs!(xs::AbstractVector{<:Integer}) = rmprocs!(map(OSProc, xs))
+function rmprocs!(xs::AbstractVector{<:OSProc})
+    ctx = Sch.eager_context()
+    state = Sch.EAGER_STATE[]
+
+    timespan_start(ctx, :rmprocs!, nothing, nothing)
+
+    for p in xs
+        Sch.cleanup_proc(state, p, ctx.log_sink)
+
+        # Empty the processor cache list
+        lock(state.lock) do
+            state.procs_cache_list[] = nothing
+        end
+    end
+
+    lock(ctx) do
+        filter!(p -> (p ∉ xs), ctx.procs)
+    end
+    lock(ctx.proc_notify) do
+        notify(ctx.proc_notify)
+    end
+    timespan_finish(ctx, :rmprocs!, nothing, nothing)
+end
+
+"""
+    addprocs!(ctx::Context, xs)
+
+The lazy API version of [`addprocs!(xs)`](@ref).
+
+!!! warning
+    The lazy API is buggy and unsupported, prefer using the eager API.
+"""
+addprocs!(ctx::Context, xs::AbstractVector{<:Integer}) = addprocs!(ctx, map(OSProc, xs))
+function addprocs!(ctx::Context, xs::AbstractVector{<:OSProc})
+    lock(ctx) do
+        append!(ctx.procs, xs)
+    end
+    lock(ctx.proc_notify) do
+        notify(ctx.proc_notify)
+    end
+end
+
+"""
+    rmprocs!(ctx::Context, xs)
+
+The lazy API version of [`rmprocs!(xs)`](@ref).
+
+!!! warning
+    The lazy API is buggy and unsupported, prefer using the eager API.
+"""
 rmprocs!(ctx::Context, xs::AbstractVector{<:Integer}) = rmprocs!(ctx, map(OSProc, xs))
 function rmprocs!(ctx::Context, xs::AbstractVector{<:OSProc})
     lock(ctx) do

src/processor.jl

@@ -1,4 +1,4 @@
-export OSProc, Context, addprocs!, rmprocs!
+export OSProc
 
 import Base: @invokelatest