Skip to content

Add missing parallel=:threads implementations #429

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Add missing parallel=:threads implementations #429

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 34 additions & 3 deletions src/Parallel/distance.jl
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,20 +1,51 @@
# used in shortest path calculations

function eccentricity(
g::AbstractGraph,
vs=vertices(g),
distmx::AbstractMatrix{T}=weights(g);
parallel=:distributed,
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggested change
parallel=:distributed,
parallel::Symbol=:distributed,

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I know we haven't done that in other places - but perhaps to make this code more future proof would it make sense to use a specific structure to specify the parallel policy? Then we could later add things such as the number of threads/cores used.

) where {T<:Number}
return if parallel === :threads
threaded_eccentricity(g, vs, distmx)
elseif parallel === :distributed
distr_eccentricity(g, vs, distmx)
else
error(

Check warning on line 14 in src/Parallel/distance.jl

View check run for this annotation

Codecov / codecov/patch

src/Parallel/distance.jl#L14 

Added line #L14 was not covered by tests
"Unsupported parallel argument '$(repr(parallel))' (supported: ':threads' or ':distributed')",
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should also add a test for the case that someone uses the wrong argument.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We repeat the same error multiple times - Perhaps we can throw a custom error type instead?

)
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)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I am a bit unsure about the usage of d′ - it is hardly distinguishable from d' - in that case ' is the adjoint operator.

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

Check warning on line 42 in src/Parallel/distance.jl

View check run for this annotation

Codecov / codecov/patch

src/Parallel/distance.jl#L42 

Added line #L42 was not covered by tests
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))
Expand Down
37 changes: 37 additions & 0 deletions src/Parallel/shortestpaths/dijkstra.jl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,43 @@
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(

Check warning on line 30 in src/Parallel/shortestpaths/dijkstra.jl

View check run for this annotation

Codecov / codecov/patch

src/Parallel/shortestpaths/dijkstra.jl#L30 

Added line #L30 was not covered by tests
"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

Check warning on line 50 in src/Parallel/shortestpaths/dijkstra.jl

View check run for this annotation

Codecov / codecov/patch

src/Parallel/shortestpaths/dijkstra.jl#L50 

Added line #L50 was not covered by tests

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)
Expand Down
31 changes: 28 additions & 3 deletions src/Parallel/traversals/greedy_color.jl
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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(

Check warning on line 9 in src/Parallel/traversals/greedy_color.jl

View check run for this annotation

Codecov / codecov/patch

src/Parallel/traversals/greedy_color.jl#L9 

Added line #L9 was not covered by tests
"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]
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is there a reason to use Any[nothing for _ in 1:reps] instead of

Vector{Graphs.Coloring{T}}(undef, reps)

or

Vector{Union{Nothing, Graphs.Coloring{T}}}(nothing, reps)

?

Base.Threads.@threads for i in 1:reps
seq = shuffle(vertices(g))
local_best[i] = Graphs.perm_greedy_color(g, seq)
end

Check warning on line 20 in src/Parallel/traversals/greedy_color.jl

View check run for this annotation

Codecov / codecov/patch

src/Parallel/traversals/greedy_color.jl#L20 

Added line #L20 was not covered by tests
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)
Expand All @@ -8,11 +33,11 @@
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
60 changes: 36 additions & 24 deletions test/parallel/distance.jl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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]
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It might make sense to write the test as

@testset "Parallel.Distance" for parallel in [:threads, :distributed]
[...]
end

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
112 changes: 58 additions & 54 deletions test/parallel/shortestpaths/dijkstra.jl
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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
Expand Down
Loading
Loading