Implemented queue length functionality for RubyThreadPoolExecutor.

Author: jdantonio

lib/concurrent/java_thread_pool_executor.rb

@@ -158,8 +158,8 @@ def <<(task)
       # but no new tasks will be accepted. Has no additional effect if the
       # thread pool is not running.
       def shutdown
-        @executor.shutdown
         @executor.getQueue.clear
+        @executor.shutdown
         return nil
       end
 
@@ -168,8 +168,8 @@ def shutdown
       # will be accepted. Has no additional effect if the thread pool is
       # not running.
       def kill
-        @executor.shutdownNow
         @executor.getQueue.clear
+        @executor.shutdownNow
         return nil
       end
 

lib/concurrent/ruby_thread_pool_executor.rb

@@ -37,6 +37,8 @@ class RubyThreadPoolExecutor
 
     attr_reader :max_queue
 
+    attr_reader :overflow_policy
+
     # Create a new thread pool.
     #
     # @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ThreadPoolExecutor.html
@@ -59,7 +61,6 @@ def initialize(opts = {})
       @completed_task_count = 0
       @largest_length = 0
 
-      #@busy = 0
       @gc_interval = opts.fetch(:gc_interval, 1).to_i # undocumented
       @last_gc_time = Time.now.to_f - [1.0, (@gc_interval * 2.0)].max
     end
@@ -72,9 +73,11 @@ def length
     alias_method :current_length, :length
 
     def queue_length
+      @queue.length
     end
 
     def remaining_capacity
+      @mutex.synchronize { @max_queue == 0 ? -1 : @max_queue - @queue.length }
     end
 
     # Is the thread pool running?
@@ -118,6 +121,7 @@ def post(*args, &task)
       raise ArgumentError.new('no block given') unless block_given?
       @mutex.synchronize do
         break false unless @state == :running
+        return false if @max_queue != 0 && @queue.length >= @max_queue
         @scheduled_task_count += 1
         @queue << [args, task]
         if Time.now.to_f - @gc_interval >= @last_gc_time
@@ -145,6 +149,7 @@ def <<(task)
     def shutdown
       @mutex.synchronize do
         break unless @state == :running
+        @queue.clear
         if @pool.empty?
           @state = :shutdown
           @terminator.set
@@ -162,24 +167,16 @@ def shutdown
     def kill
       @mutex.synchronize do
         break if @state == :shutdown
-        @state = :shutdown
         @queue.clear
+        @state = :shutdown
         drain_pool
         @terminator.set
       end
     end
 
-    ## @!visibility private
-    #def on_start_task # :nodoc:
-      #@mutex.synchronize do
-        #@busy += 1
-      #end
-    #end
-
     # @!visibility private
     def on_end_task # :nodoc:
       @mutex.synchronize do
-        #@busy -= 1
         @completed_task_count += 1 #if success
         break unless @state == :running
       end

lib/concurrent/ruby_thread_pool_worker.rb

@@ -49,7 +49,6 @@ def run(thread = Thread.current) # :nodoc:
         end
 
         begin
-          #@parent.on_start_task
           task.last.call(*task.first)
         rescue => ex
           # let it fail

spec/concurrent/thread_pool_executor_shared.rb

@@ -64,6 +64,15 @@
 
   context '#queue_length' do
 
+    let!(:expected_max){ 10 }
+    subject do
+      described_class.new(
+        min_threads: 2,
+        max_threads: 5,
+        max_queue: expected_max
+      )
+    end
+
     it 'returns zero on creation' do
       subject.queue_length.should eq 0
     end
@@ -87,6 +96,12 @@
       subject.wait_for_termination(1)
       subject.queue_length.should eq 0
     end
+
+    it 'can never be greater than :max_queue' do
+      100.times{ subject.post{ sleep(0.5) } }
+      sleep(0.1)
+      subject.queue_length.should <= expected_max
+    end
   end
 
   context '#remaining_capacity' do
@@ -116,6 +131,7 @@
 
     it 'returns :max_length when stopped' do
       100.times{ subject.post{ sleep(0.5) } }
+      sleep(0.1)
       subject.shutdown
       subject.wait_for_termination(1)
       subject.remaining_capacity.should eq expected_max