diff --git a/src/cluster.jl b/src/cluster.jl index b899c1c..0ce2494 100644 --- a/src/cluster.jl +++ b/src/cluster.jl @@ -9,6 +9,14 @@ Cluster managers implement how workers can be added, removed and communicated wi """ abstract type ClusterManager end +# cluster_manager is a global constant +const cluster_manager = Ref{ClusterManager}() + +function throw_if_cluster_manager_unassigned() + isassigned(cluster_manager) || error("cluster_manager is unassigned") + return nothing +end + """ WorkerConfig @@ -379,8 +387,7 @@ function init_worker(cookie::AbstractString, manager::ClusterManager=DefaultClus # On workers, the default cluster manager connects via TCP sockets. Custom # transports will need to call this function with their own manager. - global cluster_manager - cluster_manager = manager + cluster_manager[] = manager # Since our pid has yet to be set, ensure no RemoteChannel / Future have been created or addprocs() called. @assert nprocs() <= 1 diff --git a/src/process_messages.jl b/src/process_messages.jl index 6a4b8c2..393c764 100644 --- a/src/process_messages.jl +++ b/src/process_messages.jl @@ -315,8 +315,10 @@ function handle_msg(msg::ResultMsg, header, r_stream, w_stream, version) end function handle_msg(msg::IdentifySocketMsg, header, r_stream, w_stream, version) + throw_if_cluster_manager_unassigned() + # register a new peer worker connection - w = Worker(msg.from_pid, r_stream, w_stream, cluster_manager; version=version)::Worker + w = Worker(msg.from_pid, r_stream, w_stream, cluster_manager[]; version=version)::Worker send_connection_hdr(w, false) send_msg_now(w, MsgHeader(), IdentifySocketAckMsg()) notify(w.initialized) @@ -328,8 +330,10 @@ function handle_msg(msg::IdentifySocketAckMsg, header, r_stream, w_stream, versi end function handle_msg(msg::JoinPGRPMsg, header, r_stream, w_stream, version) + throw_if_cluster_manager_unassigned() + LPROC.id = msg.self_pid - controller = Worker(1, r_stream, w_stream, cluster_manager; version=version)::Worker + controller = Worker(1, r_stream, w_stream, cluster_manager[]; version=version)::Worker notify(controller.initialized) register_worker(LPROC) topology(msg.topology) @@ -348,9 +352,9 @@ function handle_msg(msg::JoinPGRPMsg, header, r_stream, w_stream, version) let rpid=rpid, wconfig=wconfig if lazy # The constructor registers the object with a global registry. - Worker(rpid, ()->connect_to_peer(cluster_manager, rpid, wconfig)) + Worker(rpid, ()->connect_to_peer(cluster_manager[], rpid, wconfig)) else - @async connect_to_peer(cluster_manager, rpid, wconfig) + @async connect_to_peer(cluster_manager[], rpid, wconfig) end end end