Stabilize

Author: pitr-ch

spec/concurrent/agent_spec.rb

@@ -447,8 +447,8 @@ def update(time, old_value, new_value)
         end
 
         it 'posts to the global fast executor' do
-          expect(Concurrent.global_fast_executor).to receive(:post).with(any_args).and_call_original
           subject = Agent.new(0)
+          expect(subject).to receive(:enqueue_action_job).with(anything, anything, Concurrent.global_fast_executor).and_call_original
           subject.send { nil }
         end
 
@@ -476,8 +476,8 @@ def update(time, old_value, new_value)
         end
 
         it 'posts to the global fast executor' do
-          expect(Concurrent.global_fast_executor).to receive(:post).with(any_args).and_call_original
           subject = Agent.new(0)
+          expect(subject).to receive(:enqueue_action_job).with(anything, anything, Concurrent.global_fast_executor).and_call_original
           subject.send! { nil }
         end
 
@@ -503,8 +503,8 @@ def update(time, old_value, new_value)
         end
 
         it 'posts to the global io executor' do
-          expect(Concurrent.global_io_executor).to receive(:post).with(any_args).and_call_original
           subject = Agent.new(0)
+          expect(subject).to receive(:enqueue_action_job).with(anything, anything, Concurrent.global_io_executor).and_call_original
           subject.send_off { nil }
         end
 
@@ -532,8 +532,8 @@ def update(time, old_value, new_value)
         end
 
         it 'posts to the global io executor' do
-          expect(Concurrent.global_io_executor).to receive(:post).with(any_args).and_call_original
           subject = Agent.new(0)
+          expect(subject).to receive(:enqueue_action_job).with(anything, anything, Concurrent.global_io_executor).and_call_original
           subject.send_off! { nil }
         end
 
@@ -615,8 +615,8 @@ def update(time, old_value, new_value)
         end
 
         it 'posts to the global io executor' do
-          expect(Concurrent.global_io_executor).to receive(:post).with(any_args).and_call_original
           subject = Agent.new(0)
+          expect(subject).to receive(:enqueue_action_job).with(anything, anything, Concurrent.global_io_executor).and_call_original
           subject.post { nil }
         end
 
@@ -1121,7 +1121,7 @@ def update(time, old_value, new_value)
           agents = 3.times.collect { Agent.new(0) }
           agents.each { |agent| agent.send_via(executor, latch) { |_, l| l.wait(1) } }
           in_thread { latch.count_down }
-          ok = Agent.await_for(1, *agents)
+          ok = Agent.await_for(5, *agents)
           expect(ok).to be true
         end
 
@@ -1146,7 +1146,7 @@ def update(time, old_value, new_value)
           agents = 3.times.collect { Agent.new(0) }
           agents.each { |agent| agent.send_via(executor, latch) { |_, l| l.wait(1) } }
           in_thread { latch.count_down }
-          ok = Agent.await_for!(1, *agents)
+          ok = Agent.await_for!(5, *agents)
           expect(ok).to be true
         end
 

spec/concurrent/atomic/cyclic_barrier_spec.rb

@@ -196,17 +196,16 @@ module Concurrent
 
       context '#broken barrier' do
         it 'should not accept new threads' do
-          t = in_thread { barrier.wait(0.1) }
-          t.join(0.2)
+          t = in_thread { barrier.wait(0.01) }
+          join_with t
 
           expect(barrier).to be_broken
-
           expect(barrier.wait).to be_falsey
         end
 
         it 'can be reset' do
-          t = in_thread { barrier.wait(0.1) }
-          t.join(0.2)
+          t = in_thread { barrier.wait(0.01) }
+          join_with t
 
           expect(barrier).to be_broken
 

spec/concurrent/atomic/event_spec.rb

@@ -139,6 +139,7 @@ module Concurrent
       end
 
       it 'behaves appropriately if wait begins while #set is processing' do
+        subject = subject()
         subject.reset
         latch = CountDownLatch.new(5)
         5.times{ in_thread{ subject.wait(5) } }

spec/concurrent/map_spec.rb

@@ -282,31 +282,33 @@ module Concurrent
           compute_finished.count_down
         end
 
-        getter_threads = (1..getters_count).map do
+        getter_threads = getters_count.times.map do
           in_thread do
             getters_started.count_down
             inserted_keys.each do |inserted_key|
-              expect(true).to eq @cache.key?(inserted_key)
-              expect(1).to    eq @cache[inserted_key]
+              expect(@cache.key?(inserted_key)).to eq true
+              expect(@cache[inserted_key]).to eq 1
             end
             expect(false).to eq @cache.key?(key)
+
             compute_started.wait
+
             inserted_keys.each do |inserted_key|
-              expect(true).to eq @cache.key?(inserted_key)
-              expect(1).to    eq @cache[inserted_key]
+              expect(@cache.key?(inserted_key)).to eq true
+              expect(@cache[inserted_key]).to eq 1
             end
-            expect(false).to eq @cache.key?(key)
-            expect(nil).to   eq @cache[key]
+            expect(@cache.key?(key)).to eq false
+            expect(@cache[key]).to eq nil
             getters_finished.count_down
+
             compute_finished.wait
-            expect(true).to eq @cache.key?(key)
-            expect(1).to    eq @cache[key]
+
+            expect(@cache.key?(key)).to eq true
+            expect(@cache[key]).to eq 1
           end
         end
 
-        (getter_threads + [computer_thread]).map do |t|
-          expect(t.join(2)).to be_truthy
-        end # asserting no deadlocks
+        join_with getter_threads + [computer_thread]
         inserted_keys << key
       end
     end

spec/concurrent/synchronization_spec.rb

@@ -125,7 +125,7 @@ def ns_initialize
             subject.wait
           end
           t2 = in_thread { Thread.pass until t1.status == 'sleep' }
-          join_with t2, 5
+          join_with t2
         end
 
         it 'allows the sleeping thread to be killed' do
@@ -136,7 +136,7 @@ def ns_initialize
           sleep 0.1
           t.kill
           sleep 0.1
-          expect(t.status).to eq false
+          expect(t.join).not_to eq nil
           expect(t.alive?).to eq false
         end
 
@@ -148,7 +148,7 @@ def ns_initialize
             subject.synchronize {} # it will deadlock here if #wait doesn't release lock
             t2
           end
-          join_with t1, 5
+          join_with t1
           expect(t1.value.status).to eq 'sleep'
         end
 
@@ -159,7 +159,7 @@ def ns_initialize
             subject.synchronize {} # it will deadlock here if #wait doesn't release lock
             t2
           end
-          join_with t1, 5
+          join_with t1
           expect(t1.value.status).to eq 'sleep'
         end
       end

spec/support/example_group_extensions.rb

@@ -28,12 +28,20 @@ def monotonic_interval
 
     def in_thread(*args, &block)
       @created_threads ||= Queue.new
-      @created_threads.push t = Thread.new(*args, &block)
-      t
+      new_thread = Thread.new(*args) do |*args, &b|
+        Thread.abort_on_exception = true
+        block.call *args, &b
+      end
+      @created_threads.push new_thread
+      new_thread
     end
 
-    def join_with(threads, timeout = 0.1)
-      Array(threads).each { |t| expect(t.join(timeout)).not_to eq nil }
+    def join_with(threads, timeout = 5)
+      threads = Array(threads)
+      threads.each do |t|
+        joined_thread = t.join(timeout * threads.size)
+        expect(joined_thread).not_to eq nil
+      end
     end
   end
 end
@@ -42,10 +50,14 @@ class RSpec::Core::ExampleGroup
   include Concurrent::TestHelpers
   extend Concurrent::TestHelpers
 
+  before :each do
+    expect(@created_threads.nil? || @created_threads.empty?).to be_truthy
+  end
+
   after :each do
     while (thread = (@created_threads.pop(true) rescue nil))
       thread.kill
-      expect(thread.join(0.25)).not_to eq nil
+      expect(thread.join(0.25)).not_to be_nil
     end
   end
 end