Merge pull request #143 from ruby-concurrency/actress

Adding pools to Actor utils and other improvements

Author: Petr Chalupa

doc/actor/celluloid_benchmark.rb

@@ -48,7 +48,7 @@ def counting(count, ivar)
     b.report(format('%5d %4d %s', ADD_TO*counts_size, adders_size, 'concurrent')) do
       counts = Array.new(counts_size) { [0, Concurrent::IVar.new] }
       adders = Array.new(adders_size) do |i|
-        Concurrent::Actor::AdHoc.spawn("adder#{i}") do
+        Concurrent::Actor::Utils::AdHoc.spawn("adder#{i}") do
           lambda do |(count, ivar)|
             if count < ADD_TO
               adders[(i+1) % adders_size].tell [count+1, ivar]

lib/concurrent/actor.rb

@@ -26,7 +26,7 @@ module Actor
 
     require 'concurrent/actor/default_dead_letter_handler'
     require 'concurrent/actor/root'
-    require 'concurrent/actor/ad_hoc'
+    require 'concurrent/actor/utils'
 
     # @return [Reference, nil] current executing actor if any
     def self.current

lib/concurrent/actor/ad_hoc.rb

@@ -0,0 +1,10 @@
+module Concurrent
+  module Actor
+    module Utils
+      require 'concurrent/actor/utils/ad_hoc'
+      require 'concurrent/actor/utils/broadcast'
+      require 'concurrent/actor/utils/balancer'
+      require 'concurrent/actor/utils/pool'
+    end
+  end
+end

lib/concurrent/actor/utills.rb

@@ -0,0 +1,21 @@
+module Concurrent
+  module Actor
+    module Utils
+      # Allows quick creation of actors with behaviour defined by blocks.
+      # @example ping
+      #   AdHoc.spawn :forward, an_actor do |where|
+      #     # this block has to return proc defining #on_message behaviour
+      #     -> message { where.tell message  }
+      #   end
+      class AdHoc < Context
+        def initialize(*args, &initializer)
+          @on_message = Type! initializer.call(*args), Proc
+        end
+
+        def on_message(message)
+          instance_exec message, &@on_message
+        end
+      end
+    end
+  end
+end

lib/concurrent/actor/utils.rb

@@ -0,0 +1,40 @@
+module Concurrent
+  module Actor
+    module Utils
+
+      # Distributes messages between subscribed actors. Each actor'll get only one message then
+      # it's unsubscribed. The actor needs to resubscribe when it's ready to receive next message.
+      # @see Pool
+      class Balancer < RestartingContext
+
+        def initialize
+          @receivers = []
+          @buffer    = []
+        end
+
+        def on_message(message)
+          case message
+          when :subscribe
+            @receivers << envelope.sender
+            distribute
+            true
+          when :unsubscribe
+            @receivers.delete envelope.sender
+            true
+          when :subscribed?
+            @receivers.include? envelope.sender
+          else
+            @buffer << message
+            distribute
+          end
+        end
+
+        def distribute
+          while !@receivers.empty? && !@buffer.empty?
+            @receivers.shift << @buffer.shift
+          end
+        end
+      end
+    end
+  end
+end

lib/concurrent/actor/utils/ad_hoc.rb

@@ -4,8 +4,20 @@ module Concurrent
   module Actor
     module Utils
 
-      # TODO doc
-      class Broadcast < Context
+      # Allows to build pub/sub easily.
+      # @example news
+      #   news_channel = Concurrent::Actor::Utils::Broadcast.spawn :news
+      #
+      #   2.times do |i|
+      #     Concurrent::Actor::Utils::AdHoc.spawn "listener-#{i}" do
+      #       news_channel << :subscribe
+      #       -> message { puts message }
+      #     end
+      #   end
+      #
+      #   news_channel << 'Ruby rocks!'
+      #   # prints: 'Ruby rocks!' twice
+      class Broadcast < RestartingContext
 
         def initialize
           @receivers = Set.new
@@ -14,11 +26,14 @@ def initialize
         def on_message(message)
           case message
           when :subscribe
-            @receivers.add envelope.sender
-            true
+            if envelope.sender.is_a? Reference
+              @receivers.add envelope.sender
+              true
+            else
+              false
+            end
           when :unsubscribe
-            @receivers.delete envelope.sender
-            true
+            !!@receivers.delete(envelope.sender)
           when :subscribed?
             @receivers.include? envelope.sender
           else
@@ -31,6 +46,7 @@ def filtered_receivers
           @receivers
         end
       end
+
     end
   end
 end

lib/concurrent/actor/utils/balancer.rb

@@ -0,0 +1,59 @@
+require 'concurrent/actor/utils/balancer'
+
+module Concurrent
+  module Actor
+    module Utils
+
+      # Allows to create a pool of workers and distribute work between them
+      # @param [Integer] size number of workers
+      # @yield [balancer, index] a block spawning an worker instance. called +size+ times.
+      #   The worker should be descendant of AbstractWorker and supervised, see example.
+      # @yieldparam [Balancer] balancer to pass to the worker
+      # @yieldparam [Integer] index of the worker, usually used in its name
+      # @yieldreturn [Reference] the reference of newly created worker
+      # @example
+      #     class Worker < Concurrent::Actor::Utils::AbstractWorker
+      #       def work(message)
+      #         p message * 5
+      #       end
+      #     end
+      #
+      #     pool = Concurrent::Actor::Utils::Pool.spawn! 'pool', 5 do |balancer, index|
+      #       Worker.spawn name: "worker-#{index}", supervise: true, args: [balancer]
+      #     end
+      #
+      #     pool << 'asd' << 2
+      #     # prints:
+      #     # "asdasdasdasdasd"
+      #     # 10
+      class Pool < RestartingContext
+        def initialize(size, &worker_initializer)
+          @balancer = Balancer.spawn name: :balancer, supervise: true
+          @workers  = Array.new(size, &worker_initializer.curry[@balancer])
+          @workers.each { |w| Type! w, Reference }
+        end
+
+        def on_message(message)
+          @balancer << message
+        end
+      end
+
+      class AbstractWorker < RestartingContext
+        def initialize(balancer)
+          @balancer = balancer
+          @balancer << :subscribe
+        end
+
+        def on_message(message)
+          work message
+        ensure
+          @balancer << :subscribe
+        end
+
+        def work(message)
+          raise NotImplementedError
+        end
+      end
+    end
+  end
+end

lib/concurrent/actor/utils/broadcast.rb

@@ -1,12 +1,14 @@
 require 'delegate'
 require 'concurrent/executor/executor'
 require 'concurrent/logging'
+require 'concurrent/atomic/synchronization'
 
 module Concurrent
 
   # Ensures passed jobs in a serialized order never running at the same time.
   class SerializedExecution
     include Logging
+    include Synchronization
 
     Job = Struct.new(:executor, :args, :block) do
       def call
@@ -15,9 +17,10 @@ def call
     end
 
     def initialize
-      @being_executed = false
-      @stash          = []
-      @mutex          = Mutex.new
+      synchronize do
+        @being_executed = false
+        @stash          = []
+      end
     end
 
     # Submit a task to the executor for asynchronous processing.
@@ -33,23 +36,36 @@ def initialize
     #
     # @raise [ArgumentError] if no task is given
     def post(executor, *args, &task)
-      return nil if task.nil?
-
-      job = Job.new executor, args, task
-
-      begin
-        @mutex.lock
-        post = if @being_executed
-                 @stash << job
-                 false
-               else
-                 @being_executed = true
-               end
-      ensure
-        @mutex.unlock
+      posts [[executor, args, task]]
+      true
+    end
+
+    # As {#post} but allows to submit multiple tasks at once, it's guaranteed that they will not
+    # be interleaved by other tasks.
+    #
+    # @param [Array<Array(Executor, Array<Object>, Proc)>] posts array of triplets where
+    #   first is a {Executor}, second is array of args for task, third is a task (Proc)
+    def posts(posts)
+      # if can_overflow?
+      #   raise ArgumentError, 'SerializedExecution does not support thread-pools which can overflow'
+      # end
+
+      return nil if posts.empty?
+
+      jobs = posts.map { |executor, args, task| Job.new executor, args, task }
+
+      job_to_post = synchronize do
+        if @being_executed
+          @stash.push(*jobs)
+          nil
+        else
+          @being_executed = true
+          @stash.push(*jobs[1..-1])
+          jobs.first
+        end
       end
 
-      call_job job if post
+      call_job job_to_post if job_to_post
       true
     end
 
@@ -78,11 +94,8 @@ def call_job(job)
     def work(job)
       job.call
     ensure
-      begin
-        @mutex.lock
+      synchronize do
         job = @stash.shift || (@being_executed = false)
-      ensure
-        @mutex.unlock
       end
 
       call_job job if job
@@ -98,7 +111,7 @@ class SerializedExecutionDelegator < SimpleDelegator
     include SerialExecutor
 
     def initialize(executor)
-      @executor = executor
+      @executor   = executor
       @serializer = SerializedExecution.new
       super(executor)
     end