Rename Supervising to Supervised, Add Supervising

allows to set handler: :reset!, :resume!, :terminate!
and strategies: :one_for_one, :one_for_all

Author: pitr-ch

lib/concurrent/actor/behaviour.rb

@@ -27,6 +27,7 @@ module Behaviour
       require 'concurrent/actor/behaviour/pausing'
       require 'concurrent/actor/behaviour/removes_child'
       require 'concurrent/actor/behaviour/sets_results'
+      require 'concurrent/actor/behaviour/supervised'
       require 'concurrent/actor/behaviour/supervising'
       require 'concurrent/actor/behaviour/termination'
       require 'concurrent/actor/behaviour/terminates_children'
@@ -38,7 +39,8 @@ def self.basic_behaviour_definition
 
       def self.restarting_behaviour_definition
         [*base,
-         *supervising,
+         *supervised,
+         [Behaviour::Supervising, [:reset!, :one_for_one]],
          *user_messages(:pause)]
       end
 
@@ -51,8 +53,8 @@ def self.base
          [Linking, []]]
       end
 
-      def self.supervising
-        [[Supervising, []],
+      def self.supervised
+        [[Supervised, []],
          [Pausing, []]]
       end
 

lib/concurrent/actor/behaviour/pausing.rb

@@ -40,16 +40,16 @@ def pause!(error = nil)
           true
         end
 
-        def resume!
+        def resume!(broadcast = true)
           @buffer.each { |envelope| core.schedule_execution { pass envelope } }
           @buffer.clear
           @paused = false
-          broadcast(:resumed)
+          broadcast(:resumed) if broadcast
           true
         end
 
         def from_supervisor?(envelope)
-          if behaviour!(Supervising).supervisor == envelope.sender
+          if behaviour!(Supervised).supervisor == envelope.sender
             yield
           else
             false
@@ -59,8 +59,8 @@ def from_supervisor?(envelope)
         def reset!
           core.allocate_context
           core.build_context
+          resume!(false)
           broadcast(:reset)
-          resume!
           true
         end
 

lib/concurrent/actor/behaviour/sets_results.rb

@@ -23,6 +23,8 @@ def on_envelope(envelope)
             terminate!
           when :pause
             behaviour!(Pausing).pause!(error)
+          when :just_log
+            # nothing
           else
             raise
           end

lib/concurrent/actor/behaviour/supervised.rb

@@ -0,0 +1,51 @@
+module Concurrent
+  module Actor
+    module Behaviour
+
+      # Sets nad holds the supervisor of the actor if any. There is only one or none supervisor
+      # for each actor. Each supervisor is automatically linked.
+      class Supervised < Abstract
+        attr_reader :supervisor
+
+        def initialize(core, subsequent)
+          super core, subsequent
+          @supervisor = nil
+        end
+
+        def on_envelope(envelope)
+          case envelope.message
+          when :supervise
+            supervise envelope.sender
+          when :supervisor
+            supervisor
+          when :un_supervise
+            un_supervise envelope.sender
+          else
+            pass envelope
+          end
+        end
+
+        def supervise(ref)
+          @supervisor = ref
+          behaviour!(Linking).link ref
+          true
+        end
+
+        def un_supervise(ref)
+          if @supervisor == ref
+            behaviour!(Linking).unlink ref
+            @supervisor = nil
+            true
+          else
+            false
+          end
+        end
+
+        def on_event(event)
+          @supervisor = nil if event == :terminated
+          super event
+        end
+      end
+    end
+  end
+end

lib/concurrent/actor/behaviour/supervising.rb

@@ -1,50 +1,33 @@
 module Concurrent
   module Actor
     module Behaviour
-
-      # Sets nad holds the supervisor of the actor if any. There is only one or none supervisor
-      # for each actor. Each supervisor is automatically linked.
       class Supervising < Abstract
-        attr_reader :supervisor
-
-        def initialize(core, subsequent)
+        def initialize(core, subsequent, handle, strategy)
           super core, subsequent
-          @supervisor = nil
+          @handle   = Match! handle, :terminate!, :resume!, :reset!, :restart!
+          @strategy = case @handle
+                      when :terminate!
+                        Match! strategy, nil
+                      when :resume!
+                        Match! strategy, :one_for_one
+                      when :reset!, :restart!
+                        Match! strategy, :one_for_one, :one_for_all
+                      end
         end
 
         def on_envelope(envelope)
           case envelope.message
-          when :supervise
-            supervise envelope.sender
-          when :supervisor
-            supervisor
-          when :un_supervise
-            un_supervise envelope.sender
+          when Exception, :paused
+            receivers = if @strategy == :one_for_all
+                          children
+                        else
+                          [envelope.sender]
+                        end
+            receivers.each { |ch| ch << @handle }
           else
             pass envelope
           end
         end
-
-        def supervise(ref)
-          @supervisor = ref
-          behaviour!(Linking).link ref
-          true
-        end
-
-        def un_supervise(ref)
-          if @supervisor == ref
-            behaviour!(Linking).unlink ref
-            @supervisor = nil
-            true
-          else
-            false
-          end
-        end
-
-        def on_event(event)
-          @supervisor = nil if event == :terminated
-          super event
-        end
       end
     end
   end

lib/concurrent/actor/core.rb

@@ -50,7 +50,7 @@ def initialize(opts = {}, &block)
         @serialized_execution = SerializedExecution.new
         @executor             = Type! opts.fetch(:executor, Concurrent.configuration.global_task_pool), Executor
         @children             = Set.new
-        @context_class        = Child! opts.fetch(:class), Context
+        @context_class        = Child! opts.fetch(:class), AbstractContext
         allocate_context
         @reference = (Child! opts[:reference_class] || @context.default_reference_class, Reference).new self
         @name      = (Type! opts.fetch(:name), String, Symbol).to_s

lib/concurrent/actor/default_dead_letter_handler.rb

@@ -1,6 +1,6 @@
 module Concurrent
   module Actor
-    class DefaultDeadLetterHandler < Context
+    class DefaultDeadLetterHandler < RestartingContext
       def on_message(dead_letter)
         log Logging::INFO, "got dead letter #{dead_letter.inspect}"
       end

lib/concurrent/actor/root.rb

@@ -1,12 +1,14 @@
 module Concurrent
   module Actor
     # implements the root actor
-    class Root < Context
+    class Root < AbstractContext
 
       def initialize
-        @dead_letter_router = Core.new(parent: reference,
-                                       class:  DefaultDeadLetterHandler,
-                                       name:   :default_dead_letter_handler).reference
+        # noinspection RubyArgCount
+        @dead_letter_router = Core.new(parent:    reference,
+                                       class:     DefaultDeadLetterHandler,
+                                       supervise: true,
+                                       name:      :default_dead_letter_handler).reference
       end
 
       # to allow spawning of new actors, spawn needs to be called inside the parent Actor
@@ -21,6 +23,12 @@ def on_message(message)
       def dead_letter_routing
         @dead_letter_router
       end
+
+      def behaviour_definition
+        [*Behaviour.base,
+         [Behaviour::Supervising, [:reset!, :one_for_one]],
+         *Behaviour.user_messages(:just_log)]
+      end
     end
   end
 end

spec/concurrent/actor_spec.rb

@@ -277,8 +277,16 @@ def on_message(message)
 
       describe 'pausing' do
         it 'pauses on error' do
-          queue = Queue.new
-          test  = AdHoc.spawn :tester do
+          queue              = Queue.new
+          resuming_behaviour = Behaviour.restarting_behaviour_definition.map do |c, args|
+            if Behaviour::Supervising == c
+              [c, [:resume!, :one_for_one]]
+            else
+              [c, args]
+            end
+          end
+
+          test = AdHoc.spawn name: :tester, behaviour_definition: resuming_behaviour do
             actor = AdHoc.spawn name:                 :pausing,
                                 behaviour_definition: Behaviour.restarting_behaviour_definition do
               queue << :init
@@ -288,50 +296,44 @@ def on_message(message)
             actor << :supervise
             queue << actor.ask!(:supervisor)
             actor << nil
+            queue << actor.ask(:add)
 
             -> m do
               queue << m
-              if UnknownMessage === m
-                ivar = envelope.sender.ask(:add)
-                envelope.sender << :resume!
-                queue << ivar.value!
-              end
             end
           end
 
           expect(queue.pop).to eq :init
           expect(queue.pop).to eq test
-          expect(queue.pop).to be_kind_of(UnknownMessage)
-          expect(queue.pop).to eq 1
+          expect(queue.pop.value).to eq 1
           expect(queue.pop).to eq :resumed
           terminate_actors test
 
-          test = AdHoc.spawn :tester do
+          test = AdHoc.spawn name:                 :tester,
+                             behaviour_definition: Behaviour.restarting_behaviour_definition do
             actor = AdHoc.spawn name:                 :pausing,
                                 supervise:            true,
                                 behaviour_definition: Behaviour.restarting_behaviour_definition do
               queue << :init
-              -> m { m == :add ? 1 : pass }
+              -> m { m == :object_id ? self.object_id : pass }
             end
 
             queue << actor.ask!(:supervisor)
+            queue << actor.ask!(:object_id)
             actor << nil
+            queue << actor.ask(:object_id)
 
             -> m do
               queue << m
-              if UnknownMessage === m
-                ivar = envelope.sender.ask(:add)
-                envelope.sender << :reset!
-                queue << ivar.value!
-              end
             end
           end
 
           expect(queue.pop).to eq :init
           expect(queue.pop).to eq test
-          expect(queue.pop).to be_kind_of(UnknownMessage)
+          first_id  = queue.pop
+          second_id = queue.pop.value
+          expect(first_id).not_to eq second_id # context already reset
           expect(queue.pop).to eq :init # rebuilds context
-          expect(queue.pop).to eq 1
           expect(queue.pop).to eq :reset
           terminate_actors test
         end