@@ -151,7 +151,7 @@ function set_worker_state(w, state)
151
151
end
152
152
153
153
function check_worker_state (w:: Worker )
154
- if w. state === W_CREATED
154
+ if ( @atomic w. state) === W_CREATED
155
155
if ! isclusterlazy ()
156
156
if PGRP. topology === :all_to_all
157
157
# Since higher pids connect with lower pids, the remote worker
@@ -190,7 +190,7 @@ function exec_conn_func(w::Worker)
190
190
end
191
191
192
192
function wait_for_conn (w)
193
- if w. state === W_CREATED
193
+ if ( @atomic w. state) === W_CREATED
194
194
timeout = worker_timeout () - (time () - w. ct_time)
195
195
timeout <= 0 && error (" peer $(w. id) has not connected to $(myid ()) " )
196
196
@@ -654,7 +654,7 @@ function create_worker(manager, wconfig)
654
654
for jw in PGRP. workers
655
655
if (jw. id != 1 ) && (jw. id < w. id)
656
656
# wait for wl to join
657
- if jw. state === W_CREATED
657
+ if ( @atomic jw. state) === W_CREATED
658
658
lock (jw. c_state) do
659
659
wait (jw. c_state)
660
660
end
@@ -682,7 +682,7 @@ function create_worker(manager, wconfig)
682
682
683
683
for wl in wlist
684
684
lock (wl. c_state) do
685
- if wl. state === W_CREATED
685
+ if ( @atomic wl. state) === W_CREATED
686
686
# wait for wl to join
687
687
wait (wl. c_state)
688
688
end
@@ -884,7 +884,7 @@ function nprocs()
884
884
n = length (PGRP. workers)
885
885
# filter out workers in the process of being setup/shutdown.
886
886
for jw in PGRP. workers
887
- if ! isa (jw, LocalProcess) && (jw. state != = W_CONNECTED)
887
+ if ! isa (jw, LocalProcess) && (( @atomic jw. state) != = W_CONNECTED)
888
888
n = n - 1
889
889
end
890
890
end
@@ -935,7 +935,7 @@ julia> procs()
935
935
function procs ()
936
936
if myid () == 1 || (PGRP. topology === :all_to_all && ! isclusterlazy ())
937
937
# filter out workers in the process of being setup/shutdown.
938
- return Int[x. id for x in PGRP. workers if isa (x, LocalProcess) || (x. state === W_CONNECTED)]
938
+ return Int[x. id for x in PGRP. workers if isa (x, LocalProcess) || (( @atomic x. state) === W_CONNECTED)]
939
939
else
940
940
return Int[x. id for x in PGRP. workers]
941
941
end
944
944
function id_in_procs (id) # faster version of `id in procs()`
945
945
if myid () == 1 || (PGRP. topology === :all_to_all && ! isclusterlazy ())
946
946
for x in PGRP. workers
947
- if (x. id:: Int ) == id && (isa (x, LocalProcess) || (x:: Worker ). state === W_CONNECTED)
947
+ if (x. id:: Int ) == id && (isa (x, LocalProcess) || (@atomic ( x:: Worker ). state) === W_CONNECTED)
948
948
return true
949
949
end
950
950
end
@@ -966,7 +966,7 @@ Specifically all workers bound to the same ip-address as `pid` are returned.
966
966
"""
967
967
function procs (pid:: Integer )
968
968
if myid () == 1
969
- all_workers = [x for x in PGRP. workers if isa (x, LocalProcess) || (x. state === W_CONNECTED)]
969
+ all_workers = [x for x in PGRP. workers if isa (x, LocalProcess) || (( @atomic x. state) === W_CONNECTED)]
970
970
if (pid == 1 ) || (isa (map_pid_wrkr[pid]. manager, LocalManager))
971
971
Int[x. id for x in filter (w -> (w. id== 1 ) || (isa (w. manager, LocalManager)), all_workers)]
972
972
else
@@ -1073,11 +1073,11 @@ function _rmprocs(pids, waitfor)
1073
1073
1074
1074
start = time_ns ()
1075
1075
while (time_ns () - start) < waitfor* 1e9
1076
- all (w -> w. state === W_TERMINATED, rmprocset) && break
1076
+ all (w -> ( @atomic w. state) === W_TERMINATED, rmprocset) && break
1077
1077
sleep (min (0.1 , waitfor - (time_ns () - start)/ 1e9 ))
1078
1078
end
1079
1079
1080
- unremoved = [wrkr. id for wrkr in filter (w -> w. state != = W_TERMINATED, rmprocset)]
1080
+ unremoved = [wrkr. id for wrkr in filter (w -> ( @atomic w. state) != = W_TERMINATED, rmprocset)]
1081
1081
if length (unremoved) > 0
1082
1082
estr = string (" rmprocs: pids " , unremoved, " not terminated after " , waitfor, " seconds." )
1083
1083
throw (ErrorException (estr))
0 commit comments