make the pruning of thread pools a public API

This makes the pruning of thread pools a public API, which may
or may not be a no-op, depending on the underlying thread pool
implementation.

It also reworks the Ruby implementation to be more aggressive in
the pruning - it will kill all to-be-pruned threads in one call
instead of doing them one by one. Moreover, workers do not get
woken up to check for their idleness anymore, the thread of the
caller does it instead.

Author: doudou

lib/concurrent/executor/fixed_thread_pool.rb

@@ -64,9 +64,16 @@ module Concurrent
   #   @return [Integer] Number of tasks that may be enqueued before reaching `max_queue` and rejecting
   #     new tasks. A value of -1 indicates that the queue may grow without bound.
 
-
-
-
+  # @!macro thread_pool_executor_method_prune_pool
+  #   Prune the thread pool of unneeded threads
+  #
+  #   What is being pruned is controlled by the min_threads and idletime
+  #   parameters passed at pool creation time
+  #
+  #   This is a no-op on some pool implementation (e.g. the Java one).  The Ruby
+  #   pool will auto-prune each time a new job is posted. You will need to call
+  #   this method explicitely in case your application post jobs in bursts (a
+  #   lot of jobs and then nothing for long periods)
 
   # @!macro thread_pool_executor_public_api
   #
@@ -104,6 +111,9 @@ module Concurrent
   #
   #   @!method can_overflow?
   #     @!macro executor_service_method_can_overflow_question
+  #
+  #   @!method prune_pool
+  #     @!macro thread_pool_executor_method_prune_pool
 
 
 

lib/concurrent/executor/java_thread_pool_executor.rb

@@ -87,6 +87,10 @@ def running?
         super && !@executor.isTerminating
       end
 
+      # @!macro thread_pool_executor_method_prune_pool
+      def prune_pool
+      end
+
       private
 
       def ns_initialize(opts)

lib/concurrent/executor/ruby_thread_pool_executor.rb

@@ -87,13 +87,8 @@ def remove_busy_worker(worker)
     end
 
     # @!visibility private
-    def ready_worker(worker)
-      synchronize { ns_ready_worker worker }
-    end
-
-    # @!visibility private
-    def worker_not_old_enough(worker)
-      synchronize { ns_worker_not_old_enough worker }
+    def ready_worker(worker, last_message)
+      synchronize { ns_ready_worker worker, last_message }
     end
 
     # @!visibility private
@@ -106,6 +101,11 @@ def worker_task_completed
       synchronize { @completed_task_count += 1 }
     end
 
+    # @!macro thread_pool_executor_method_prune_pool
+    def prune_pool
+      synchronize { ns_prune_pool }
+    end
+
     private
 
     # @!visibility private
@@ -185,7 +185,7 @@ def ns_kill_execution
     # @!visibility private
     def ns_assign_worker(*args, &task)
       # keep growing if the pool is not at the minimum yet
-      worker = (@ready.pop if @pool.size >= @min_length) || ns_add_busy_worker
+      worker, _ = (@ready.pop if @pool.size >= @min_length) || ns_add_busy_worker
       if worker
         worker << [task, args]
         true
@@ -214,7 +214,7 @@ def ns_enqueue(*args, &task)
     def ns_worker_died(worker)
       ns_remove_busy_worker worker
       replacement_worker = ns_add_busy_worker
-      ns_ready_worker replacement_worker, false if replacement_worker
+      ns_ready_worker replacement_worker, Concurrent.monotonic_time, false if replacement_worker
     end
 
     # creates new worker which has to receive work to do after it's added
@@ -232,29 +232,21 @@ def ns_add_busy_worker
     # handle ready worker, giving it new job or assigning back to @ready
     #
     # @!visibility private
-    def ns_ready_worker(worker, success = true)
+    def ns_ready_worker(worker, last_message, success = true)
       task_and_args = @queue.shift
       if task_and_args
         worker << task_and_args
       else
         # stop workers when !running?, do not return them to @ready
         if running?
-          @ready.push(worker)
+          raise unless last_message
+          @ready.push([worker, last_message])
         else
           worker.stop
         end
       end
     end
 
-    # returns back worker to @ready which was not idle for enough time
-    #
-    # @!visibility private
-    def ns_worker_not_old_enough(worker)
-      # let's put workers coming from idle_test back to the start (as the oldest worker)
-      @ready.unshift(worker)
-      true
-    end
-
     # removes a worker which is not in not tracked in @ready
     #
     # @!visibility private
@@ -268,10 +260,17 @@ def ns_remove_busy_worker(worker)
     #
     # @!visibility private
     def ns_prune_pool
-      return if @pool.size <= @min_length
-
-      last_used = @ready.shift
-      last_used << :idle_test if last_used
+      now = Concurrent.monotonic_time
+      stopped_workers = 0
+      while !@ready.empty? && (@pool.size - stopped_workers > @min_length)
+        worker, last_message = @ready.first
+        if now - last_message > self.idletime
+          stopped_workers += 1
+          @ready.shift
+          worker << :stop
+        else break
+        end
+      end
 
       @next_gc_time = Concurrent.monotonic_time + @gc_interval
     end
@@ -315,29 +314,18 @@ def kill
 
       def create_worker(queue, pool, idletime)
         Thread.new(queue, pool, idletime) do |my_queue, my_pool, my_idletime|
-          last_message = Concurrent.monotonic_time
           catch(:stop) do
             loop do
 
               case message = my_queue.pop
-              when :idle_test
-                if (Concurrent.monotonic_time - last_message) > my_idletime
-                  my_pool.remove_busy_worker(self)
-                  throw :stop
-                else
-                  my_pool.worker_not_old_enough(self)
-                end
-
               when :stop
                 my_pool.remove_busy_worker(self)
                 throw :stop
 
               else
                 task, args = message
                 run_task my_pool, task, args
-                last_message = Concurrent.monotonic_time
-
-                my_pool.ready_worker(self)
+                my_pool.ready_worker(self, Concurrent.monotonic_time)
               end
             end
           end

spec/concurrent/executor/java_thread_pool_executor_spec.rb

@@ -25,6 +25,13 @@ module Concurrent
 
       it_should_behave_like :thread_pool_executor
 
+      context :prune do
+        it "is a no-op, pruning is handled by the JVM" do
+          executor = JavaThreadPoolExecutor.new
+          executor.prune_pool
+        end
+      end
+
       context '#overload_policy' do
 
         specify ':abort maps to AbortPolicy' do

spec/concurrent/executor/ruby_thread_pool_executor_spec.rb

@@ -23,6 +23,109 @@ module Concurrent
 
     it_should_behave_like :thread_pool_executor
 
+
+    context :prune do
+      subject do
+        RubyThreadPoolExecutor.new(idletime: 5, min_threads: 2, max_threads: 10)
+      end
+
+      Group = Struct.new :waiting_threads, :threads, :mutex, :cond
+
+      def prepare_thread_group(size)
+        cond = ConditionVariable.new
+        mutex = Mutex.new
+        threads = []
+        size.times do
+          subject.post do
+            mutex.synchronize do
+              threads << Thread.current
+              cond.wait(mutex)
+              threads.delete(Thread.current)
+            end
+          end
+        end
+        eventually(mutex: mutex) { expect(threads).to have_attributes(size: size) }
+        Group.new(threads, threads.dup, mutex, cond)
+      end
+
+      def wakeup_thread_group(group)
+        group.cond.broadcast
+        eventually(mutex: group.mutex) do
+          expect(group.waiting_threads).to have_attributes(size: 0)
+        end
+      end
+
+      before(:each) do
+        @now = Concurrent.monotonic_time
+        allow(Concurrent).to receive(:monotonic_time) { @now }
+
+        @group1 = prepare_thread_group(5)
+        @group2 = prepare_thread_group(5)
+      end
+
+      def eventually(mutex: nil, timeout: 5, &block)
+          start = Time.now
+          while Time.now - start < timeout
+            begin
+              if mutex
+                mutex.synchronize do
+                  return yield
+                end
+              else
+                return yield
+              end
+            rescue Exception => last_failure
+            end
+            Thread.pass
+          end
+          raise last_failure
+      end
+
+      it "triggers pruning when posting work if the last prune happened more than gc_interval ago" do
+        wakeup_thread_group(@group1)
+        @now += 6
+        wakeup_thread_group(@group2)
+        subject.post { }
+
+        eventually { expect(@group1.threads).to all(have_attributes(status: false)) }
+        eventually { expect(@group2.threads).to all(have_attributes(status: 'sleep')) }
+      end
+
+      it "does not trigger pruning when posting work if the last prune happened less than gc_interval ago" do
+        wakeup_thread_group(@group1)
+        @now += 3
+        subject.prune_pool
+        @now += 3
+        wakeup_thread_group(@group2)
+        subject.post { }
+
+        eventually { expect(@group1.threads).to all(have_attributes(status: false)) }
+        eventually { expect(@group2.threads).to all(have_attributes(status: 'sleep')) }
+      end
+
+      it "reclaims threads that have been idle for more than idletime seconds" do
+        wakeup_thread_group(@group1)
+        @now += 6
+        wakeup_thread_group(@group2)
+        subject.prune_pool
+
+        eventually { expect(@group1.threads).to all(have_attributes(status: false)) }
+        eventually { expect(@group2.threads).to all(have_attributes(status: 'sleep')) }
+      end
+
+      it "keeps at least min_length workers" do
+        wakeup_thread_group(@group1)
+        wakeup_thread_group(@group2)
+        @now += 12
+        subject.prune_pool
+        all_threads = @group1.threads + @group2.threads
+        eventually do
+          finished_threads = all_threads.find_all { |t| !t.status }
+          expect(finished_threads).to have_attributes(size: 8)
+        end
+      end
+    end
+
     context '#remaining_capacity' do
 
       let!(:expected_max){ 100 }