incorporating last commits of the original version: af89e6c (Aug 14, 2024) -> 3b889ee (Apr 9, 2024)

Author: decarvalhojunior-fh

.github/workflows/ci.yml

@@ -34,12 +34,12 @@ jobs:
           - os: macOS-latest
             arch: x86
     steps:
-      - uses: actions/checkout@v2
-      - uses: julia-actions/setup-julia@v1
+      - uses: actions/checkout@v4
+      - uses: julia-actions/setup-julia@v2
         with:
           version: ${{ matrix.version }}
           arch: ${{ matrix.arch }}
-      - uses: actions/cache@v1
+      - uses: actions/cache@v4
         env:
           cache-name: cache-artifacts
         with:
@@ -54,13 +54,14 @@ jobs:
         env:
           JULIA_DISTRIBUTED_TESTING_STANDALONE: 1
       - uses: julia-actions/julia-processcoverage@v1
-      - uses: codecov/codecov-action@v1
+      - uses: codecov/codecov-action@v4
         with:
           file: lcov.info
+          token: ${{ secrets.CODECOV_TOKEN }}
   docs:
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v2
+      - uses: actions/checkout@v4
       - uses: julia-actions/setup-julia@latest
         with:
           # version: '1.6'

README.md

@@ -1,6 +1,6 @@
 # Distributed (with a multiscale parallelism extension)
 
-The `Distributed` package provides functionality for creating and controlling multiple Julia processes remotely, and for performing distributed and parallel computing. It uses network sockets or other supported interfaces to communicate between Julia processes, and relies on Julia's `Serialization` stdlib package to transform Julia objects into a format that can be transferred between processes efficiently. It provides a full set of utilities to create and destroy new Julia processes and add them to a "cluster" (a collection of Julia processes connected together), as well as functions to perform Remote Procedure Calls (RPC) between the processes within a cluster. See [`API`](@ref) for details.
+The `Distributed` package provides functionality for creating and controlling multiple Julia processes remotely, and for performing distributed and parallel computing. It uses network sockets or other supported interfaces to communicate between Julia processes, and relies on Julia's `Serialization` stdlib package to transform Julia objects into a format that can be transferred between processes efficiently. It provides a full set of utilities to create and destroy new Julia processes and add them to a "cluster" (a collection of Julia processes connected together), as well as functions to perform Remote Procedure Calls (RPC) between the processes within a cluster. See the `API` section for details.
 
 This package ships as part of the Julia stdlib.
 
@@ -64,5 +64,4 @@ For controlling multiple processes at once:
 
 Julia processes connected with `Distributed` are all assigned a cluster-unique `Int` identifier, starting from `1`. The first Julia process within a cluster is given ID `1`, while other processes added via `addprocs` get incrementing IDs (`2`, `3`, etc.). Functions and macros which communicate from one process to another usually take one or more identifiers to determine which process they target - for example, `remotecall_fetch(myid, 2)` calls `myid()` on process 2.
 
-!!! note
-    Only process 1 (often called the "head", "primary", or "master") may add or remove processes, and manages the rest of the cluster. Other processes (called "workers" or "worker processes") may still call functions on each other and send and receive data, but `addprocs`/`rmprocs` on worker processes will fail with an error.
+**Note:** Only process 1 (often called the "head", "primary", or "master") may add or remove processes, and manages the rest of the cluster. Other processes (called "workers" or "worker processes") may still call functions on each other and send and receive data, but `addprocs`/`rmprocs` on worker processes will fail with an error.

src/Distributed.jl

@@ -15,7 +15,7 @@ using Base: Process, Semaphore, JLOptions, buffer_writes, @async_unwrap,
             julia_cmd, AsyncGenerator, acquire, release, invokelatest,
             shell_escape_posixly, shell_escape_csh,
             shell_escape_wincmd, escape_microsoft_c_args,
-            uv_error, something, notnothing, isbuffered, mapany
+            uv_error, something, notnothing, isbuffered, mapany, SizeUnknown
 using Base.Threads: Event
 
 using Serialization, Sockets

src/clusterserialize.jl

@@ -167,10 +167,11 @@ function deserialize_global_from_main(s::ClusterSerializer, sym)
             return nothing
         end
     end
+    Core.eval(Main, Expr(:global, sym))
     if sym_isconst
         ccall(:jl_set_const, Cvoid, (Any, Any, Any), Main, sym, v)
     else
-        setglobal!(Main, sym, v)
+        invokelatest(setglobal!, Main, sym, v)
     end
     return nothing
 end

src/managers.jl

@@ -111,7 +111,9 @@ addprocs([
   version is used on all remote machines because serialization and code distribution might
   fail otherwise.
 
-* `exeflags`: additional flags passed to the worker processes.
+* `exeflags`: additional flags passed to the worker processes. It can either be a `Cmd`, a `String`
+  holding one flag, or a collection of strings, with one element per flag.
+  E.g. `\`--threads=auto project=.\``, `"--compile-trace=stderr"` or `["--threads=auto", "--compile=all"]`.
 
 * `topology`: Specifies how the workers connect to each other. Sending a message between
   unconnected workers results in an error.
@@ -770,12 +772,12 @@ function kill(manager::LocalManager, pid::Int, config::WorkerConfig; exit_timeou
 
         # Check to see if our child exited, and if not, send an actual kill signal
         if !process_exited(config.process)
-            @warn("Failed to gracefully kill worker $(pid), sending SIGTERM")
-            kill(config.process, Base.SIGTERM)
+            @warn("Failed to gracefully kill worker $(pid), sending SIGQUIT")
+            kill(config.process, Base.SIGQUIT)
 
             sleep(term_timeout)
             if !process_exited(config.process)
-                @warn("Worker $(pid) ignored SIGTERM, sending SIGKILL")
+                @warn("Worker $(pid) ignored SIGQUIT, sending SIGKILL")
                 kill(config.process, Base.SIGKILL)
             end
         end

src/workerpool.jl

@@ -8,6 +8,7 @@ An `AbstractWorkerPool` should implement:
   - [`push!`](@ref) - add a new worker to the overall pool (available + busy)
   - [`put!`](@ref) - put back a worker to the available pool
   - [`take!`](@ref) - take a worker from the available pool (to be used for remote function execution)
+  - [`wait`](@ref) - block until a worker is available
   - [`length`](@ref) - number of workers available in the overall pool
   - [`isready`](@ref) - return false if a `take!` on the pool would block, else true
 
@@ -120,6 +121,11 @@ function wp_local_take!(pool::AbstractWorkerPool; role= :default)
     return worker
 end
 
+function wp_local_wait(pool::AbstractWorkerPool)
+    wait(pool.channel)
+    return nothing
+end
+
 function remotecall_pool(rc_f, f, pool::AbstractWorkerPool, args...; role= :default, kwargs...)
     worker = take!(pool; role=role)
     try
@@ -133,7 +139,7 @@ end
 # NOTE: remotecall_fetch does it automatically, but this will be more efficient as
 # it avoids the overhead associated with a local remotecall.
 
-for (func, rt) = ((:length, Int), (:isready, Bool), (:workers, Vector{Int}), (:nworkers, Int), (:take!, Int))
+for (func, rt) = ((:length, Int), (:isready, Bool), (:workers, Vector{Int}), (:nworkers, Int), (:take!, Int), (:wait, Nothing))
     func_local = Symbol(string("wp_local_", func))
     @eval begin
         function ($func)(pool::WorkerPool; role= :default)

test/distributed_exec.jl

@@ -518,6 +518,8 @@ test_iteration(RemoteChannel(() -> Channel(10)), RemoteChannel(() -> Channel(10)
     return count
 end
 
+@everywhere test_iteration_collect(ch) = length(collect(ch))
+
 @everywhere function test_iteration_put(ch, total)
     for i in 1:total
         put!(ch, i)
@@ -528,10 +530,16 @@ end
 let ch = RemoteChannel(() -> Channel(1))
     @async test_iteration_put(ch, 10)
     @test 10 == @fetchfrom id_other test_iteration_take(ch)
+    ch = RemoteChannel(() -> Channel(1))
+    @async test_iteration_put(ch, 10)
+    @test 10 == @fetchfrom id_other test_iteration_collect(ch)    
     # now reverse
     ch = RemoteChannel(() -> Channel(1))
     @spawnat id_other test_iteration_put(ch, 10)
     @test 10 == test_iteration_take(ch)
+    ch = RemoteChannel(() -> Channel(1))
+    @spawnat id_other test_iteration_put(ch, 10)
+    @test 10 == test_iteration_collect(ch)
 end
 
 # make sure exceptions propagate when waiting on Tasks
@@ -751,6 +759,28 @@ wp = WorkerPool(workers())
 wp = WorkerPool(2:3)
 @test sort(unique(pmap(_->myid(), wp, 1:100))) == [2,3]
 
+# wait on worker pool
+wp = WorkerPool(2:2)
+w = take!(wp)
+
+# local call to _wait
+@test !isready(wp)
+t = @async wait(wp)
+@test !istaskdone(t)
+put!(wp, w)
+status = timedwait(() -> istaskdone(t), 10)
+@test status == :ok
+
+# remote call to _wait
+take!(wp)
+@test !isready(wp)
+f = @spawnat w wait(wp)
+@test !isready(f)
+put!(wp, w)
+status = timedwait(() -> isready(f), 10)
+@test status == :ok
+
+
 # CachingPool tests
 wp = CachingPool(workers())
 @test [1:100...] == pmap(x->x, wp, 1:100)
@@ -2037,7 +2067,7 @@ begin
 
     # Next, ensure we get a log message when a worker does not cleanly exit
     w = only(addprocs(1))
-    @test_logs (:warn, r"sending SIGTERM") begin
+    @test_logs (:warn, r"sending SIGQUIT") begin
         remote_do(w) do
             # Cause the 'exit()' message that `rmprocs()` sends to do nothing
             Core.eval(Base, :(exit() = nothing))