Merge branch 'master' into ib/worker_output_customization

Author: DilumAluthge

.github/workflows/ci.yml

@@ -15,6 +15,9 @@ concurrency:
   group: ${{ github.workflow }}-${{ github.ref }}-${{ github.ref != 'refs/heads/main' || startsWith(github.ref, 'refs/heads/release-') || github.run_number }}
   cancel-in-progress: ${{ startsWith(github.ref, 'refs/pull/') }}
 
+permissions:
+  contents: read
+
 jobs:
   test:
     name: julia -t${{ matrix.threads}} - ${{ matrix.version }} - ${{ matrix.os }} - ${{ matrix.arch }} - ${{ github.event_name }}
@@ -40,6 +43,8 @@ jobs:
             arch: x86
     steps:
       - uses: actions/checkout@v6
+        with:
+            persist-credentials: false
       - uses: julia-actions/setup-julia@v2
         with:
           version: ${{ matrix.version }}
@@ -59,6 +64,8 @@ jobs:
     runs-on: ubuntu-latest
     steps:
       - uses: actions/checkout@v6
+        with:
+            persist-credentials: false
       - uses: julia-actions/setup-julia@latest
         with:
           # version: '1.6'
@@ -70,3 +77,18 @@ jobs:
           julia --project=docs --color=yes docs/make.jl pdf
         env:
           DOCUMENTER_KEY: ${{ secrets.DOCUMENTER_KEY }}
+
+  jet:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v6
+        with:
+            persist-credentials: false
+      - uses: julia-actions/setup-julia@latest
+        with:
+          version: '1'
+          # version: 'nightly'
+      - uses: julia-actions/cache@v2
+      - run: julia --color=yes --project=ci/jet -e 'import Pkg; Pkg.instantiate()'
+      - name: Run the JET tests
+        run: julia --color=yes -t4 --project=ci/jet ci/jet/check.jl

ci/jet/Project.toml

@@ -0,0 +1,8 @@
+[deps]
+Distributed = "8ba89e20-285c-5b6f-9357-94700520ee1b"
+JET = "c3a54625-cd67-489e-a8e7-0a5a0ff4e31b"
+Serialization = "9e88b42a-f829-5b0c-bbe9-9e923198166b"
+Test = "8dfed614-e22c-5e08-85e1-65c5234f0b40"
+
+[sources]
+Distributed = {path = "../.."}

ci/jet/check.jl

@@ -0,0 +1,20 @@
+using Distributed: Distributed
+
+using JET: JET
+using Serialization: Serialization
+using Test: Test, @testset
+
+# We don't want to fail PkgEval because of a JET failure
+# Therefore, we don't put the JET tests in the regular Distributed test suite
+# Instead, we put it in a separate CI job, which runs on the Distributed repo
+
+@testset "JET" begin
+    ignored_modules = (
+        # We will ignore Base:
+        Base,
+
+        # We'll ignore the Serialization stdlib:
+        Serialization,
+    )
+    JET.test_package(Distributed; ignored_modules)
+end

src/cluster.jl

@@ -664,7 +664,7 @@ function create_worker(manager::ClusterManager, wconfig::WorkerConfig)
         end
     end
 
-    w = Worker(w_stub.id, r_s, w_s, manager; config=wconfig)
+    w = Worker(w_stub.id, r_s, w_s, manager; config=wconfig)::Worker
     # install a finalizer to perform cleanup if necessary
     finalizer(w) do w
         if myid() == 1

src/macros.jl

@@ -333,6 +333,7 @@ completion. To wait for completion, prefix the call with [`@sync`](@ref), like :
 macro distributed(args...)
     na = length(args)
     if na==1
+        reducer = identity
         loop = args[1]
     elseif na==2
         reducer = args[1]

src/managers.jl

@@ -526,7 +526,7 @@ end
 
 function manage(manager::LocalManager, id::Integer, config::WorkerConfig, op::Symbol)
     if op === :interrupt
-        kill(config.process, 2)
+        kill(config.process::Process, 2)
     end
 end
 
@@ -749,21 +749,22 @@ function kill(manager::LocalManager, pid::Int, config::WorkerConfig; profile_wai
         sleep(exit_timeout)
 
         # Check to see if our child exited, and if not, send an actual kill signal
-        if !process_exited(config.process)
+        process = config.process::Process
+        if !process_exited(process)
             @warn "Failed to gracefully kill worker $(pid)"
             profile_sig = Sys.iswindows() ? nothing : Sys.isbsd() ? ("SIGINFO", 29) : ("SIGUSR1" , 10)
             if profile_sig !== nothing
                 @warn("Sending profile $(profile_sig[1]) to worker $(pid)")
-                kill(config.process, profile_sig[2])
+                kill(process, profile_sig[2])
                 sleep(profile_wait)
             end
             @warn("Sending SIGQUIT to worker $(pid)")
-            kill(config.process, Base.SIGQUIT)
+            kill(process, Base.SIGQUIT)
 
             sleep(term_timeout)
-            if !process_exited(config.process)
+            if !process_exited(process)
                 @warn("Worker $(pid) ignored SIGQUIT, sending SIGKILL")
-                kill(config.process, Base.SIGKILL)
+                kill(process, Base.SIGKILL)
             end
         end
     end

src/process_messages.jl

@@ -75,7 +75,7 @@ function run_work_thunk(thunk::Function, print_error::Bool)
     end
     return result
 end
-function run_work_thunk(rv::RemoteValue, thunk)
+function run_work_thunk_remotevalue(rv::RemoteValue, thunk)
     put!(rv, run_work_thunk(thunk, false))
     nothing
 end
@@ -85,7 +85,7 @@ function schedule_call(rid, thunk)
         rv = RemoteValue(def_rv_channel())
         (PGRP::ProcessGroup).refs[rid] = rv
         push!(rv.clientset, rid.whence)
-        errormonitor(@async run_work_thunk(rv, thunk))
+        errormonitor(@async run_work_thunk_remotevalue(rv, thunk))
         return rv
     end
 end

src/workerpool.jl

@@ -288,7 +288,7 @@ julia> default_worker_pool()
 WorkerPool(Channel{Int64}(sz_max:9223372036854775807,sz_curr:3), Set([4, 2, 3]), RemoteChannel{Channel{Any}}(1, 1, 4))
 ```
 """
-function default_worker_pool()
+function default_worker_pool()::AbstractWorkerPool
     # On workers retrieve the default worker pool from the master when accessed
     # for the first time
     if _default_worker_pool[] === nothing
@@ -298,7 +298,7 @@ function default_worker_pool()
             _default_worker_pool[] = remotecall_fetch(()->default_worker_pool(), 1)
         end
     end
-    return _default_worker_pool[]
+    return _default_worker_pool[]::AbstractWorkerPool
 end
 
 """