caller_runs execution should not block the work queue

caller_runs execution was being done while the executor lock was held.
This caused the entire executor’s work queue to be blocked, with worker
threads not able to take new work. We’ve restructured so that the
execution is done as deferred work, outside of the lock.

Author: chrisseaton

lib/concurrent-ruby/concurrent/executor/abstract_executor_service.rb

@@ -75,28 +75,31 @@ def auto_terminate=(value)
 
     private
 
-    # Handler which executes the `fallback_policy` once the queue size
-    # reaches `max_queue`.
+    # Returns an action which executes the `fallback_policy` once the queue
+    # size reaches `max_queue`. The reason for the indirection of an action
+    # is so that the work can be deferred outside of synchronization.
     #
     # @param [Array] args the arguments to the task which is being handled.
     #
     # @!visibility private
-    def handle_fallback(*args)
+    def fallback_action(*args)
       case fallback_policy
       when :abort
-        raise RejectedExecutionError
+        lambda { raise RejectedExecutionError }
       when :discard
-        false
+        lambda { false }
       when :caller_runs
-        begin
-          yield(*args)
-        rescue => ex
-          # let it fail
-          log DEBUG, ex
-        end
-        true
+        lambda {
+          begin
+            yield(*args)
+          rescue => ex
+            # let it fail
+            log DEBUG, ex
+          end
+          true
+        }
       else
-        fail "Unknown fallback policy #{fallback_policy}"
+        lambda { fail "Unknown fallback policy #{fallback_policy}" }
       end
     end
 

lib/concurrent-ruby/concurrent/executor/java_executor_service.rb

@@ -20,7 +20,7 @@ class JavaExecutorService < AbstractExecutorService
 
       def post(*args, &task)
         raise ArgumentError.new('no block given') unless block_given?
-        return handle_fallback(*args, &task) unless running?
+        return fallback_action(*args, &task).call unless running?
         @executor.submit Job.new(args, task)
         true
       rescue Java::JavaUtilConcurrent::RejectedExecutionException

lib/concurrent-ruby/concurrent/executor/ruby_executor_service.rb

@@ -16,10 +16,19 @@ def initialize(*args, &block)
 
     def post(*args, &task)
       raise ArgumentError.new('no block given') unless block_given?
+      deferred_action = nil
       synchronize do
         # If the executor is shut down, reject this task
-        return handle_fallback(*args, &task) unless running?
-        ns_execute(*args, &task)
+        unless running?
+          deferred_action = fallback_action(*args, &task)
+          break
+        end
+        deferred_action = ns_execute(*args, &task)
+      end
+
+      if deferred_action
+        deferred_action.call
+      else
         true
       end
     end

lib/concurrent-ruby/concurrent/executor/ruby_thread_pool_executor.rb

@@ -156,10 +156,11 @@ def ns_execute(*args, &task)
       if ns_assign_worker(*args, &task) || ns_enqueue(*args, &task)
         @scheduled_task_count += 1
       else
-        handle_fallback(*args, &task)
+        return fallback_action(*args, &task)
       end
 
       ns_prune_pool if @next_gc_time < Concurrent.monotonic_time
+      nil
     end
 
     # @!visibility private

spec/concurrent/executor/thread_pool_executor_shared.rb

@@ -616,6 +616,46 @@
         executor << proc { latch.count_down }
         latch.wait(0.1)
       end
+
+      specify '#post does not block other jobs running on the worker threads' do
+        log = Queue.new
+
+        # Using a custom instance
+        executor = described_class.new(
+          min_threads: 1,
+          max_threads: 1,
+          max_queue: 1,
+          fallback_policy: :caller_runs)
+        
+        worker_unblocker = Concurrent::CountDownLatch.new(1)
+        executor_unblocker = Concurrent::CountDownLatch.new(1)
+        queue_done = Concurrent::CountDownLatch.new(1)
+        
+        # Block the worker thread
+        executor << proc { worker_unblocker.wait }
+        
+        # Fill the queue
+        executor << proc { log.push :queued; queue_done.count_down }
+        
+        # Block in a caller_runs job
+        caller_runs_thread = Thread.new {
+          executor << proc { executor_unblocker.wait; log.push :unblocked }
+        }
+        
+        # Wait until the caller_runs job is blocked
+        Thread.pass until caller_runs_thread.status == 'sleep'
+        
+        # Now unblock the worker thread
+        worker_unblocker.count_down
+        queue_done.wait
+        executor_unblocker.count_down
+        
+        # Tidy up
+        caller_runs_thread.join
+        
+        # We will see the queued jobs run before the caller_runs job unblocks
+        expect([log.pop, log.pop]).to eq [:queued, :unblocked]
+      end
     end
   end
 end