Rename OneByOne to SerializedExecution add basic rejection handling

Author: pitr-ch

lib/concurrent/actress.rb

@@ -1,5 +1,5 @@
 require 'concurrent/configuration'
-require 'concurrent/executor/one_by_one'
+require 'concurrent/executor/serialized_execution'
 require 'concurrent/ivar'
 require 'concurrent/logging'
 

lib/concurrent/actress/core.rb

@@ -23,14 +23,14 @@ class Core
       #   can be used to hook actor instance to any logging system
       # @param [Proc] block for class instantiation
       def initialize(opts = {}, &block)
-        @mailbox    = Array.new
-        @one_by_one = OneByOne.new
+        @mailbox              = Array.new
+        @serialized_execution = SerializedExecution.new
         # noinspection RubyArgCount
-        @terminated = Event.new
-        @executor   = Type! opts.fetch(:executor, Concurrent.configuration.global_task_pool), Executor
-        @children   = Set.new
-        @reference  = Reference.new self
-        @name       = (Type! opts.fetch(:name), String, Symbol).to_s
+        @terminated           = Event.new
+        @executor             = Type! opts.fetch(:executor, Concurrent.configuration.global_task_pool), Executor
+        @children             = Set.new
+        @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)
@@ -180,7 +180,7 @@ def receive_envelope
       # Schedules blocks to be executed on executor sequentially,
       # sets Actress.current
       def schedule_execution
-        @one_by_one.post(@executor) do
+        @serialized_execution.post(@executor) do
           begin
             Thread.current[:__current_actress__] = reference
             yield

lib/concurrent/agent.rb

@@ -62,14 +62,14 @@ class Agent
     # @option opts [String] :copy_on_deref (nil) call the given `Proc` passing the internal value and
     #   returning the value returned from the proc
     def initialize(initial, opts = {})
-      @value              = initial
-      @rescuers           = []
-      @validator          = Proc.new { |result| true }
-      @timeout            = opts.fetch(:timeout, TIMEOUT).freeze
-      self.observers      = CopyOnWriteObserverSet.new
-      @one_by_one         = OneByOne.new
-      @task_executor      = OptionsParser.get_task_executor_from(opts)
-      @operation_executor = OptionsParser.get_operation_executor_from(opts)
+      @value                = initial
+      @rescuers             = []
+      @validator            = Proc.new { |result| true }
+      @timeout              = opts.fetch(:timeout, TIMEOUT).freeze
+      self.observers        = CopyOnWriteObserverSet.new
+      @serialized_execution = SerializedExecution.new
+      @task_executor        = OptionsParser.get_task_executor_from(opts)
+      @operation_executor   = OptionsParser.get_operation_executor_from(opts)
       init_mutex
       set_deref_options(opts)
     end
@@ -180,7 +180,7 @@ def await(timeout = nil)
 
     def post_on(executor, &block)
       return nil if block.nil?
-      @one_by_one.post(executor) { work(&block) }
+      @serialized_execution.post(executor) { work(&block) }
       true
     end
 

lib/concurrent/executor/one_by_one.rb renamed to lib/concurrent/executor/serialized_execution.rb

@@ -1,8 +1,10 @@
+require 'concurrent/logging'
+
 module Concurrent
 
-  # Ensures that jobs are passed to the given executors one by one,
-  # never running at the same time.
-  class OneByOne
+  # Ensures passed jobs in a serialized order never running at the same time.
+  class SerializedExecution
+    include Logging
 
     Job = Struct.new(:executor, :args, :block) do
       def call
@@ -30,10 +32,6 @@ def initialize
     # @raise [ArgumentError] if no task is given
     def post(executor, *args, &task)
       return nil if task.nil?
-      # FIXME Agent#send-off will blow up here
-      # if executor.can_overflow?
-      #   raise ArgumentError, 'OneByOne cannot be used in conjunction with executor which may overflow'
-      # end
 
       job = Job.new executor, args, task
 
@@ -56,7 +54,22 @@ def post(executor, *args, &task)
     private
 
     def call_job(job)
-      job.executor.post { work(job) }
+      did_it_run = begin
+        job.executor.post { work(job) }
+        true
+      rescue RejectedExecutionError => ex
+        false
+      end
+
+      # TODO not the best idea to run it myself
+      unless did_it_run
+        begin
+          work job
+        rescue => ex
+          # let it fail
+          log DEBUG, ex
+        end
+      end
     end
 
     # ensures next job is executed if any is stashed

lib/concurrent/executors.rb

@@ -6,4 +6,4 @@
 require 'concurrent/executor/single_thread_executor'
 require 'concurrent/executor/thread_pool_executor'
 require 'concurrent/executor/timer_set'
-require 'concurrent/executor/one_by_one'
+require 'concurrent/executor/serialized_execution'

spec/concurrent/agent_spec.rb

@@ -164,7 +164,7 @@ def trigger_observable(observable)
         subject.post { nil }
         sleep(0.1)
         subject.
-            instance_variable_get(:@one_by_one).
+            instance_variable_get(:@serialized_execution).
             instance_variable_get(:@stash).
             size.should eq 2
       end