More efficient locking in AbstractThreadPool.

lib/concurrent/abstract_thread_pool.rb

@@ -48,6 +48,19 @@ def wait_for_termination(timeout = nil)
       return @terminator.wait(timeout)
     end
 
+    def post(*args, &block)
+      raise ArgumentError.new('no block given') unless block_given?
+      return @mutex.synchronize do
+        if @state == :running
+          at_post
+          @queue << [args, block]
+          true
+        else
+          false
+        end
+      end
+    end
+
     def <<(block)
       self.post(&block)
       return self

lib/concurrent/abstract_thread_pool/worker.rb

@@ -7,7 +7,6 @@ class AbstractThreadPool
     protected
 
     class Worker
-      @@mutex = Mutex.new
 
       def initialize(queue)
         @queue = queue
@@ -16,12 +15,12 @@ def initialize(queue)
       end
 
       def idle?
-        return ! @idletime
+        return ! @idletime.nil?
       end
 
       def idletime
         return @mutex.synchronize do
-          @idletime.nil? ? 0 : Time.now.to_i = @idletime.to_i
+          @idletime.nil? ? 0 : Time.now.to_i - @idletime.to_i
         end
       end
 
@@ -56,15 +55,34 @@ def run(thread = Thread.current)
           end
 
           @idletime = nil
+          Worker.busy
           begin
             task.last.call(*task.first)
           rescue
             # let it fail
           ensure
+            Worker.free
             @idletime = Time.now
           end
         end
       end
+
+      class << self
+        attr_reader :working
+
+        def mutex
+          @working ||= 0
+          @mutex ||= Mutex.new
+        end
+
+        def busy
+          mutex.synchronize { @working += 1 }
+        end
+
+        def free
+          mutex.synchronize { @working -= 1 }
+        end
+      end
     end
   end
 end

lib/concurrent/cached_thread_pool.rb

@@ -11,40 +11,32 @@ def initialize(opts = {})
       super(opts)
     end
 
-    def post(*args, &block)
-      raise ArgumentError.new('no block given') unless block_given?
-      return @mutex.synchronize do
-        if @state == :running
-          @queue << [args, block]
-          at_capacity = @pool.empty? || ! @queue.empty? || @working >= @pool.size
-          if at_capacity && @pool.length < @max_threads
-            @pool << create_worker_thread
-          end
-          true
-        else
-          false
-        end
+    protected
+
+    def at_post
+      at_capacity = @pool.empty? || ! @queue.empty? || Worker.working >= @pool.size
+      if at_capacity && @pool.length < @max_threads
+        create_worker_thread
       end
     end
 
-    protected
-
-    def dead_worker?(context)
-      return context.thread.nil? || context.thread.status == 'aborting' || ! context.thread.status
+    def dead_worker?(worker)
+      thread_status = worker.status.last
+      return ! thread_status || thread_status == 'aborting'
     end
 
-    def stale_worker?(context)
-      if context.status == :idle && @idletime <= (timestamp - context.idletime)
-        context.thread.kill
+    def stale_worker?(worker)
+      if worker.idle? && worker.idletime >= @idletime
+        worker.kill
         return true
       else
         return false
       end
     end
 
     def collect_garbage
-      @pool.reject! do |context|
-        dead_worker?(context) || stale_worker?(context)
+      @pool.reject! do |worker|
+        dead_worker?(worker) || stale_worker?(worker)
       end
     end
   end

lib/concurrent/fixed_thread_pool.rb

@@ -8,25 +8,16 @@ def initialize(size, opts = {})
       super(opts.merge(max_threads: size))
     end
 
-    def post(*args, &block)
-      raise ArgumentError.new('no block given') unless block_given?
-      return @mutex.synchronize do
-        if @state == :running
-          while @pool.size < @max_threads
-            create_worker_thread
-          end
-          @queue << [args, block]
-          true
-        else
-          false
-        end
+    protected
+
+    def at_post
+      while @pool.size < @max_threads
+        create_worker_thread
       end
     end
 
-    protected
-
     def collect_garbage
-      @pool.reject! {|context| ! context.status }
+      @pool.reject! {|context| ! context.status.last }
     end
   end
 end

spec/concurrent/cached_thread_pool_spec.rb

@@ -79,19 +79,19 @@ module Concurrent
 
     context 'garbage collection' do
 
-      subject{ CachedThreadPool.new(gc_interval: 1, idletime: 0.1) }
+      subject{ CachedThreadPool.new(gc_interval: 1, idletime: 1) }
 
       it 'removes from pool any thread that has been idle too long' do
         subject << proc{ nil }
         subject.size.should eq 1
-        sleep(1.5)
+        sleep(2)
         subject.size.should eq 0
       end
 
       it 'removed from pool any dead thread' do
-        subject << proc{ raise StandardError }
+        subject << proc{ raise Exception }
         subject.size.should eq 1
-        sleep(1.5)
+        sleep(2)
         subject.size.should eq 0
       end
     end

spec/spec_helper.rb

@@ -23,7 +23,7 @@
 Dir[File.join(File.dirname(__FILE__), 'support/**/*.rb')].each { |f| require File.expand_path(f) }
 
 RSpec.configure do |config|
-  #config.order = 'random'
+  config.order = 'random'
 
   config.before(:suite) do
   end