Configurable Core

executor can be configured, other options will follow
spawn can be also called with hash instead of class, name, *args

Author: pitr-ch

lib/concurrent/actress.rb

@@ -40,25 +40,38 @@ class Root
       def on_message(message)
         case message.first
         when :spawn
-          spawn *message[1..2], *message[3], &message[4]
+          spawn message[1], &message[2]
         else
           # ignore
         end
       end
     end
 
     # A root actor, a default parent of all actors spawned outside an actor
-    ROOT = Core.new(nil, '/', Root).reference
+    ROOT = Core.new(parent: nil, name: '/', class: Root).reference
 
-    # @param [Context] actress_class to be spawned
-    # @param [String, Symbol] name of the instance, it's used to generate the path of the actor
-    # @param args for actress_class instantiation
     # @param block for actress_class instantiation
-    def self.spawn(actress_class, name, *args, &block)
+    def self.spawn(*args, &block)
       if Actress.current
-        Core.new(Actress.current, name, actress_class, *args, &block).reference
+        Core.new(spawn_optionify(*args).merge(parent: Actress.current), &block).reference
       else
-        ROOT.ask([:spawn, actress_class, name, args, block]).value
+        ROOT.ask([:spawn, spawn_optionify(*args), block]).value
+      end
+    end
+
+    # @overload spawn_optionify(actress_class, name, *args)
+    #   @param [Context] actress_class to be spawned
+    #   @param [String, Symbol] name of the instance, it's used to generate the path of the actor
+    #   @param args for actress_class instantiation
+    # @overload spawn_optionify(opts)
+    #   see {Core.new} opts
+    def self.spawn_optionify(*args)
+      if args.size == 1 && args.first.is_a?(Hash)
+        args.first
+      else
+        { class: args[0],
+          name:  args[1],
+          args:  args[2..-1] }
       end
     end
   end

lib/concurrent/actress/context.rb

@@ -37,8 +37,8 @@ def on_envelope(envelope)
       end
 
       # @see Actress.spawn
-      def spawn(actress_class, name, *args, &block)
-        Actress.spawn(actress_class, name, *args, &block)
+      def spawn(*args, &block)
+        Actress.spawn(*args, &block)
       end
 
       # @see Core#children
@@ -70,8 +70,13 @@ def self.included(base)
 
       module ClassMethods
         # behaves as {Actress.spawn} but class_name is omitted
-        def spawn(name, *args, &block)
-          Actress.spawn self, name, *args, &block
+        def spawn(name_or_opts, *args, &block)
+          opts = if name_or_opts.is_a? Hash
+                   name_or_opts.merge class: self
+                 else
+                   { class: self, name: name_or_opts, args: args }
+                 end
+          Actress.spawn opts, &block
         end
       end
     end

lib/concurrent/actress/core.rb

@@ -9,28 +9,37 @@ class Core
       attr_reader :reference, :name, :path, :logger, :parent_core
       private :parent_core
 
-      # @param [Reference, nil] parent of an actor spawning this one
-      # @param [String] name
-      # @param [Context] actress_class a class to be instantiated defining Actor's behaviour
-      # @param args arguments for actress_class instantiation
-      # @param block for actress_class instantiation
-      def initialize(parent, name, actress_class, *args, &block)
-        @mailbox         = Array.new
-        @one_by_one      = OneByOne.new
-        @executor        = Concurrent.configuration.global_task_pool # TODO make configurable
-        @parent_core     = (Type! parent, Reference, NilClass) && parent.send(:core)
-        @name            = (Type! name, String, Symbol).to_s
-        @children        = []
+      # @option opts [String] name
+      # @option opts [Reference, nil] parent of an actor spawning this one
+      # @option opts [Context] actress_class a class to be instantiated defining Actor's behaviour
+      # @option opts [Array<Object>] args arguments for actress_class instantiation
+      # @option opts [Executor] executor, default is `Concurrent.configuration.global_task_pool`
+      # @param [Proc] block for class instantiation
+      def initialize(opts = {}, &block)
+        @mailbox    = Array.new
+        @one_by_one = OneByOne.new
+        # noinspection RubyArgCount
+        @terminated = Event.new
+        @executor   = Type! opts.fetch(:executor, Concurrent.configuration.global_task_pool), Executor
+        @children   = []
+        @reference  = Reference.new self
+        @name       = (Type! opts.fetch(:name), String, Symbol).to_s
+
+        parent       = opts[:parent]
+        @parent_core = (Type! parent, Reference, NilClass) && parent.send(:core)
+        if @parent_core.nil? && @name != '/'
+          raise 'only root has no parent'
+        end
+
         @path            = @parent_core ? File.join(@parent_core.path, @name) : @name
         @logger          = Logger.new($stderr) # TODO add proper logging
         @logger.progname = @path
-        @reference       = Reference.new self
-        # noinspection RubyArgCount
-        @terminated      = Event.new
 
-        parent_core.add_child reference if parent_core
+        @parent_core.add_child reference if @parent_core
+
+        @actress_class = actress_class = Child! opts.fetch(:class), Context
+        args           = opts.fetch(:args, [])
 
-        @actress_class = Child! actress_class, Context
         schedule_execution do
           begin
             @actress = actress_class.new *args, &block
@@ -42,9 +51,9 @@ def initialize(parent, name, actress_class, *args, &block)
         end
       end
 
-      # @return [Reference] of parent actor
+      # @return [Reference, nil] of parent actor
       def parent
-        @parent_core.reference
+        @parent_core && @parent_core.reference
       end
 
       # @return [Array<Reference>] of children actors
@@ -155,7 +164,7 @@ def schedule_execution
             Thread.current[:__current_actress__] = reference
             yield
           rescue => e
-            logger.error e
+            logger.fatal e
           ensure
             Thread.current[:__current_actress__] = nil
           end

spec/actress.rb

@@ -36,18 +36,19 @@ def on_message(message)
 
 def assert condition
   unless condition
-    require 'pry'
-    binding.pry
+    # require 'pry'
+    # binding.pry
     raise
+    puts "--- \n#{caller.join("\n")}"
   end
 end
 
 
-Array.new(100).map do
-  Thread.new do
-    20.times do |i|
-      # it format('--- %3d ---', i) do
-      puts format('--- %3d ---', i)
+# it format('--- %3d ---', i) do
+10.times do |i|
+  puts format('--- %3d ---', i)
+  Array.new(10).map do
+    Thread.new do
       # trace! do
       queue = Queue.new
       actor = Ping.spawn :ping, queue
@@ -71,8 +72,7 @@ def assert condition
       actor << :terminate
       assert actor.ask(:blow_up).wait.rejected?
     end
-  end
-end.each(&:join)
-
-# end
+  end.each(&:join)
+end
+#   end
 # end