Rename the WorkerState instances and add an exterminated state

The new `WorkerState_exterminated` state is for indicating that a worker was
killed by something other than us.

Author: JamesWrigley

src/cluster.jl

@@ -100,7 +100,15 @@ mutable struct WorkerConfig
     end
 end
 
-@enum WorkerState W_CREATED W_CONNECTED W_TERMINATING W_TERMINATED W_UNKNOWN_STATE
+@enum WorkerState begin
+    WorkerState_created
+    WorkerState_connected
+    WorkerState_terminating    # rmprocs() has been called on the worker
+    WorkerState_terminated     # Worker was gracefully removed
+    WorkerState_exterminated   # Worker was forcefully removed (not by us)
+    WorkerState_unknown
+end
+
 mutable struct Worker
     id::Int
     msg_lock::Threads.ReentrantLock # Lock for del_msgs, add_msgs, and gcflag
@@ -131,7 +139,7 @@ mutable struct Worker
         w.manager = manager
         w.config = config
         w.version = version
-        set_worker_state(w, W_CONNECTED)
+        set_worker_state(w, WorkerState_connected)
         register_worker_streams(w)
         w
     end
@@ -142,7 +150,7 @@ mutable struct Worker
         @lock map_pid_wrkr if haskey(map_pid_wrkr[], id)
             return map_pid_wrkr[][id]
         end
-        w=new(id, Threads.ReentrantLock(), [], [], false, W_CREATED, Threads.Condition(), time(), conn_func)
+        w=new(id, Threads.ReentrantLock(), [], [], false, WorkerState_created, Threads.Condition(), time(), conn_func)
         w.initialized = Event()
         register_worker(w)
         w
@@ -158,8 +166,15 @@ function set_worker_state(w, state)
     end
 end
 
+# Helper function to check if a worker is dead or not. It's recommended to use
+# this instead of checking Worker.state manually.
+function is_worker_dead(w::Worker)
+    state = @atomic w.state
+    return state === WorkerState_terminated || state === WorkerState_exterminated
+end
+
 function check_worker_state(w::Worker)
-    if (@atomic w.state) === W_CREATED
+    if (@atomic w.state) === WorkerState_created
         if !isclusterlazy()
             if PGRP.topology === :all_to_all
                 # Since higher pids connect with lower pids, the remote worker
@@ -198,7 +213,7 @@ function exec_conn_func(w::Worker)
 end
 
 function wait_for_conn(w)
-    if (@atomic w.state) === W_CREATED
+    if (@atomic w.state) === WorkerState_created
         timeout =  worker_timeout() - (time() - w.ct_time)
         timeout <= 0 && error("peer $(w.id) has not connected to $(myid())")
 
@@ -211,7 +226,7 @@ function wait_for_conn(w)
         errormonitor(T)
         lock(w.c_state) do
             wait(w.c_state)
-            (@atomic w.state) === W_CREATED && error("peer $(w.id) didn't connect to $(myid()) within $timeout seconds")
+            (@atomic w.state) === WorkerState_created && error("peer $(w.id) didn't connect to $(myid()) within $timeout seconds")
         end
     end
     nothing
@@ -673,7 +688,7 @@ function create_worker(manager::ClusterManager, wconfig::WorkerConfig)
             if (jw.id != 1) && (jw.id < w.id)
                 lock(jw.c_state) do
                     # wait for wl to join
-                    if (@atomic jw.state) === W_CREATED
+                    if (@atomic jw.state) === WorkerState_created
                         wait(jw.c_state)
                     end
                 end
@@ -700,7 +715,7 @@ function create_worker(manager::ClusterManager, wconfig::WorkerConfig)
 
         for wl in wlist
             lock(wl.c_state) do
-                if (@atomic wl.state) === W_CREATED
+                if (@atomic wl.state) === WorkerState_created
                     # wait for wl to join
                     wait(wl.c_state)
                 end
@@ -918,7 +933,7 @@ function nprocs()
         n = length(PGRP.workers)
         # filter out workers in the process of being setup/shutdown.
         for jw in PGRP.workers
-            if !isa(jw, LocalProcess) && ((@atomic jw.state) !== W_CONNECTED)
+            if !isa(jw, LocalProcess) && ((@atomic jw.state) !== WorkerState_connected)
                 n = n - 1
             end
         end
@@ -971,7 +986,7 @@ julia> procs()
 function procs()
     if myid() == 1 || (PGRP.topology === :all_to_all  && !isclusterlazy())
         # filter out workers in the process of being setup/shutdown.
-        return Int[x.id for x in PGRP.workers if isa(x, LocalProcess) || ((@atomic x.state) === W_CONNECTED)]
+        return Int[x.id for x in PGRP.workers if isa(x, LocalProcess) || ((@atomic x.state) === WorkerState_connected)]
     else
         return Int[x.id for x in PGRP.workers]
     end
@@ -988,7 +1003,7 @@ other_procs() = filter(!=(myid()), procs())
 function id_in_procs(id)  # faster version of `id in procs()`
     if myid() == 1 || (PGRP.topology === :all_to_all  && !isclusterlazy())
         for x in PGRP.workers
-            if (x.id::Int) == id && (isa(x, LocalProcess) || (@atomic (x::Worker).state) === W_CONNECTED)
+            if (x.id::Int) == id && (isa(x, LocalProcess) || (@atomic (x::Worker).state) === WorkerState_connected)
                 return true
             end
         end
@@ -1012,7 +1027,7 @@ See also [`other_procs()`](@ref).
 """
 function procs(pid::Integer)
     if myid() == 1
-        all_workers = [x for x in PGRP.workers if isa(x, LocalProcess) || ((@atomic x.state) === W_CONNECTED)]
+        all_workers = [x for x in PGRP.workers if isa(x, LocalProcess) || ((@atomic x.state) === WorkerState_connected)]
         if (pid == 1) || (isa(@lock(map_pid_wrkr, map_pid_wrkr[][pid].manager), LocalManager))
             Int[x.id for x in filter(w -> (w.id==1) || (isa(w.manager, LocalManager)), all_workers)]
         else
@@ -1121,7 +1136,7 @@ function _rmprocs(pids, waitfor)
             else
                 w = @lock map_pid_wrkr get(map_pid_wrkr[], p, nothing)
                 if !isnothing(w)
-                    set_worker_state(w, W_TERMINATING)
+                    set_worker_state(w, WorkerState_terminating)
                     kill(w.manager, p, w.config)
                     push!(rmprocset, w)
                 end
@@ -1130,11 +1145,11 @@ function _rmprocs(pids, waitfor)
 
         start = time_ns()
         while (time_ns() - start) < waitfor*1e9
-            all(w -> (@atomic w.state) === W_TERMINATED, rmprocset) && break
+            all(is_worker_dead, rmprocset) && break
             sleep(min(0.1, waitfor - (time_ns() - start)/1e9))
         end
 
-        unremoved = [wrkr.id for wrkr in filter(w -> (@atomic w.state) !== W_TERMINATED, rmprocset)]
+        unremoved = [wrkr.id for wrkr in filter(!is_worker_dead, rmprocset)]
         if length(unremoved) > 0
             estr = string("rmprocs: pids ", unremoved, " not terminated after ", waitfor, " seconds.")
             throw(ErrorException(estr))

src/messages.jl

@@ -194,7 +194,7 @@ end
 function flush_gc_msgs()
     try
         for w in (PGRP::ProcessGroup).workers
-            if isa(w,Worker) && ((@atomic w.state) == W_CONNECTED) && w.gcflag
+            if isa(w,Worker) && ((@atomic w.state) == WorkerState_connected) && w.gcflag
                 flush_gc_msgs(w)
             end
         end

src/process_messages.jl

@@ -210,7 +210,7 @@ function message_handler_loop(r_stream::IO, w_stream::IO, incoming::Bool)
             handle_msg(msg, header, r_stream, w_stream, version)
         end
     catch e
-        oldstate = W_UNKNOWN_STATE
+        oldstate = WorkerState_unknown
 
         # Check again as it may have been set in a message handler but not propagated to the calling block above
         if wpid < 1
@@ -223,7 +223,7 @@ function message_handler_loop(r_stream::IO, w_stream::IO, incoming::Bool)
         elseif @lock(map_del_wrkr, !(wpid in map_del_wrkr[]))
             werr = worker_from_id(wpid)
             oldstate = @atomic werr.state
-            set_worker_state(werr, W_TERMINATED)
+            set_worker_state(werr, oldstate != WorkerState_terminating ? WorkerState_exterminated : WorkerState_terminated)
 
             # If unhandleable error occurred talking to pid 1, exit
             if wpid == 1
@@ -243,7 +243,7 @@ function message_handler_loop(r_stream::IO, w_stream::IO, incoming::Bool)
         close(w_stream)
 
         if (myid() == 1) && (wpid > 1)
-            if oldstate != W_TERMINATING
+            if oldstate != WorkerState_terminating
                 println(stderr, "Worker $wpid terminated.")
                 rethrow()
             end