PerThreadExecutor is now an executor service.

Author: jdantonio

lib/concurrent/executor/per_thread_executor.rb

@@ -1,24 +1,78 @@
+require 'concurrent/atomic/event'
+require 'concurrent/executor/executor'
+
 module Concurrent
 
   class PerThreadExecutor
-    include Executor
+    include SerialExecutor
+
+    def initialize
+      @running = Concurrent::AtomicBoolean.new(true)
+      @stopped = Concurrent::Event.new
+      @count = Concurrent::AtomicFixnum.new(0)
+    end
 
     def self.post(*args)
       raise ArgumentError.new('no block given') unless block_given?
       Thread.new(*args) do
         Thread.current.abort_on_exception = false
         yield(*args)
       end
-      return true
+      true
     end
 
     def post(*args, &task)
-      return PerThreadExecutor.post(*args, &task)
+      raise ArgumentError.new('no block given') unless block_given?
+      return false unless running?
+      @count.increment
+      Thread.new(*args) do
+        Thread.current.abort_on_exception = false
+        begin
+          yield(*args)
+        ensure
+          @count.decrement
+          @stopped.set if @running.false? && @count.value == 0
+        end
+      end
     end
 
     def <<(task)
-      PerThreadExecutor.post(&task)
-      return self
+      post(&task)
+      self
+    end
+
+    # @!macro executor_method_running_question
+    def running?
+      @running.true?
+    end
+
+    # @!macro executor_method_shuttingdown_question
+    def shuttingdown?
+      @running.false? && ! @stopped.set?
+    end
+
+    # @!macro executor_method_shutdown_question
+    def shutdown?
+      @stopped.set?
+    end
+
+    # @!macro executor_method_shutdown
+    def shutdown
+      @running.make_false
+      @stopped.set if @count.value == 0
+      true
+    end
+
+    # @!macro executor_method_kill
+    def kill
+      @running.make_false
+      @stopped.set
+      true
+    end
+
+    # @!macro executor_method_wait_for_termination
+    def wait_for_termination(timeout = nil)
+      @stopped.wait(timeout)
     end
   end
 end

spec/concurrent/executor/per_thread_executor_spec.rb

@@ -1,13 +1,13 @@
 require 'spec_helper'
-require_relative 'global_thread_pool_shared'
+require_relative 'thread_pool_shared'
 
 module Concurrent
 
   describe PerThreadExecutor do
 
     subject { PerThreadExecutor.new }
 
-    it_should_behave_like :global_thread_pool
+    it_should_behave_like :executor_service
 
     context '#post' do
 
@@ -19,10 +19,9 @@ module Concurrent
       end
 
       it 'executes a call without arguments' do
-        @expected = false
-        subject.post{ @expected = true }
-        sleep(0.1)
-        @expected.should be_true
+        latch = CountDownLatch.new(1)
+        subject.post{ latch.count_down }
+        latch.wait(1).should be_true
       end
 
       it 'creates a new thread for a call with arguments' do
@@ -33,17 +32,15 @@ module Concurrent
       end
 
       it 'executes a call with one argument' do
-        @expected = 0
-        subject.post(1){|one| @expected = one }
-        sleep(0.1)
-        @expected.should == 1
+        latch = CountDownLatch.new(3)
+        subject.post(3){|count| count.times{ latch.count_down } }
+        latch.wait(1).should be_true
       end
 
       it 'executes a call with multiple arguments' do
-        @expected = nil
-        subject.post(1,2,3,4,5){|*args| @expected = args }
-        sleep(0.1)
-        @expected.should eq [1,2,3,4,5]
+        latch = CountDownLatch.new(10)
+        subject.post(1,2,3,4){|*count| count.reduce(:+).times{ latch.count_down } }
+        latch.wait(1).should be_true
       end
 
       it 'aliases #<<' do