Extend reworking of queue overflow tests to :abort tests

Author: rkday

spec/concurrent/executor/thread_pool_executor_shared.rb

@@ -216,30 +216,100 @@
       end
 
       specify 'a #post task is never executed when the queue is at capacity' do
-        executed = Concurrent::AtomicFixnum.new(0)
-        10.times do
-          begin
-            subject.post{ executed.increment; sleep(0.1) }
-          rescue
-          end
+        all_tasks_posted = Concurrent::Event.new
+
+        latch = Concurrent::CountDownLatch.new(max_threads)
+        
+        initial_executed = Concurrent::AtomicFixnum.new(0)
+        subsequent_executed = Concurrent::AtomicFixnum.new(0)
+
+        # Fill up all the threads (with a task that won't complete until
+        # all tasks are posted)
+        max_threads.times do
+          subject.post{ latch.count_down; all_tasks_posted.wait ; initial_executed.increment;}
         end
-        sleep(0.2)
-        expect(executed.value).to be < 10
+
+        # Wait for all those tasks to be taken off the queue onto a
+        # worker thread and start executing
+        latch.wait
+        
+        # Fill up the queue (with a task that won't complete until
+        # all tasks are posted)
+        max_queue.times do
+          subject.post{ all_tasks_posted.wait; initial_executed.increment; }
+        end
+
+        # Inject 100 more tasks, which should throw an exception
+        100.times do
+          expect {
+            subject.post { subsequent_executed.increment; }
+          }.to raise_error(Concurrent::RejectedExecutionError)
+        end
+
+        # Trigger the event, so that the tasks in the threads and on
+        # the queue can run to completion
+        all_tasks_posted.set
+
+        # Wait for all tasks to finish
+        subject.shutdown
+        subject.wait_for_termination
+
+        # The tasks should have run until all the threads and the
+        # queue filled up...
+        expect(initial_executed.value).to be (max_threads + max_queue)
+
+        # ..but been dropped after that
+        expect(subsequent_executed.value).to be 0
       end
 
       specify 'a #<< task is never executed when the queue is at capacity' do
-        executed = Concurrent::AtomicFixnum.new(0)
-        10.times do
-          begin
-            subject << proc { executed.increment; sleep(0.1) }
-          rescue
-          end
+        all_tasks_posted = Concurrent::Event.new
+
+        latch = Concurrent::CountDownLatch.new(max_threads)
+        
+        initial_executed = Concurrent::AtomicFixnum.new(0)
+        subsequent_executed = Concurrent::AtomicFixnum.new(0)
+
+        # Fill up all the threads (with a task that won't complete until
+        # all tasks are posted)
+        max_threads.times do
+          subject << proc { latch.count_down; all_tasks_posted.wait ; initial_executed.increment;}
+        end
+
+        # Wait for all those tasks to be taken off the queue onto a
+        # worker thread and start executing
+        latch.wait
+        
+        # Fill up the queue (with a task that won't complete until
+        # all tasks are posted)
+        max_queue.times do
+          subject << proc { all_tasks_posted.wait; initial_executed.increment; }
         end
-        sleep(0.2)
-        expect(executed.value).to be < 10
+
+        # Inject 100 more tasks, which should throw an exeption
+        100.times do
+          expect {
+            subject << proc { subsequent_executed.increment; }
+          }.to raise_error(Concurrent::RejectedExecutionError)
+        end
+
+        # Trigger the event, so that the tasks in the threads and on
+        # the queue can run to completion
+        all_tasks_posted.set
+
+        # Wait for all tasks to finish
+        subject.shutdown
+        subject.wait_for_termination
+
+        # The tasks should have run until all the threads and the
+        # queue filled up...
+        expect(initial_executed.value).to be (max_threads + max_queue)
+
+        # ..but been rejected after that
+        expect(subsequent_executed.value).to be 0
       end
     end
-
+    
     context ':discard' do
 
       subject do
@@ -367,9 +437,8 @@
       end
 
       specify '#post returns false when the executor is shutting down' do
-        executed = Concurrent::AtomicFixnum.new(0)
         subject.shutdown
-        ret = subject.post{ executed.increment }
+        ret = subject.post{ nil }
         expect(ret).to be false
       end
     end