Skip to content

Commit b8e2303

Browse files
jishnubjishnub
committed
finalize channels, added tests
1 parent 03f07c4 commit b8e2303

File tree

3 files changed

+98
-10
lines changed

3 files changed

+98
-10
lines changed

‎Project.toml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,7 +1,7 @@
11
name = "ParallelUtilities"
22
uuid = "fad6cfc8-4f83-11e9-06cc-151124046ad0"
33
authors = ["Jishnu Bhattacharya <jishnuonline@gmail.com>"]
4-
version = "0.2.0"
4+
version = "0.2.1"
55

66
[deps]
77
Distributed = "8ba89e20-285c-5b6f-9357-94700520ee1b"

‎src/ParallelUtilities.jl‎

Lines changed: 55 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -577,6 +577,8 @@ function pmapreduce_commutative(fmap::Function,freduce::Function,iterators::Tupl
577577

578578
eachnode_reducechannel = node_channels[node].out
579579
eachnode_errorchannel = node_channels[node].err
580+
581+
p_parent = eachnode_reducechannel.where
580582

581583
np_node = nprocs_node_dict[node]
582584

@@ -591,14 +593,33 @@ function pmapreduce_commutative(fmap::Function,freduce::Function,iterators::Tupl
591593
put!(eachnode_errorchannel,true)
592594
rethrow()
593595
finally
594-
if p ∉ procid_rank1_on_node
596+
# Don't need references to the intermediate reduction channels on other nodes
597+
for (node_i,channels_i) in node_channels
598+
if node_i == node
599+
continue
600+
end
601+
602+
finalize(channels_i.out)
603+
finalize(channels_i.err)
604+
end
605+
if p != p_parent
595606
finalize(eachnode_errorchannel)
596607
finalize(eachnode_reducechannel)
608+
finalize(finalnode_errorchannel)
609+
finalize(finalnode_reducechannel)
610+
end
611+
if p != p_final
612+
if p != result_channel.where
613+
finalize(result_channel)
614+
end
615+
if p != error_channel.where
616+
finalize(error_channel)
617+
end
597618
end
598619
end
599620
end
600621

601-
@async if p in procid_rank1_on_node
622+
@async if p == p_parent
602623
@spawnat p begin
603624
try
604625
anyerror = any(take!(eachnode_errorchannel) for i=1:np_node)
@@ -613,7 +634,9 @@ function pmapreduce_commutative(fmap::Function,freduce::Function,iterators::Tupl
613634
put!(finalnode_errorchannel,true)
614635
rethrow()
615636
finally
637+
finalize(eachnode_errorchannel)
616638
finalize(eachnode_reducechannel)
639+
617640
if p != p_final
618641
finalize(finalnode_errorchannel)
619642
finalize(finalnode_reducechannel)
@@ -700,27 +723,29 @@ function pmapreduce(fmap::Function,freduce::Function,iterable::Tuple,args...;kwa
700723
nprocs_node_dict = nprocs_node(procs_used)
701724
node_channels = Dict(
702725
node=>(
703-
out = RemoteChannel(()->Channel{Any}(nprocs_node_dict[node]),procid_node),
726+
out = RemoteChannel(()->Channel{pval}(nprocs_node_dict[node]),procid_node),
704727
err = RemoteChannel(()->Channel{Bool}(nprocs_node_dict[node]),procid_node),
705728
)
706729
for (node,procid_node) in zip(nodes,procid_rank1_on_node))
707730

708731
# Worker at which the final reduction takes place
709732
p_final = first(procid_rank1_on_node)
710733

711-
finalnode_reducechannel = RemoteChannel(()->Channel{pval}(length(procid_rank1_on_node)),p_final)
712-
finalnode_errorchannel = RemoteChannel(()->Channel{Bool}(length(procid_rank1_on_node)),p_final)
734+
finalnode_reducechannel = RemoteChannel(()->Channel{pval}(Nnodes_reduction),p_final)
735+
finalnode_errorchannel = RemoteChannel(()->Channel{Bool}(Nnodes_reduction),p_final)
713736

714737
result_channel = RemoteChannel(()->Channel{Any}(1))
715738
error_channel = RemoteChannel(()->Channel{Bool}(1))
716739

717-
# Run the function on each processor and compute the sum at each node
740+
# Run the function on each processor and compute the reduction at each node
718741
@sync for (rank,(p,node)) in enumerate(zip(procs_used,hostnames))
719742
@async begin
720743

721744
eachnode_reducechannel = node_channels[node].out
722745
eachnode_errorchannel = node_channels[node].err
723746

747+
p_parent = eachnode_reducechannel.where
748+
724749
np_node = nprocs_node_dict[node]
725750

726751
iterable_on_proc = evenlyscatterproduct(iterable,num_workers,rank)
@@ -733,14 +758,33 @@ function pmapreduce(fmap::Function,freduce::Function,iterable::Tuple,args...;kwa
733758
put!(eachnode_errorchannel,true)
734759
rethrow()
735760
finally
736-
if p ∉ procid_rank1_on_node
761+
# Don't need references to the intermediate reduction channels on other nodes
762+
for (node_i,channels_i) in node_channels
763+
if node_i == node
764+
continue
765+
end
766+
767+
finalize(channels_i.out)
768+
finalize(channels_i.err)
769+
end
770+
if p != p_parent
737771
finalize(eachnode_errorchannel)
738772
finalize(eachnode_reducechannel)
773+
finalize(finalnode_errorchannel)
774+
finalize(finalnode_reducechannel)
739775
end
740-
end
776+
if p != p_final
777+
if p != result_channel.where
778+
finalize(result_channel)
779+
end
780+
if p != error_channel.where
781+
finalize(error_channel)
782+
end
783+
end
784+
end
741785
end
742786

743-
@async if p in procid_rank1_on_node
787+
@async if p == p_parent
744788
@spawnat p begin
745789
try
746790
anyerror = any(take!(eachnode_errorchannel) for i=1:np_node)
@@ -759,6 +803,7 @@ function pmapreduce(fmap::Function,freduce::Function,iterable::Tuple,args...;kwa
759803
finally
760804
finalize(eachnode_errorchannel)
761805
finalize(eachnode_reducechannel)
806+
762807
if p != p_final
763808
finalize(finalnode_errorchannel)
764809
finalize(finalnode_reducechannel)
@@ -786,6 +831,7 @@ function pmapreduce(fmap::Function,freduce::Function,iterable::Tuple,args...;kwa
786831
finally
787832
finalize(finalnode_errorchannel)
788833
finalize(finalnode_reducechannel)
834+
789835
if p != result_channel.where
790836
finalize(result_channel)
791837
end

‎test/runtests.jl‎

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -483,6 +483,14 @@ end
483483
iters = (1:100,1:2)
484484
@test pmapsum(x->sum(y[1] for y in x),iters) == sum(iters[1])*length(iters[2])
485485
end
486+
487+
@testset "run elsewhere" begin
488+
res_exp = sum(workers())
489+
for p in workers()
490+
res = @fetchfrom p pmapsum(x->myid(),1:nworkers())
491+
@test res == res_exp
492+
end
493+
end
486494

487495
@testset "errors" begin
488496
@test_throws exceptiontype pmapsum(x->throws(BoundsError()),1:10)
@@ -505,6 +513,15 @@ end
505513
@test res == pmapsum(plist->sum(x[1]^2 for x in plist),iterable)
506514
end
507515

516+
@testset "run elsewhere" begin
517+
iterable = 1:100
518+
res_exp = sum(iterable)
519+
for p in workers()
520+
res = @fetchfrom p pmapsum_elementwise(identity,iterable)
521+
@test res == res_exp
522+
end
523+
end
524+
508525
@testset "errors" begin
509526
@test_throws exceptiontype pmapsum_elementwise(x->throws(BoundsError()),1:10)
510527
end
@@ -529,6 +546,14 @@ end
529546
@test pmapreduce_commutative(x->myid(),prod,Iterators.product(1:nworkers(),1:1)) == prod(workers())
530547
end
531548

549+
@testset "run elsewhere" begin
550+
res_exp = prod(workers())
551+
for p in workers()
552+
res = @fetchfrom p pmapreduce_commutative(x->myid(),prod,1:nworkers())
553+
@test res == res_exp
554+
end
555+
end
556+
532557
@testset "errors" begin
533558
@test_throws exceptiontype pmapreduce_commutative(
534559
x->throws(BoundsError()),sum,1:10)
@@ -552,6 +577,15 @@ end
552577
@test res == sum(x->x^2,iter)
553578
end
554579

580+
@testset "run elsewhere" begin
581+
iter = 1:1000
582+
res_exp = sum(x->x^2,iter)
583+
for p in workers()
584+
res = @fetchfrom p pmapreduce_commutative_elementwise(x->x^2,sum,iter)
585+
@test res == res_exp
586+
end
587+
end
588+
555589
@testset "errors" begin
556590
@test_throws exceptiontype pmapreduce_commutative_elementwise(
557591
x->throws(BoundsError()),sum,1:10)
@@ -590,6 +624,14 @@ end
590624
@test pmapreduce(x->myid(),x->vcat(x...),1:nworkers()) == workers()
591625
end
592626

627+
@testset "run elsewhere" begin
628+
res_exp = sum(workers())
629+
for p in workers()
630+
res = @fetchfrom p pmapreduce(x->myid(),sum,1:nworkers())
631+
@test res == res_exp
632+
end
633+
end
634+
593635
@testset "errors" begin
594636
@test_throws exceptiontype pmapreduce(x->throws(BoundsError()),sum,1:10)
595637
@test_throws exceptiontype pmapreduce(identity,x->throws(BoundsError()),1:10)

0 commit comments

Comments
 (0)