Add missing parallel=:threads implementations

These implementations are extremely basic, but they try to follow the
patterns in the other parts of the Parallel module. My real motivation
is to be able to move the Distributed implementations into an extension,
so that Graphs.jl does not depend on Distributed.

Author: topolarity

src/Parallel/distance.jl

@@ -1,20 +1,51 @@
 # used in shortest path calculations
 
 function eccentricity(
+    g::AbstractGraph,
+    vs=vertices(g),
+    distmx::AbstractMatrix{T}=weights(g);
+    parallel=:distributed,
+) where {T<:Number}
+    return if parallel === :threads
+        threaded_eccentricity(g, vs, distmx)
+    elseif parallel === :distributed
+        distr_eccentricity(g, vs, distmx)
+    else
+        error(
+            "Unsupported parallel argument '$(repr(parallel))' (supported: ':threads' or ':distributed')",
+        )
+    end
+end
+
+function distr_eccentricity(
     g::AbstractGraph, vs=vertices(g), distmx::AbstractMatrix{T}=weights(g)
 ) where {T<:Number}
     vlen = length(vs)
     eccs = SharedVector{T}(vlen)
     @sync @distributed for i in 1:vlen
-        eccs[i] = maximum(Graphs.dijkstra_shortest_paths(g, vs[i], distmx).dists)
+        d′ = Graphs.dijkstra_shortest_paths(g, vs[i], distmx)
+        eccs[i] = maximum(d′.dists)
     end
     d = sdata(eccs)
     maximum(d) == typemax(T) && @warn("Infinite path length detected")
     return d
 end
 
-function eccentricity(g::AbstractGraph, distmx::AbstractMatrix)
-    return eccentricity(g, vertices(g), distmx)
+function threaded_eccentricity(
+    g::AbstractGraph, vs=vertices(g), distmx::AbstractMatrix{T}=weights(g)
+) where {T<:Number}
+    vlen = length(vs)
+    eccs = Vector{T}(undef, vlen)
+    Base.Threads.@threads for i in 1:vlen
+        d = Graphs.dijkstra_shortest_paths(g, vs[i], distmx)
+        eccs[i] = maximum(d.dists)
+    end
+    maximum(eccs) == typemax(T) && @warn("Infinite path length detected")
+    return eccs
+end
+
+function eccentricity(g::AbstractGraph, distmx::AbstractMatrix; parallel=:distributed)
+    return eccentricity(g, vertices(g), distmx; parallel)
 end
 
 function diameter(g::AbstractGraph, distmx::AbstractMatrix=weights(g))

src/Parallel/shortestpaths/dijkstra.jl

@@ -17,6 +17,43 @@ an optional distance matrix `distmx`. Return a [`Parallel.MultipleDijkstraState`
 traversal information.
 """
 function dijkstra_shortest_paths(
+    g::AbstractGraph{U},
+    sources=vertices(g),
+    distmx::AbstractMatrix{T}=weights(g);
+    parallel=:distributed,
+) where {T<:Number} where {U}
+    return if parallel === :threads
+        threaded_dijkstra_shortest_paths(g, sources, distmx)
+    elseif parallel === :distributed
+        distr_dijkstra_shortest_paths(g, sources, distmx)
+    else
+        error(
+            "Unsupported parallel argument '$(repr(parallel))' (supported: ':threads' or ':distributed')",
+        )
+    end
+end
+
+function threaded_dijkstra_shortest_paths(
+    g::AbstractGraph{U}, sources=vertices(g), distmx::AbstractMatrix{T}=weights(g)
+) where {T<:Number} where {U}
+    n_v = nv(g)
+    r_v = length(sources)
+
+    # TODO: remove `Int` once julialang/#23029 / #23032 are resolved
+    dists = Matrix{T}(undef, Int(r_v), Int(n_v))
+    parents = Matrix{U}(undef, Int(r_v), Int(n_v))
+
+    Base.Threads.@threads for i in 1:r_v
+        state = Graphs.dijkstra_shortest_paths(g, sources[i], distmx)
+        dists[i, :] = state.dists
+        parents[i, :] = state.parents
+    end
+
+    result = MultipleDijkstraState(dists, parents)
+    return result
+end
+
+function distr_dijkstra_shortest_paths(
     g::AbstractGraph{U}, sources=vertices(g), distmx::AbstractMatrix{T}=weights(g)
 ) where {T<:Number} where {U}
     n_v = nv(g)

src/Parallel/traversals/greedy_color.jl

@@ -1,4 +1,29 @@
-function random_greedy_color(g::AbstractGraph{T}, reps::Integer) where {T<:Integer}
+function random_greedy_color(
+    g::AbstractGraph{T}, reps::Integer; parallel=:distributed
+) where {T<:Integer}
+    return if parallel === :threads
+        threaded_random_greedy_color(g, reps)
+    elseif parallel === :distributed
+        distr_random_greedy_color(g, reps)
+    else
+        error(
+            "Unsupported parallel argument '$(repr(parallel))' (supported: ':threads' or ':distributed')",
+        )
+    end
+end
+
+function threaded_random_greedy_color(g::AbstractGraph{T}, reps::Integer) where {T<:Integer}
+    local_best = Any[nothing for _ in 1:reps]
+    Base.Threads.@threads for i in 1:reps
+        seq = shuffle(vertices(g))
+        local_best[i] = Graphs.perm_greedy_color(g, seq)
+    end
+    best = reduce(Graphs.best_color, local_best)
+
+    return convert(Graphs.Coloring{T}, best)
+end
+
+function distr_random_greedy_color(g::AbstractGraph{T}, reps::Integer) where {T<:Integer}
     best = @distributed (Graphs.best_color) for i in 1:reps
         seq = shuffle(vertices(g))
         Graphs.perm_greedy_color(g, seq)
@@ -8,11 +33,11 @@ function random_greedy_color(g::AbstractGraph{T}, reps::Integer) where {T<:Integ
 end
 
 function greedy_color(
-    g::AbstractGraph{U}; sort_degree::Bool=false, reps::Integer=1
+    g::AbstractGraph{U}; sort_degree::Bool=false, reps::Integer=1, parallel=:distributed
 ) where {U<:Integer}
     return if sort_degree
         Graphs.degree_greedy_color(g)
     else
-        Parallel.random_greedy_color(g, reps)
+        Parallel.random_greedy_color(g, reps; parallel)
     end
 end

test/parallel/distance.jl

@@ -7,31 +7,43 @@
     distmx1 = [Inf 2.0 Inf; 2.0 Inf 4.2; Inf 4.2 Inf]
     distmx2 = [Inf 2.0 Inf; 3.2 Inf 4.2; 5.5 6.1 Inf]
 
-    for g in testgraphs(a1)
-        z = @inferred(Graphs.eccentricity(g, distmx1))
-        y = @inferred(Parallel.eccentricity(g, distmx1))
-        @test isapprox(y, z)
-        @test @inferred(Graphs.diameter(y)) ==
-            @inferred(Parallel.diameter(g, distmx1)) ==
-            6.2
-        @test @inferred(Graphs.periphery(y)) ==
-            @inferred(Parallel.periphery(g, distmx1)) ==
-            [1, 3]
-        @test @inferred(Graphs.radius(y)) == @inferred(Parallel.radius(g, distmx1)) == 4.2
-        @test @inferred(Graphs.center(y)) == @inferred(Parallel.center(g, distmx1)) == [2]
+    for parallel in [:threads, :distributed]
+        for g in testgraphs(a1)
+            z = @inferred(Graphs.eccentricity(g, distmx1))
+            y = @inferred(Parallel.eccentricity(g, distmx1; parallel))
+            @test isapprox(y, z)
+            @test @inferred(Graphs.diameter(y)) ==
+                @inferred(Parallel.diameter(g, distmx1)) ==
+                6.2
+            @test @inferred(Graphs.periphery(y)) ==
+                @inferred(Parallel.periphery(g, distmx1)) ==
+                [1, 3]
+            @test @inferred(Graphs.radius(y)) ==
+                @inferred(Parallel.radius(g, distmx1)) ==
+                4.2
+            @test @inferred(Graphs.center(y)) ==
+                @inferred(Parallel.center(g, distmx1)) ==
+                [2]
+        end
     end
 
-    for g in testdigraphs(a2)
-        z = @inferred(Graphs.eccentricity(g, distmx2))
-        y = @inferred(Parallel.eccentricity(g, distmx2))
-        @test isapprox(y, z)
-        @test @inferred(Graphs.diameter(y)) ==
-            @inferred(Parallel.diameter(g, distmx2)) ==
-            6.2
-        @test @inferred(Graphs.periphery(y)) ==
-            @inferred(Parallel.periphery(g, distmx2)) ==
-            [1]
-        @test @inferred(Graphs.radius(y)) == @inferred(Parallel.radius(g, distmx2)) == 4.2
-        @test @inferred(Graphs.center(y)) == @inferred(Parallel.center(g, distmx2)) == [2]
+    for parallel in [:threads, :distributed]
+        for g in testdigraphs(a2)
+            z = @inferred(Graphs.eccentricity(g, distmx2))
+            y = @inferred(Parallel.eccentricity(g, distmx2; parallel))
+            @test isapprox(y, z)
+            @test @inferred(Graphs.diameter(y)) ==
+                @inferred(Parallel.diameter(g, distmx2)) ==
+                6.2
+            @test @inferred(Graphs.periphery(y)) ==
+                @inferred(Parallel.periphery(g, distmx2)) ==
+                [1]
+            @test @inferred(Graphs.radius(y)) ==
+                @inferred(Parallel.radius(g, distmx2)) ==
+                4.2
+            @test @inferred(Graphs.center(y)) ==
+                @inferred(Parallel.center(g, distmx2)) ==
+                [2]
+        end
     end
 end

test/parallel/shortestpaths/dijkstra.jl

@@ -6,42 +6,44 @@
     g3 = path_graph(5)
     d = [0 1 2 3 4; 1 0 1 0 1; 2 1 0 11 12; 3 0 11 0 5; 4 1 19 5 0]
 
-    for g in testgraphs(g3)
-        z = floyd_warshall_shortest_paths(g, d)
-        zp = @inferred(Parallel.dijkstra_shortest_paths(g, collect(1:5), d))
-        @test all(isapprox(z.dists, zp.dists))
+    for parallel in [:threads, :distributed]
+        for g in testgraphs(g3)
+            z = floyd_warshall_shortest_paths(g, d)
+            zp = @inferred(Parallel.dijkstra_shortest_paths(g, collect(1:5), d; parallel))
+            @test all(isapprox(z.dists, zp.dists))
 
-        for i in 1:5
-            state = Graphs.dijkstra_shortest_paths(g, i; allpaths=true)
-            for j in 1:5
-                if zp.parents[i, j] != 0
-                    @test zp.parents[i, j] in state.predecessors[j]
+            for i in 1:5
+                state = Graphs.dijkstra_shortest_paths(g, i; allpaths=true)
+                for j in 1:5
+                    if zp.parents[i, j] != 0
+                        @test zp.parents[i, j] in state.predecessors[j]
+                    end
                 end
             end
-        end
 
-        z = floyd_warshall_shortest_paths(g)
-        zp = @inferred(Parallel.dijkstra_shortest_paths(g))
-        @test all(isapprox(z.dists, zp.dists))
+            z = floyd_warshall_shortest_paths(g)
+            zp = @inferred(Parallel.dijkstra_shortest_paths(g; parallel))
+            @test all(isapprox(z.dists, zp.dists))
 
-        for i in 1:5
-            state = Graphs.dijkstra_shortest_paths(g, i; allpaths=true)
-            for j in 1:5
-                if zp.parents[i, j] != 0
-                    @test zp.parents[i, j] in state.predecessors[j]
+            for i in 1:5
+                state = Graphs.dijkstra_shortest_paths(g, i; allpaths=true)
+                for j in 1:5
+                    if zp.parents[i, j] != 0
+                        @test zp.parents[i, j] in state.predecessors[j]
+                    end
                 end
             end
-        end
 
-        z = floyd_warshall_shortest_paths(g)
-        zp = @inferred(Parallel.dijkstra_shortest_paths(g, [1, 2]))
-        @test all(isapprox(z.dists[1:2, :], zp.dists))
+            z = floyd_warshall_shortest_paths(g)
+            zp = @inferred(Parallel.dijkstra_shortest_paths(g, [1, 2]; parallel))
+            @test all(isapprox(z.dists[1:2, :], zp.dists))
 
-        for i in 1:2
-            state = Graphs.dijkstra_shortest_paths(g, i; allpaths=true)
-            for j in 1:5
-                if zp.parents[i, j] != 0
-                    @test zp.parents[i, j] in state.predecessors[j]
+            for i in 1:2
+                state = Graphs.dijkstra_shortest_paths(g, i; allpaths=true)
+                for j in 1:5
+                    if zp.parents[i, j] != 0
+                        @test zp.parents[i, j] in state.predecessors[j]
+                    end
                 end
             end
         end
@@ -51,42 +53,44 @@
     g3 = path_digraph(5)
     d = float([0 1 2 3 4; 5 0 6 7 8; 9 10 0 11 12; 13 14 15 0 16; 17 18 19 20 0])
 
-    for g in testdigraphs(g3)
-        z = floyd_warshall_shortest_paths(g, d)
-        zp = @inferred(Parallel.dijkstra_shortest_paths(g, collect(1:5), d))
-        @test all(isapprox(z.dists, zp.dists))
+    for parallel in [:threads, :distributed]
+        for g in testdigraphs(g3)
+            z = floyd_warshall_shortest_paths(g, d)
+            zp = @inferred(Parallel.dijkstra_shortest_paths(g, collect(1:5), d; parallel))
+            @test all(isapprox(z.dists, zp.dists))
 
-        for i in 1:5
-            state = Graphs.dijkstra_shortest_paths(g, i; allpaths=true)
-            for j in 1:5
-                if z.parents[i, j] != 0
-                    @test zp.parents[i, j] in state.predecessors[j]
+            for i in 1:5
+                state = Graphs.dijkstra_shortest_paths(g, i; allpaths=true)
+                for j in 1:5
+                    if z.parents[i, j] != 0
+                        @test zp.parents[i, j] in state.predecessors[j]
+                    end
                 end
             end
-        end
 
-        z = floyd_warshall_shortest_paths(g)
-        zp = @inferred(Parallel.dijkstra_shortest_paths(g))
-        @test all(isapprox(z.dists, zp.dists))
+            z = floyd_warshall_shortest_paths(g)
+            zp = @inferred(Parallel.dijkstra_shortest_paths(g; parallel))
+            @test all(isapprox(z.dists, zp.dists))
 
-        for i in 1:5
-            state = Graphs.dijkstra_shortest_paths(g, i; allpaths=true)
-            for j in 1:5
-                if zp.parents[i, j] != 0
-                    @test zp.parents[i, j] in state.predecessors[j]
+            for i in 1:5
+                state = Graphs.dijkstra_shortest_paths(g, i; allpaths=true)
+                for j in 1:5
+                    if zp.parents[i, j] != 0
+                        @test zp.parents[i, j] in state.predecessors[j]
+                    end
                 end
             end
-        end
 
-        z = floyd_warshall_shortest_paths(g)
-        zp = @inferred(Parallel.dijkstra_shortest_paths(g, [1, 2]))
-        @test all(isapprox(z.dists[1:2, :], zp.dists))
+            z = floyd_warshall_shortest_paths(g)
+            zp = @inferred(Parallel.dijkstra_shortest_paths(g, [1, 2]; parallel))
+            @test all(isapprox(z.dists[1:2, :], zp.dists))
 
-        for i in 1:2
-            state = Graphs.dijkstra_shortest_paths(g, i; allpaths=true)
-            for j in 1:5
-                if zp.parents[i, j] != 0
-                    @test zp.parents[i, j] in state.predecessors[j]
+            for i in 1:2
+                state = Graphs.dijkstra_shortest_paths(g, i; allpaths=true)
+                for j in 1:5
+                    if zp.parents[i, j] != 0
+                        @test zp.parents[i, j] in state.predecessors[j]
+                    end
                 end
             end
         end