Merge branch 'master' into dw/aqua

Author: andreasnoack

.github/workflows/CI.yml

@@ -31,15 +31,8 @@ jobs:
           - ubuntu-latest
           - windows-latest
           - macOS-latest
-        exclude:
-          # For Julia 1.6 no aarch64 binary exists
-          - version: 'min'
-            os: macOS-latest
-        include:
-          - version: 'min'
-            os: macOS-13 # uses x64
     steps:
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v6
       - uses: julia-actions/setup-julia@v2
         with:
           version: ${{ matrix.version }}
@@ -56,7 +49,7 @@ jobs:
     name: Documentation
     runs-on: ubuntu-latest
     steps:
-      - uses: actions/checkout@v5
+      - uses: actions/checkout@v6
       - uses: julia-actions/setup-julia@v2
         with:
           version: '1'

.github/workflows/CompatHelper.yml

@@ -24,7 +24,7 @@ jobs:
       - name: "Run CompatHelper"
         run: |
           import CompatHelper
-          CompatHelper.main(; dirs = ["", "docs"])
+          CompatHelper.main(; subdirs = ["", "docs"])
         shell: julia --color=yes {0}
         env:
           GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

Project.toml

@@ -8,12 +8,19 @@ LinearAlgebra = "37e2e46d-f89d-539d-b4ee-838fcccc9c8e"
 Primes = "27ebfcd6-29c5-5fa9-bf4b-fb8fc14df3ae"
 Random = "9a3f8284-a2c9-5f02-9a11-845980a1fd5c"
 Serialization = "9e88b42a-f829-5b0c-bbe9-9e923198166b"
+
+[weakdeps]
 SparseArrays = "2f01184e-e22b-5df5-ae63-d93ebab69eaf"
 Statistics = "10745b16-79ce-11e8-11f9-7d13ad32a3b2"
 
+[extensions]
+SparseArraysExt = "SparseArrays"
+StatisticsExt = "Statistics"
+
 [compat]
 Aqua = "0.8.12"
 Distributed = "<0.0.1, 1"
+ExplicitImports = "1.13.2"
 LinearAlgebra = "<0.0.1, 1"
 Primes = "0.4, 0.5"
 Random = "<0.0.1, 1"
@@ -22,12 +29,15 @@ SparseArrays = "<0.0.1, 1"
 SpecialFunctions = "0.8, 1, 2"
 Statistics = "<0.0.1, 1"
 Test = "<0.0.1, 1"
-julia = "1"
+julia = "1.10"
 
 [extras]
 Aqua = "4c88cf16-eb10-579e-8560-4a9242c79595"
+ExplicitImports = "7d51a73a-1435-4ff3-83d9-f097790105c7"
+SparseArrays = "2f01184e-e22b-5df5-ae63-d93ebab69eaf"
 SpecialFunctions = "276daf66-3868-5448-9aa4-cd146d93841b"
+Statistics = "10745b16-79ce-11e8-11f9-7d13ad32a3b2"
 Test = "8dfed614-e22c-5e08-85e1-65c5234f0b40"
 
 [targets]
-test = ["Aqua", "Test", "SpecialFunctions"]
+test = ["Aqua", "ExplicitImports", "SparseArrays", "SpecialFunctions", "Statistics", "Test"]

README.md

@@ -33,7 +33,11 @@ julia> import Pkg; Pkg.add("DistributedArrays")
 
 ## Project Status
 
-The package is tested against Julia `0.7`, `1.0` and the nightly builds of the Julia `master` branch on Linux, and macOS.
+The package is tested against
+Julia 1.10.0 (oldest supported Julia version),
+the Julia LTS version,
+the latest stable release of Julia,
+and the pre-release version of Julia.
 
 ## Questions and Contributions
 

ext/SparseArraysExt.jl

@@ -0,0 +1,14 @@
+module SparseArraysExt
+
+using DistributedArrays: DArray, localpart
+using DistributedArrays.Distributed: remotecall_fetch
+using SparseArrays: SparseArrays, nnz
+
+function SparseArrays.nnz(A::DArray)
+    B = asyncmap(A.pids) do p
+        remotecall_fetch(nnz∘localpart, p, A)
+    end
+    return reduce(+, B)
+end
+
+end

ext/StatisticsExt.jl

@@ -0,0 +1,8 @@
+module StatisticsExt
+
+using DistributedArrays: DArray
+using Statistics: Statistics
+
+Statistics._mean(f, A::DArray, region) = sum(f, A, dims = region) ./ prod((size(A, i) for i in region))
+
+end

src/DistributedArrays.jl

@@ -1,18 +1,14 @@
-__precompile__()
-
 module DistributedArrays
 
-using Distributed
-using Serialization
-using LinearAlgebra
-using Statistics
+using Base: Callable
+using Base.Broadcast: BroadcastStyle, Broadcasted
 
-import Base: +, -, *, div, mod, rem, &, |, xor
-import Base.Callable
-import LinearAlgebra: axpy!, dot, norm, mul!
+using Distributed: Distributed, RemoteChannel, Future, myid, nworkers, procs, remotecall, remotecall_fetch, remotecall_wait, worker_id_from_socket, workers
+using LinearAlgebra: LinearAlgebra, Adjoint, Diagonal, I, Transpose, adjoint, adjoint!, axpy!, dot, lmul!, mul!, norm, rmul!, transpose, transpose!
+using Random: Random, rand!
+using Serialization: Serialization, AbstractSerializer, deserialize, serialize
 
-import Primes
-import Primes: factor
+using Primes: factor
 
 import SparseArrays
 
@@ -24,7 +20,7 @@ export dzeros, dones, dfill, drand, drandn, distribute, localpart, localindices,
 export ddata, gather
 
 # immediate release of localparts
-export close, d_closeall
+export d_closeall
 
 include("darray.jl")
 include("core.jl")

src/broadcast.jl

@@ -2,9 +2,6 @@
 # Distributed broadcast implementation
 ##
 
-using Base.Broadcast
-import Base.Broadcast: BroadcastStyle, Broadcasted
-
 # We define a custom ArrayStyle here since we need to keep track of
 # the fact that it is Distributed and what kind of underlying broadcast behaviour
 # we will encounter.
@@ -13,11 +10,11 @@ DArrayStyle(::S) where {S} = DArrayStyle{S}()
 DArrayStyle(::S, ::Val{N}) where {S,N} = DArrayStyle(S(Val(N)))
 DArrayStyle(::Val{N}) where N = DArrayStyle{Broadcast.DefaultArrayStyle{N}}()
 
-BroadcastStyle(::Type{<:DArray{<:Any, N, A}}) where {N, A} = DArrayStyle(BroadcastStyle(A), Val(N))
+Broadcast.BroadcastStyle(::Type{<:DArray{<:Any, N, A}}) where {N, A} = DArrayStyle(BroadcastStyle(A), Val(N))
 
 # promotion rules
 # TODO: test this
-function BroadcastStyle(::DArrayStyle{AStyle}, ::DArrayStyle{BStyle}) where {AStyle, BStyle}
+function Broadcast.BroadcastStyle(::DArrayStyle{AStyle}, ::DArrayStyle{BStyle}) where {AStyle, BStyle}
     DArrayStyle(BroadcastStyle(AStyle, BStyle))
 end
 
@@ -70,18 +67,18 @@ end
     # This will turn local AbstractArrays into DArrays
     dbc = bcdistribute(bc)
 
-    asyncmap(procs(dest)) do p
-        remotecall_fetch(p) do
+    @sync for p in procs(dest)
+        @async remotecall_wait(p) do
             # get the indices for the localpart
             lpidx = localpartindex(dest)
             @assert lpidx != 0
             # create a local version of the broadcast, by constructing views
             # Note: creates copies of the argument
             lbc = bclocal(dbc, dest.indices[lpidx])
-            Base.copyto!(localpart(dest), lbc)
-            return nothing
+            copyto!(localpart(dest), lbc)
         end
     end
+
     return dest
 end
 

src/core.jl

@@ -1,58 +1,109 @@
-const registry=Dict{Tuple, Any}()
-const refs=Set()  # Collection of darray identities created on this node
+# Thread-safe registry of DArray references
+struct DArrayRegistry
+    data::Dict{Tuple{Int,Int}, Any}
+    lock::ReentrantLock
+    DArrayRegistry() = new(Dict{Tuple{Int,Int}, Any}(), ReentrantLock())
+end
+const REGISTRY = DArrayRegistry()
+
+function Base.get(r::DArrayRegistry, id::Tuple{Int,Int}, default)
+    @lock r.lock begin
+        return get(r.data, id, default)
+    end
+end
+function Base.getindex(r::DArrayRegistry, id::Tuple{Int,Int})
+    @lock r.lock begin
+        return r.data[id]
+    end
+end
+function Base.setindex!(r::DArrayRegistry, val, id::Tuple{Int,Int})
+    @lock r.lock begin
+        r.data[id] = val
+    end
+    return r
+end
+function Base.delete!(r::DArrayRegistry, id::Tuple{Int,Int})
+    @lock r.lock delete!(r.data, id)
+    return r
+end
+
+# Thread-safe set of IDs of DArrays created on this node
+struct DArrayRefs
+    data::Set{Tuple{Int,Int}}
+    lock::ReentrantLock
+    DArrayRefs() = new(Set{Tuple{Int,Int}}(), ReentrantLock())
+end
+const REFS = DArrayRefs()
 
-let DID::Int = 1
-    global next_did
-    next_did() = (id = DID; DID += 1; (myid(), id))
+function Base.push!(r::DArrayRefs, id::Tuple{Int,Int})
+    # Ensure id refers to a DArray created on this node
+    if first(id) != myid()
+        throw(
+            ArgumentError(
+                lazy"`DArray` is not created on the current worker: Only `DArray`s created on worker $(myid()) can be stored in this set but the `DArray` was created on worker $(first(id))."))
+    end
+    @lock r.lock begin
+        return push!(r.data, id)
+    end
+end
+function Base.delete!(r::DArrayRefs, id::Tuple{Int,Int})
+    @lock r.lock delete!(r.data, id)
+    return r
 end
 
+# Global counter to generate a unique ID for each DArray
+const DID = Threads.Atomic{Int}(1)
+
 """
     next_did()
 
-Produces an incrementing ID that will be used for DArrays.
-"""
-next_did
+Increment a global counter and return a tuple of the current worker ID and the incremented
+value of the counter.
 
-release_localpart(id::Tuple) = (delete!(registry, id); nothing)
-release_localpart(d) = release_localpart(d.id)
+This tuple is used as a unique ID for a new `DArray`.
+"""
+next_did() = (myid(), Threads.atomic_add!(DID, 1))
 
-function close_by_id(id, pids)
-#   @async println("Finalizer for : ", id)
-    global refs
+release_localpart(id::Tuple{Int,Int}) = (delete!(REGISTRY, id); nothing)
+function release_allparts(id::Tuple{Int,Int}, pids::Array{Int})
     @sync begin
+        released_myid = false
         for p in pids
-            @async remotecall_fetch(release_localpart, p, id)
+            if p == myid()
+                @async release_localpart(id)
+                released_myid = true
+            else
+                @async remotecall_fetch(release_localpart, p, id)
+            end
         end
-        if !(myid() in pids)
-            release_localpart(id)
+        if !released_myid
+            @async release_localpart(id)
         end
     end
-    delete!(refs, id)
-    nothing
+    return nothing
 end
 
-function Base.close(d::DArray)
-#    @async println("close : ", d.id, ", object_id : ", object_id(d), ", myid : ", myid() )
-    if (myid() == d.id[1]) && d.release
-        @async close_by_id(d.id, d.pids)
-        d.release = false
-    end
+function close_by_id(id::Tuple{Int,Int}, pids::Array{Int})
+    release_allparts(id, pids)
+    delete!(REFS, id)
     nothing
 end
 
 function d_closeall()
-    crefs = copy(refs)
-    for id in crefs
-        if id[1] ==  myid() # sanity check
-            if haskey(registry, id)
-                d = d_from_weakref_or_d(id)
-                (d === nothing) || close(d)
+    @lock REFS.lock begin
+        while !isempty(REFS.data)
+            id = pop!(REFS.data)
+            d = d_from_weakref_or_d(id)
+            if d isa DArray
+                finalize(d)
             end
-            yield()
         end
     end
+    return nothing
 end
 
+Base.close(d::DArray) = finalize(d)
+
 """
     procs(d::DArray)
 
@@ -67,4 +118,3 @@ Distributed.procs(d::SubDArray) = procs(parent(d))
 The identity when input is not distributed
 """
 localpart(A) = A
-