Merge pull request #174 from pitr-ch/pools

Fix leaking threads when pruning the pool

Author: Petr Chalupa

lib/concurrent/executor/ruby_thread_pool_executor.rb

@@ -11,20 +11,20 @@ class RubyThreadPoolExecutor
     include RubyExecutor
 
     # Default maximum number of threads that will be created in the pool.
-    DEFAULT_MAX_POOL_SIZE = 2**15 # 32768
+    DEFAULT_MAX_POOL_SIZE      = 2**15 # 32768
 
     # Default minimum number of threads that will be retained in the pool.
-    DEFAULT_MIN_POOL_SIZE = 0
+    DEFAULT_MIN_POOL_SIZE      = 0
 
     # Default maximum number of tasks that may be added to the task queue.
-    DEFAULT_MAX_QUEUE_SIZE = 0
+    DEFAULT_MAX_QUEUE_SIZE     = 0
 
     # Default maximum number of seconds a thread in the pool may remain idle
     # before being reclaimed.
     DEFAULT_THREAD_IDLETIMEOUT = 60
 
     # The set of possible overflow policies that may be set at thread pool creation.
-    OVERFLOW_POLICIES = [:abort, :discard, :caller_runs]
+    OVERFLOW_POLICIES          = [:abort, :discard, :caller_runs]
 
     # The maximum number of threads that may be created in the pool.
     attr_reader :max_length
@@ -77,10 +77,10 @@ class RubyThreadPoolExecutor
     #
     # @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ThreadPoolExecutor.html
     def initialize(opts = {})
-      @min_length = opts.fetch(:min_threads, DEFAULT_MIN_POOL_SIZE).to_i
-      @max_length = opts.fetch(:max_threads, DEFAULT_MAX_POOL_SIZE).to_i
-      @idletime = opts.fetch(:idletime, DEFAULT_THREAD_IDLETIMEOUT).to_i
-      @max_queue = opts.fetch(:max_queue, DEFAULT_MAX_QUEUE_SIZE).to_i
+      @min_length      = opts.fetch(:min_threads, DEFAULT_MIN_POOL_SIZE).to_i
+      @max_length      = opts.fetch(:max_threads, DEFAULT_MAX_POOL_SIZE).to_i
+      @idletime        = opts.fetch(:idletime, DEFAULT_THREAD_IDLETIMEOUT).to_i
+      @max_queue       = opts.fetch(:max_queue, DEFAULT_MAX_QUEUE_SIZE).to_i
       @overflow_policy = opts.fetch(:overflow_policy, :abort)
 
       raise ArgumentError.new('max_threads must be greater than zero') if @max_length <= 0
@@ -90,13 +90,13 @@ def initialize(opts = {})
 
       init_executor
 
-      @pool = []
-      @queue = Queue.new
+      @pool                 = []
+      @queue                = Queue.new
       @scheduled_task_count = 0
       @completed_task_count = 0
-      @largest_length = 0
+      @largest_length       = 0
 
-      @gc_interval = opts.fetch(:gc_interval, 1).to_i # undocumented
+      @gc_interval  = opts.fetch(:gc_interval, 1).to_i # undocumented
       @last_gc_time = Time.now.to_f - [1.0, (@gc_interval * 2.0)].max
     end
 
@@ -109,15 +109,16 @@ def can_overflow?
     #
     # @return [Integer] the length
     def length
-      mutex.synchronize{ running? ? @pool.length : 0 }
+      mutex.synchronize { running? ? @pool.length : 0 }
     end
+
     alias_method :current_length, :length
 
     # The number of tasks in the queue awaiting execution.
     #
     # @return [Integer] the queue_length
     def queue_length
-      mutex.synchronize{ running? ? @queue.length : 0 }
+      mutex.synchronize { running? ? @queue.length : 0 }
     end
 
     # Number of tasks that may be enqueued before reaching `max_queue` and rejecting
@@ -152,7 +153,7 @@ def on_end_task
     def on_worker_exit(worker)
       mutex.synchronize do
         @pool.delete(worker)
-        if @pool.empty? && ! running?
+        if @pool.empty? && !running?
           stop_event.set
           stopped_event.set
         end
@@ -177,7 +178,7 @@ def shutdown_execution
       if @pool.empty?
         stopped_event.set
       else
-        @pool.length.times{ @queue << :stop }
+        @pool.length.times { @queue << :stop }
       end
     end
 
@@ -196,7 +197,7 @@ def kill_execution
     # @!visibility private
     def ensure_capacity?
       additional = 0
-      capacity = true
+      capacity   = true
 
       if @pool.size < @min_length
         additional = @min_length - @pool.size
@@ -254,10 +255,11 @@ def handle_overflow(*args)
     # @!visibility private
     def prune_pool
       if Time.now.to_f - @gc_interval >= @last_gc_time
-        @pool.delete_if do |worker|
-          worker.dead? ||
-            (@idletime == 0 ? false : Time.now.to_f - @idletime > worker.last_activity)
-        end
+        @pool.delete_if { |worker| worker.dead? }
+        # send :stop for each thread over idletime
+        @pool.
+            select { |worker| @idletime != 0 && Time.now.to_f - @idletime > worker.last_activity }.
+            each { @queue << :stop }
         @last_gc_time = Time.now.to_f
       end
     end
@@ -266,7 +268,7 @@ def prune_pool
     #
     # @!visibility private
     def drain_pool
-      @pool.each {|worker| worker.kill }
+      @pool.each { |worker| worker.kill }
       @pool.clear
     end
 

spec/concurrent/executor/ruby_cached_thread_pool_spec.rb

@@ -24,17 +24,19 @@ module Concurrent
       subject{ described_class.new(idletime: 1, max_threads: 2, gc_interval: 0) }
 
       it 'removes from pool any thread that has been idle too long' do
-        subject.instance_variable_set(:@idletime, 1)
         latch = Concurrent::CountDownLatch.new(3)
-        3.times { subject << proc{ sleep(0.1); latch.count_down } }
+        4.times { subject << proc { sleep 0.1; latch.count_down } }
         expect(latch.wait(1)).to be true
 
+        subject.instance_variable_set(:@idletime, 1)
+
+        sleep 1.5
+
         max_threads = subject.length
-        sleep(2)
 
-        latch = Concurrent::CountDownLatch.new(1)
-        subject << proc{ latch.count_down }
-        expect(latch.wait(1)).to be true
+        subject.send :prune_pool
+        sleep 0.1
+        subject.send :prune_pool
 
         expect(subject.length).to be < max_threads
       end