Fixing simple carrier to await the full event chain.

Author: rcschrg

src/carrier/simple.jl

@@ -6,14 +6,16 @@ using UUIDs
 abstract type AbstractSimpleCarrier <: Carrier end
 
 """
-    ActorContainer  
+    ActorContainer
 
     A container to manage multiple `SimpleCarrier`.
 """
-struct ActorContainer
+mutable struct ActorContainer
     actors::Vector{AbstractSimpleCarrier}
+    active_tasks::Threads.Atomic{Int}
+    done_event::Base.Event
     function ActorContainer()
-        return new(Vector{AbstractSimpleCarrier}())
+        return new(Vector{AbstractSimpleCarrier}(), Threads.Atomic{Int}(0), Base.Event())
     end
 end
 
@@ -65,8 +67,15 @@ function send_to_other(carrier::SimpleCarrier, content::Any, receiver::Real; met
     other_carrier::Carrier = carrier.container.actors[receiver]
     main_meta = Dict(:sender => carrier.aid, :message_id => uuid4())
     union_meta = merge(main_meta, meta) # important: meta can override main_meta entries
+    Threads.atomic_add!(carrier.container.active_tasks, 1)
     return @spawnlog begin
-        _dispatch_to(other_carrier, content, union_meta)
+        try
+            _dispatch_to(other_carrier, content, union_meta)
+        finally
+            if Threads.atomic_sub!(carrier.container.active_tasks, 1) == 1
+                notify(carrier.container.done_event)
+            end
+        end
     end
 end
 
@@ -113,7 +122,8 @@ Return a waitable object that can be used to monitor the progress of the optimiz
 function start_distributed_optimization(actors::Vector{<:DistributedAlgorithm}, start_message::Any)
     actor_container = ActorContainer()
     carriers = [SimpleCarrier(actor_container, actor) for actor in actors]
-    return send_to_other(carriers[1], start_message, cid(carriers[2]))
+    send_to_other(carriers[1], start_message, cid(carriers[2]))
+    return Threads.@spawn wait(actor_container.done_event)
 end
 
 """

test/cohda/cohda_sc_tests.jl

@@ -10,7 +10,8 @@ using Test
 
     initial_message = create_cohda_start_message([1.2, 2, 3])
 
-    wait(send_to_other(actor_one, initial_message, cid(actor_two)))
+    send_to_other(actor_one, initial_message, cid(actor_two))
+    wait(container.done_event)
 
     @test actor_one.actor.memory.solution_candidate.perf < 0
 end
@@ -23,6 +24,6 @@ end
     initial_message = create_cohda_start_message([1.2, 2, 3])
 
     wait(start_distributed_optimization([actor_one, actor_two], initial_message))
-
+    @test !isnothing(actor_one.memory.solution_candidate.perf)
     @test actor_one.memory.solution_candidate.perf < 0
 end