Allow TimerSet to safely handle an executor raising RejectedExecutionError (#999)

Author: bensheldon

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

@@ -3,7 +3,7 @@
 require 'concurrent/collection/non_concurrent_priority_queue'
 require 'concurrent/executor/executor_service'
 require 'concurrent/executor/single_thread_executor'
-
+require 'concurrent/errors'
 require 'concurrent/options'
 
 module Concurrent
@@ -162,7 +162,11 @@ def process_tasks
           # queue now must have the same pop time, or a closer one, as
           # when we peeked).
           task = synchronize { @queue.pop }
-          task.executor.post { task.process_task }
+          begin
+            task.executor.post { task.process_task }
+          rescue RejectedExecutionError
+            # ignore and continue
+          end
         else
           @condition.wait([diff, 60].min)
         end

spec/concurrent/executor/timer_set_spec.rb

@@ -122,6 +122,20 @@ module Concurrent
           expect(task.value).to eq i
         end
       end
+
+      it 'safely handles an executor raising RejectedExecutionError' do
+        # force a task's executor to raise RejectedExecutionError within the TimerSet
+        abort_executor = ImmediateExecutor.new
+        allow(abort_executor).to receive(:post).and_raise(Concurrent::RejectedExecutionError)
+        ScheduledTask.execute(0.2, executor: abort_executor, timer_set: subject){ nil }
+        abort_executor.shutdown
+
+        latch = CountDownLatch.new(1)
+        ScheduledTask.execute(0.3, timer_set: subject) do
+          latch.count_down
+        end
+        expect(latch.wait(1)).to be_truthy
+      end
     end
 
     context 'resolution' do