actor.ask(:terminate!) is only completed after all its children terminate

Author: pitr-ch

lib/concurrent/actor/behaviour.rb

@@ -37,10 +37,6 @@ module Actor
     #
     #     > {include:Actor::Behaviour::Termination}
     #
-    # -   {Behaviour::TerminatesChildren}:
-    #
-    #     > {include:Actor::Behaviour::TerminatesChildren}
-    #
     # -   {Behaviour::RemovesChild}:
     #
     #     > {include:Actor::Behaviour::RemovesChild}
@@ -66,14 +62,12 @@ module Behaviour
       require 'concurrent/actor/behaviour/sets_results'
       require 'concurrent/actor/behaviour/supervising'
       require 'concurrent/actor/behaviour/termination'
-      require 'concurrent/actor/behaviour/terminates_children'
 
       # Array of behaviours and their construction parameters.
       #
       #     [[Behaviour::SetResults, :terminate!],
       #      [Behaviour::RemovesChild],
       #      [Behaviour::Termination],
-      #      [Behaviour::TerminatesChildren],
       #      [Behaviour::Linking],
       #      [Behaviour::Awaits],
       #      [Behaviour::ExecutesContext],
@@ -91,7 +85,6 @@ def self.basic_behaviour_definition
       #     [[Behaviour::SetResults, :pause!],
       #      [Behaviour::RemovesChild],
       #      [Behaviour::Termination],
-      #      [Behaviour::TerminatesChildren],
       #      [Behaviour::Linking],
       #      [Behaviour::Pausing],
       #      [Behaviour::Supervising, :reset!, :one_for_one],
@@ -113,8 +106,7 @@ def self.base(on_error)
         [[SetResults, on_error],
          # has to be before Termination to be able to remove children from terminated actor
          RemovesChild,
-         Termination,
-         TerminatesChildren]
+         Termination]
       end
 
       # @see '' its source code

lib/concurrent/actor/behaviour/terminates_children.rb

@@ -2,22 +2,22 @@ module Concurrent
   module Actor
     module Behaviour
 
-      # Handles actor termination.
+      # Handles actor termination. Waits until all its children are terminated,
+      # can be configured on behaviour initialization.
       # @note Actor rejects envelopes when terminated.
       # @note TODO missing example
       class Termination < Abstract
 
         # @!attribute [r] terminated
         #   @return [Edge::Event] event which will become set when actor is terminated.
-        # @!attribute [r] reason
-        attr_reader :terminated, :reason
+        attr_reader :terminated
 
-        def initialize(core, subsequent, core_options, trapping = false)
+        def initialize(core, subsequent, core_options, trapping = false, terminate_children = true)
           super core, subsequent, core_options
-          @terminated        = Concurrent.event
-          @public_terminated = @terminated.hide_completable
-          @reason            = nil
-          @trapping          = trapping
+          @terminated         = Concurrent.future
+          @public_terminated  = @terminated.hide_completable
+          @trapping           = trapping
+          @terminate_children = terminate_children
         end
 
         # @note Actor rejects envelopes when terminated.
@@ -43,7 +43,7 @@ def on_envelope(envelope)
             if trapping? && reason != :kill
               pass envelope
             else
-              terminate! reason
+              terminate! reason, envelope
             end
           when :termination_event
             @public_terminated
@@ -59,14 +59,23 @@ def on_envelope(envelope)
 
         # Terminates the actor. Any Envelope received after termination is rejected.
         # Terminates all its children, does not wait until they are terminated.
-        def terminate!(reason = :normal)
-          # TODO return after all children are terminated
+        def terminate!(reason = nil, envelope = nil)
           return true if terminated?
-          @reason = reason
-          terminated.complete
+
+          self_termination = Concurrent.completed_future(reason.nil?, reason.nil? || nil, reason)
+          all_terminations = if @terminate_children
+                               Concurrent.zip(*children.map { |ch| ch.ask(:terminate!) }, self_termination)
+                             else
+                               self_termination
+                             end
+
+          all_terminations.chain_completable(@terminated)
+          all_terminations.chain_completable(envelope.future) if envelope && envelope.future
+
           broadcast(true, [:terminated, reason]) # TODO do not end up in Dead Letter Router
           parent << :remove_child if parent
-          true
+
+          MESSAGE_PROCESSED
         end
       end
     end

lib/concurrent/actor/behaviour/termination.rb

@@ -18,10 +18,10 @@ def terminated?
         behaviour!(Behaviour::Termination).terminated?
       end
 
-      # @see Termination#reason
-      def reason
-        behaviour!(Behaviour::Termination).reason
-      end
+      # # @see Termination#reason
+      # def reason
+      #   behaviour!(Behaviour::Termination).reason
+      # end
 
       # delegates to core.log
       # @see Logging#log

lib/concurrent/actor/internal_delegations.rb

@@ -18,7 +18,6 @@ def on_message(message)
       #     # this block has to return proc defining #on_message behaviour
       #     -> message { where.tell message  }
       #   end
-      # @note TODO remove in favor of the module
       class AdHoc < Context
         include AsAdHoc
       end

lib/concurrent/actor/utils/ad_hoc.rb

@@ -153,7 +153,6 @@ def on_message(message)
           expect(subject.ask!(:terminated?)).to be_falsey
           subject.ask(:terminate!).wait
           expect(subject.ask!(:terminated?)).to be_truthy
-          child.ask!(:termination_event).wait
           expect(child.ask!(:terminated?)).to be_truthy
 
           terminate_actors subject, child
@@ -315,18 +314,24 @@ def on_message(message)
 
       describe 'pool' do
         it 'supports asks' do
-          pool = Concurrent::Actor::Utils::Pool.spawn! 'pool', 5 do |index|
-            Concurrent::Actor::Utils::AdHoc.spawn name: "worker-#{index}", supervised: true do
+          children = Queue.new
+          pool     = Concurrent::Actor::Utils::Pool.spawn! 'pool', 5 do |index|
+            worker = Concurrent::Actor::Utils::AdHoc.spawn name: "worker-#{index}", supervised: true do
               lambda do |message|
                 fail if message == :fail
                 5 + message
               end
             end
+            children.push worker
+            worker
           end
 
           10.times { expect(pool.ask!(5)).to eq 10 }
           expect(pool.ask(:fail).reason).to be_kind_of RuntimeError
           expect(pool.ask!(5)).to eq 10
+          expect(pool.ask!(:terminate!)).to be_truthy
+          5.times { expect(children.pop.ask!(:terminated?)).to be_truthy }
+
           terminate_actors pool
         end
       end