Merge pull request #115 from billdueber/pool_spec_fixes

Pool spec fixes

Author: jdantonio

lib/concurrent/executor/java_fixed_thread_pool.rb

@@ -17,14 +17,22 @@ class JavaFixedThreadPool < JavaThreadPoolExecutor
       #
       # @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newFixedThreadPool-int-
       def initialize(num_threads, opts = {})
-        @overflow_policy = opts.fetch(:overflow_policy, :abort)
-        @max_queue = 0
 
-        raise ArgumentError.new('number of threads must be greater than zero') if num_threads < 1
-        raise ArgumentError.new("#{@overflow_policy} is not a valid overflow policy") unless OVERFLOW_POLICIES.keys.include?(@overflow_policy)
+        opts = {
+            min_threads: num_threads,
+            max_threads: num_threads
+        }.merge(opts)
+        super(opts)
 
-        @executor = java.util.concurrent.Executors.newFixedThreadPool(num_threads)
-        @executor.setRejectedExecutionHandler(OVERFLOW_POLICIES[@overflow_policy].new)
+
+        #@overflow_policy = opts.fetch(:overflow_policy, :abort)
+        #@max_queue = 0
+        #
+        #raise ArgumentError.new('number of threads must be greater than zero') if num_threads < 1
+        #raise ArgumentError.new("#{@overflow_policy} is not a valid overflow policy") unless OVERFLOW_POLICIES.keys.include?(@overflow_policy)
+        #
+        #@executor = java.util.concurrent.Executors.newFixedThreadPool(num_threads)
+        #@executor.setRejectedExecutionHandler(OVERFLOW_POLICIES[@overflow_policy].new)
 
         set_shutdown_hook
       end

lib/concurrent/executor/java_thread_pool_executor.rb

@@ -71,6 +71,7 @@ def initialize(opts = {})
 
         raise ArgumentError.new('max_threads must be greater than zero') if max_length <= 0
         raise ArgumentError.new('min_threads cannot be less than zero') if min_length < 0
+        raise ArgumentError.new('min_threads cannot be more than max_threads') if min_length > max_length
         raise ArgumentError.new("#{@overflow_policy} is not a valid overflow policy") unless OVERFLOW_POLICIES.keys.include?(@overflow_policy)
 
         if min_length == 0 && @max_queue == 0
@@ -171,16 +172,7 @@ def status
       #
       # @return [Boolean] `true` when running, `false` when shutting down or shutdown
       def running?
-        super && ! @executor.isTerminating
-      end
-
-      # Begin an orderly shutdown. Tasks already in the queue will be executed,
-      # but no new tasks will be accepted. Has no additional effect if the
-      # thread pool is not running.
-      def shutdown
-        super
-        @executor.getQueue.clear
-        nil
+        super && !@executor.isTerminating
       end
     end
   end

lib/concurrent/executor/ruby_fixed_thread_pool.rb

@@ -19,13 +19,13 @@ def initialize(num_threads, opts = {})
       raise ArgumentError.new('number of threads must be greater than zero') if num_threads < 1
       raise ArgumentError.new("#{overflow_policy} is not a valid overflow policy") unless OVERFLOW_POLICIES.include?(overflow_policy)
 
-      opts = opts.merge(
+      opts = {
         min_threads: num_threads,
         max_threads: num_threads,
-        num_threads: overflow_policy,
+        overflow_policy: overflow_policy,
         max_queue: DEFAULT_MAX_QUEUE_SIZE,
-        idletime: 0
-      )
+        idletime: DEFAULT_THREAD_IDLETIMEOUT,
+      }.merge(opts)
       super(opts)
     end
   end

lib/concurrent/executor/ruby_thread_pool_executor.rb

@@ -86,6 +86,7 @@ def initialize(opts = {})
       raise ArgumentError.new('max_threads must be greater than zero') if @max_length <= 0
       raise ArgumentError.new('min_threads cannot be less than zero') if @min_length < 0
       raise ArgumentError.new("#{overflow_policy} is not a valid overflow policy") unless OVERFLOW_POLICIES.include?(@overflow_policy)
+      raise ArgumentError.new('min_threads cannot be more than max_threads') if min_length > max_length
 
       init_executor
 

spec/concurrent/executor/fixed_thread_pool_shared.rb

@@ -1,5 +1,6 @@
 require 'spec_helper'
 require_relative 'thread_pool_shared'
+require 'thread'
 
 share_examples_for :fixed_thread_pool do
 
@@ -13,13 +14,64 @@
 
   it_should_behave_like :thread_pool
 
-  context '#initialize' do
+  context '#initialize default values' do
+
+    subject { described_class.new(5) }
+
+    it 'defaults :min_length correctly' do
+      subject.min_length.should eq 5
+    end
+
+    it 'defaults :max_length correctly' do
+      subject.max_length.should eq 5
+    end
+
+    it 'defaults :overflow_policy to :abort' do
+      subject.overflow_policy.should eq :abort
+    end
+
+
+    it 'defaults :idletime correctly' do
+      subject.idletime.should eq subject.class.const_get(:DEFAULT_THREAD_IDLETIMEOUT)
+    end
+
+    it 'defaults default :max_queue to zero' do
+      subject.max_queue.should eq 0
+    end
+
+  end
+
+  context '#initialize explicit values' do
 
     it 'raises an exception when the pool length is less than one' do
       lambda {
         described_class.new(0)
       }.should raise_error(ArgumentError)
     end
+
+
+    it 'sets explicit :max_queue correctly' do
+      subject = described_class.new(5, :max_queue => 10)
+      subject.max_queue.should eq 10
+    end
+
+    it 'correctly sets valid :overflow_policy' do
+      subject = described_class.new(5, :overflow_policy => :caller_runs)
+      subject.overflow_policy.should eq :caller_runs
+    end
+
+    it "correctly sets valid :idletime" do
+      subject = described_class.new(5, :idletime => 10)
+      subject.idletime.should eq 10
+    end
+
+    it 'raises an exception if given an invalid :overflow_policy' do
+      expect {
+        described_class.new(5, overflow_policy: :bogus)
+      }.to raise_error(ArgumentError)
+    end
+
+
   end
 
   context '#min_length' do
@@ -102,12 +154,6 @@
     end
   end
 
-  context '#idletime' do
-
-    it 'returns zero' do
-      subject.idletime.should eq 0
-    end
-  end
 
   context '#kill' do
 
@@ -133,4 +179,54 @@
       pool.kill
     end
   end
+
+  context 'overflow policy' do
+
+    before do
+      @queue = Queue.new
+    end
+
+    # On abort, it should raise an error
+    it "raises an error when overflow on abort" do
+      subject = described_class.new(2, :max_queue => 2, :overflow_policy => :abort)
+      expect {
+        5.times do |i|
+          subject.post { sleep 0.1; @queue << i}
+        end
+        subject.shutdown
+        subject.wait_for_termination(1)
+      }.to raise_error
+    end
+
+    # On discard, we'd expect no error, but also not all five results
+    it 'discards when overflow is :discard' do
+      subject = described_class.new(2, :max_queue => 2, :overflow_policy => :discard)
+      5.times do |i|
+        subject.post { sleep 0.1; @queue << i}
+      end
+      subject.shutdown
+      subject.wait_for_termination(1)
+      @queue.length.should be < 5
+    end
+
+    # To check for caller_runs, we'll check how many unique threads
+    # actually ran the block
+
+    it 'uses the calling thread for overflow under caller_runs' do
+      subject = described_class.new(2, :max_queue => 2, :overflow_policy => :caller_runs)
+      5.times do |i|
+        subject.post { sleep 0.1; @queue << Thread.current}
+      end
+      subject.shutdown
+      subject.wait_for_termination(1)
+      # Turn the queue into an array
+      a = []
+      a << @queue.shift until @queue.empty?
+
+      a.size.should eq 5 # one for each run of the block
+      a.uniq.size.should eq 3 # one for each of teh two threads, plus the caller
+    end
+  end
+
+
 end

spec/concurrent/executor/java_cached_thread_pool_spec.rb

@@ -6,7 +6,7 @@
 
   module Concurrent
 
-    describe JavaCachedThreadPool do
+    describe JavaCachedThreadPool, :type=>:jruby do
 
       subject { described_class.new(overflow_policy: :discard) }
 

spec/concurrent/executor/java_fixed_thread_pool_spec.rb

@@ -6,7 +6,7 @@
 
   module Concurrent
 
-    describe JavaFixedThreadPool do
+    describe JavaFixedThreadPool, :type=>:jruby do
 
       subject { described_class.new(5, overflow_policy: :discard) }
 
@@ -19,45 +19,16 @@ module Concurrent
 
       context '#initialize' do
 
-        it 'sets :min_length correctly' do
-          subject = JavaFixedThreadPool.new(10)
-          subject.min_length.should eq 10
-        end
-
-        it 'sets :max_length correctly' do
-          subject = JavaFixedThreadPool.new(5)
-          subject.max_length.should eq 5
-        end
-
-        it 'sets :idletime correctly' do
-          subject = JavaFixedThreadPool.new(5)
-          subject.idletime.should eq 0
-        end
-
-        it 'sets :max_queue correctly' do
-          subject = JavaFixedThreadPool.new(5)
-          subject.max_queue.should eq 0
-        end
 
         it 'sets :overflow_policy correctly' do
-          clazz = java.util.concurrent.ThreadPoolExecutor::DiscardPolicy
+          clazz  = java.util.concurrent.ThreadPoolExecutor::DiscardPolicy
           policy = clazz.new
           clazz.should_receive(:new).at_least(:once).with(any_args).and_return(policy)
 
           subject = JavaFixedThreadPool.new(5, overflow_policy: :discard)
           subject.overflow_policy.should eq :discard
         end
 
-        it 'defaults :overflow_policy to :abort' do
-          subject = JavaFixedThreadPool.new(5)
-          subject.overflow_policy.should eq :abort
-        end
-
-        it 'raises an exception if given an invalid :overflow_policy' do
-          expect {
-            JavaFixedThreadPool.new(5, overflow_policy: :bogus)
-          }.to raise_error(ArgumentError)
-        end
       end
     end
   end

spec/concurrent/executor/java_single_thread_executor_spec.rb

@@ -6,7 +6,7 @@
 
   module Concurrent
 
-    describe JavaSingleThreadExecutor do
+    describe JavaSingleThreadExecutor, :type=>:jruby do
 
       after(:each) do
         subject.kill

spec/concurrent/executor/java_thread_pool_executor_spec.rb

@@ -6,7 +6,7 @@
 
   module Concurrent
 
-    describe JavaThreadPoolExecutor do
+    describe JavaThreadPoolExecutor, :type => :jruby do
 
       after(:each) do
         subject.kill

spec/concurrent/executor/ruby_cached_thread_pool_spec.rb

@@ -3,7 +3,7 @@
 
 module Concurrent
 
-  describe RubyCachedThreadPool do
+  describe RubyCachedThreadPool, :type=>:mrirbx do
 
     subject do
       described_class.new(