Ensure all tested thread pools terminate

Author: pitr-ch

spec/concurrent/agent_spec.rb

@@ -860,6 +860,9 @@ def update(time, old_value, new_value)
 
           subject.await_for(5)
           expect(bucket).to eq [1, 2, 3]
+
+          executor.kill
+          expect(executor.wait_for_termination(1)).to eq true
         end
       end
 

spec/concurrent/executor/cached_thread_pool_spec.rb

@@ -9,8 +9,8 @@ module Concurrent
     end
 
     after(:each) do
-      subject.kill
-      subject.wait_for_termination(1)
+      subject.shutdown
+      expect(subject.wait_for_termination(1)).to eq true
     end
 
     it_should_behave_like :thread_pool
@@ -19,20 +19,22 @@ module Concurrent
 
     context '#initialize' do
 
+      subject { described_class.new }
+
       it 'sets :max_length to DEFAULT_MAX_POOL_SIZE' do
-        expect(described_class.new.max_length).to eq described_class::DEFAULT_MAX_POOL_SIZE
+        expect(subject.max_length).to eq described_class::DEFAULT_MAX_POOL_SIZE
       end
 
       it 'sets :min_length to DEFAULT_MIN_POOL_SIZE' do
-        subject = expect(described_class.new.min_length).to eq described_class::DEFAULT_MIN_POOL_SIZE
+        expect(subject.min_length).to eq described_class::DEFAULT_MIN_POOL_SIZE
       end
 
       it 'sets :idletime to DEFAULT_THREAD_IDLETIMEOUT' do
-        subject = expect(described_class.new.idletime).to eq described_class::DEFAULT_THREAD_IDLETIMEOUT
+        expect(subject.idletime).to eq described_class::DEFAULT_THREAD_IDLETIMEOUT
       end
 
       it 'sets :max_queue to DEFAULT_MAX_QUEUE_SIZE' do
-        subject = expect(described_class.new.max_queue).to eq described_class::DEFAULT_MAX_QUEUE_SIZE
+        expect(subject.max_queue).to eq described_class::DEFAULT_MAX_QUEUE_SIZE
       end
     end
 
@@ -54,7 +56,7 @@ module Concurrent
         subject.post { latch.count_down }
         latch.wait(0.1)
         subject.shutdown
-        subject.wait_for_termination(1)
+        expect(subject.wait_for_termination(1)).to eq true
         expect(subject.min_length).to eq 0
       end
     end
@@ -77,7 +79,7 @@ module Concurrent
         subject.post { latch.count_down }
         latch.wait(0.1)
         subject.shutdown
-        subject.wait_for_termination(1)
+        expect(subject.wait_for_termination(1)).to eq true
         expect(subject.max_length).to eq described_class::DEFAULT_MAX_POOL_SIZE
       end
     end
@@ -100,7 +102,7 @@ module Concurrent
         subject.post { latch.count_down }
         latch.wait(0.1)
         subject.shutdown
-        subject.wait_for_termination(1)
+        expect(subject.wait_for_termination(1)).to eq true
         expect(subject.largest_length).to be > 0
       end
     end
@@ -127,11 +129,15 @@ module Concurrent
 
             subject = CachedThreadPool.new(fallback_policy: :discard)
             expect(subject.fallback_policy).to eq :discard
+            subject.shutdown
+            expect(subject.wait_for_termination(1)).to eq true
           end
 
           it 'defaults :fallback_policy to :abort' do
             subject = CachedThreadPool.new
             expect(subject.fallback_policy).to eq :abort
+            subject.shutdown
+            expect(subject.wait_for_termination(1)).to eq true
           end
 
           it 'raises an exception if given an invalid :fallback_policy' do

spec/concurrent/executor/executor_service_shared.rb

@@ -3,8 +3,8 @@
 RSpec.shared_examples :executor_service do
 
   after(:each) do
-    subject.kill
-    subject.wait_for_termination(0.1)
+    subject.shutdown
+    expect(subject.wait_for_termination(1)).to eq true
   end
 
   it_should_behave_like :global_thread_pool
@@ -40,21 +40,21 @@
     end
 
     it 'returns false when the thread pool is shutting down' do
-      subject.post{ sleep(1) }
+      subject.post{ sleep(0.5) }
       subject.shutdown
-      subject.wait_for_termination(1)
       expect(subject).not_to be_running
+      expect(subject.wait_for_termination(1)).to eq true
     end
 
     it 'returns false when the thread pool is shutdown' do
       subject.shutdown
-      subject.wait_for_termination(1)
+      expect(subject.wait_for_termination(1)).to eq true
       expect(subject).not_to be_running
     end
 
     it 'returns false when the thread pool is killed' do
       subject.kill
-      subject.wait_for_termination(1)
+      expect(subject.wait_for_termination(1)).to eq true
       expect(subject).not_to be_running
     end
   end
@@ -67,7 +67,7 @@
       subject.post{ sleep(0.1); latch1.count_down }
       latch1.wait(1)
       subject.shutdown
-      subject.wait_for_termination
+      expect(subject.wait_for_termination(1)).to eq true
       begin
         subject.post{ latch2.count_down }
       rescue Concurrent::RejectedExecutionError
@@ -97,7 +97,7 @@
       latch = Concurrent::CountDownLatch.new(1)
       subject.post{ sleep(0.1); latch.count_down }
       subject.shutdown
-      subject.wait_for_termination(1)
+      expect(subject.wait_for_termination(1)).to eq true
       expect(latch.wait(1)).to be_truthy
     end
 
@@ -107,7 +107,7 @@
         subject.post { sleep 0.1; q << i }
       end
       subject.shutdown
-      subject.wait_for_termination(1)
+      expect(subject.wait_for_termination(1)).to eq true
       expect(q.length).to eq 5
     end
 
@@ -120,7 +120,7 @@
         subject.post{ expected.increment }
       rescue Concurrent::RejectedExecutionError
       end
-      subject.wait_for_termination(1)
+      expect(subject.wait_for_termination(1)).to eq true
       expect(expected.value).to eq(2)
     end
   end
@@ -176,10 +176,12 @@
 
     it 'returns false when shutdown fails to complete before timeout' do
       unless subject.serialized?
-        100.times{ subject.post{ sleep(1) } }
+        latch = Concurrent::CountDownLatch.new 1
+        100.times{ subject.post{ latch.wait } }
         sleep(0.1)
         subject.shutdown
-        expect(subject.wait_for_termination(0)).to be_falsey
+        expect(subject.wait_for_termination(0.01)).to be_falsey
+        latch.count_down
       end
     end
 

spec/concurrent/executor/fixed_thread_pool_spec.rb

@@ -8,8 +8,8 @@ module Concurrent
     subject { described_class.new(num_threads) }
 
     after(:each) do
-      subject.kill
-      subject.wait_for_termination(0.1)
+      subject.shutdown
+      expect(subject.wait_for_termination(1)).to eq true
     end
 
     it_should_behave_like :thread_pool
@@ -55,25 +55,29 @@ module Concurrent
       it 'sets explicit :max_queue correctly' do
         subject = described_class.new(5, :max_queue => 10)
         expect(subject.max_queue).to eq 10
+        subject.shutdown
+        expect(subject.wait_for_termination(1)).to eq true
       end
 
       it 'correctly sets valid :fallback_policy' do
         subject = described_class.new(5, :fallback_policy => :caller_runs)
         expect(subject.fallback_policy).to eq :caller_runs
+        subject.shutdown
+        expect(subject.wait_for_termination(1)).to eq true
       end
 
       it "correctly sets valid :idletime" do
         subject = described_class.new(5, :idletime => 10)
         expect(subject.idletime).to eq 10
+        subject.shutdown
+        expect(subject.wait_for_termination(1)).to eq true
       end
 
       it 'raises an exception if given an invalid :fallback_policy' do
         expect {
           described_class.new(5, fallback_policy: :bogus)
         }.to raise_error(ArgumentError)
       end
-
-
     end
 
     context '#min_length' do
@@ -94,7 +98,7 @@ module Concurrent
         subject.post { latch.count_down }
         latch.wait(0.1)
         subject.shutdown
-        subject.wait_for_termination(1)
+        expect(subject.wait_for_termination(1)).to eq true
         expect(subject.min_length).to eq num_threads
       end
     end
@@ -117,7 +121,7 @@ module Concurrent
         subject.post { latch.count_down }
         latch.wait(0.1)
         subject.shutdown
-        subject.wait_for_termination(1)
+        expect(subject.wait_for_termination(1)).to eq true
         expect(subject.max_length).to eq num_threads
       end
     end
@@ -150,7 +154,7 @@ module Concurrent
         subject.post { latch.count_down }
         latch.wait(0.1)
         subject.shutdown
-        subject.wait_for_termination(1)
+        expect(subject.wait_for_termination(1)).to eq true
         expect(subject.largest_length).to eq num_threads
       end
     end
@@ -176,10 +180,13 @@ module Concurrent
 
       it 'never creates more than :num_threads threads' do
         pool = described_class.new(5)
-        100.times{ pool << proc{ sleep(1) } }
+        latch = Concurrent::CountDownLatch.new 1
+        100.times{ pool << proc{ latch.wait } }
         sleep(0.1)
         expect(pool.length).to eq 5
-        pool.kill
+        latch.count_down
+        pool.shutdown
+        expect(pool.wait_for_termination(1)).to eq true
       end
     end
 
@@ -190,7 +197,7 @@ module Concurrent
       end
 
       after(:each) do
-        subject.kill
+        subject.shutdown
       end
 
       # On abort, it should raise an error
@@ -209,6 +216,8 @@ module Concurrent
           end
           latch.wait(1)
         }.to raise_error(RejectedExecutionError)
+        subject.shutdown
+        expect(subject.wait_for_termination(1)).to eq true
       end
 
       # On discard, we'd expect no error, but also not all five results
@@ -227,6 +236,8 @@ module Concurrent
         latch.wait(1)
 
         expect(@queue.length).to be < 5
+        subject.shutdown
+        expect(subject.wait_for_termination(1)).to eq true
       end
 
       # To check for caller_runs, we'll check how many unique threads
@@ -254,6 +265,9 @@ module Concurrent
         #NOTE: This test is very, very difficult to setup properly. Hence the 'be_within' matcher
         expect(a.size).to be_within(1).of(5) # one for each run of the block
         expect(a.uniq.size).to be_within(1).of(3) # one for each of the two threads, plus the caller
+
+        subject.shutdown
+        expect(subject.wait_for_termination(1)).to eq true
       end
     end
 
@@ -268,6 +282,8 @@ module Concurrent
 
           subject = FixedThreadPool.new(5, fallback_policy: :discard)
           expect(subject.fallback_policy).to eq :discard
+          subject.shutdown
+          expect(subject.wait_for_termination(1)).to eq true
         end
 
       else
@@ -290,7 +306,8 @@ module Concurrent
             5.times{ pool << proc{ sleep(1) } }
             sleep(0.1)
             expect(pool.length).to eq 5
-            pool.kill
+            pool.shutdown
+            expect(pool.wait_for_termination(1)).to eq true
           end
         end
       end

spec/concurrent/executor/java_single_thread_executor_spec.rb

@@ -7,8 +7,8 @@ module Concurrent
     RSpec.describe JavaSingleThreadExecutor, :type=>:jruby do
 
       after(:each) do
-        subject.kill
-        subject.wait_for_termination(0.1)
+        subject.shutdown
+        expect(subject.wait_for_termination(1)).to eq true
       end
 
       subject { JavaSingleThreadExecutor.new }

spec/concurrent/executor/java_thread_pool_executor_spec.rb

@@ -7,8 +7,8 @@ module Concurrent
     RSpec.describe JavaThreadPoolExecutor, :type => :jruby do
 
       after(:each) do
-        subject.kill
-        subject.wait_for_termination(0.1)
+        subject.shutdown
+        expect(subject.wait_for_termination(1)).to eq true
       end
 
       subject do

spec/concurrent/executor/ruby_single_thread_executor_spec.rb

@@ -5,8 +5,8 @@ module Concurrent
   RSpec.describe RubySingleThreadExecutor, :type=>:mrirbx do
 
     after(:each) do
-      subject.kill
-      subject.wait_for_termination(0.1)
+      subject.shutdown
+      expect(subject.wait_for_termination(1)).to eq true
     end
 
     subject { RubySingleThreadExecutor.new }

spec/concurrent/executor/ruby_thread_pool_executor_spec.rb

@@ -5,8 +5,8 @@ module Concurrent
   RSpec.describe RubyThreadPoolExecutor, :type=>:mrirbx do
 
     after(:each) do
-      subject.kill
-      subject.wait_for_termination(0.1)
+      subject.shutdown
+      expect(subject.wait_for_termination(1)).to eq true
     end
 
     subject do
@@ -46,10 +46,12 @@ module Concurrent
       end
 
       it 'returns the remaining capacity when tasks are enqueued' do
-        100.times{ subject.post{ sleep(0.5) } }
+        block = Concurrent::CountDownLatch.new
+        100.times{ subject.post{ block.wait } }
         subject.post { latch.count_down }
         latch.wait(0.1)
         expect(subject.remaining_capacity).to be < expected_max
+        block.count_down
       end
     end
   end