Merge pull request #938 from ruby-concurrency/prune

Expose the prune operation from thread pools

Author: chrisseaton

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

@@ -71,9 +71,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
   #
@@ -111,6 +118,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-ruby/concurrent/executor/java_thread_pool_executor.rb

@@ -93,6 +93,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-ruby/concurrent/executor/ruby_thread_pool_executor.rb

@@ -93,13 +93,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
@@ -112,6 +107,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
@@ -192,7 +192,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
@@ -223,7 +223,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
@@ -242,29 +242,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
@@ -278,10 +270,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
@@ -330,29 +329,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 }