Merge pull request #326 from ruby-concurrency/mri-max-threads

Prevent Ruby thread pools from creating too many threads.

Author: jdantonio

examples/stress_ruby_thread_pool.rb

@@ -0,0 +1,26 @@
+#!/usr/bin/env ruby
+
+$: << File.expand_path('../../lib', __FILE__)
+
+require 'benchmark'
+require 'concurrent/executors'
+
+COUNT = 100_000
+
+executor = Concurrent::CachedThreadPool.new
+latch = Concurrent::CountDownLatch.new
+
+COUNT.times { executor.post{ nil } }
+
+#COUNT.times do |i|
+#  executor.post{ nil }
+#  sleep(0.01) if i % 1000 == 0
+#end
+
+executor.post{ latch.count_down }
+latch.wait
+
+puts "Max length:           #{executor.max_length}" if executor.respond_to?(:max_length)
+puts "Largest length:       #{executor.largest_length}" if executor.respond_to?(:largest_length)
+puts "Scheduled task count: #{executor.scheduled_task_count}" if executor.respond_to?(:scheduled_task_count)
+puts "Completed task count: #{executor.completed_task_count}" if executor.respond_to?(:completed_task_count)

lib/concurrent/executor/java_cached_thread_pool.rb

@@ -17,22 +17,21 @@ class JavaCachedThreadPool < JavaThreadPoolExecutor
       #
       # @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newCachedThreadPool--
       def initialize(opts = {})
-        super(opts)
+        defaults  = { idletime: DEFAULT_THREAD_IDLETIMEOUT }
+        overrides = { min_threads:     0,
+                      max_threads:     DEFAULT_MAX_POOL_SIZE,
+                      max_queue:       0 }
+        super(defaults.merge(opts).merge(overrides))
       end
 
       protected
 
       def ns_initialize(opts)
-        @fallback_policy = opts.fetch(:fallback_policy, opts.fetch(:overflow_policy, :abort))
-        deprecated ':overflow_policy is deprecated terminology, please use :fallback_policy instead' if opts.has_key?(:overflow_policy)
+        super(opts)
         @max_queue = 0
-
-        raise ArgumentError.new("#{@fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICY_CLASSES.keys.include?(@fallback_policy)
-
         @executor = java.util.concurrent.Executors.newCachedThreadPool
         @executor.setRejectedExecutionHandler(FALLBACK_POLICY_CLASSES[@fallback_policy].new)
         @executor.setKeepAliveTime(opts.fetch(:idletime, DEFAULT_THREAD_IDLETIMEOUT), java.util.concurrent.TimeUnit::SECONDS)
-
         self.auto_terminate = opts.fetch(:auto_terminate, true)
       end
     end

lib/concurrent/executor/java_fixed_thread_pool.rb

@@ -18,12 +18,12 @@ class JavaFixedThreadPool < JavaThreadPoolExecutor
       #
       # @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newFixedThreadPool-int-
       def initialize(num_threads, opts = {})
-
-        opts = {
-            min_threads: num_threads,
-            max_threads: num_threads
-        }.merge(opts)
-        super(opts)
+        raise ArgumentError.new('number of threads must be greater than zero') if num_threads.to_i < 1
+        defaults  = { max_queue:   DEFAULT_MAX_QUEUE_SIZE,
+                      idletime:    DEFAULT_THREAD_IDLETIMEOUT }
+        overrides = { min_threads: num_threads,
+                      max_threads: num_threads }
+        super(defaults.merge(opts).merge(overrides))
       end
     end
   end

lib/concurrent/executor/java_thread_pool_executor.rb

@@ -141,9 +141,10 @@ def ns_initialize(opts)
         @fallback_policy = opts.fetch(:fallback_policy, opts.fetch(:overflow_policy, :abort))
         deprecated ' :overflow_policy is deprecated terminology, please use :fallback_policy instead' if opts.has_key?(:overflow_policy)
 
-        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("`max_threads` cannot be less than #{DEFAULT_MIN_POOL_SIZE}") if max_length < DEFAULT_MIN_POOL_SIZE
+        raise ArgumentError.new("`max_threads` cannot be greater than #{DEFAULT_MAX_POOL_SIZE}") if max_length > DEFAULT_MAX_POOL_SIZE
+        raise ArgumentError.new("`min_threads` cannot be less than #{DEFAULT_MIN_POOL_SIZE}") if min_length < DEFAULT_MIN_POOL_SIZE
+        raise ArgumentError.new("`min_threads` cannot be more than `max_threads`") if min_length > max_length
         raise ArgumentError.new("#{fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICY_CLASSES.include?(@fallback_policy)
 
         if @max_queue == 0

lib/concurrent/executor/ruby_cached_thread_pool.rb

@@ -13,15 +13,10 @@ class RubyCachedThreadPool < RubyThreadPoolExecutor
     #
     # @raise [ArgumentError] if `fallback_policy` is not a known policy
     def initialize(opts = {})
-      fallback_policy = opts.fetch(:fallback_policy, opts.fetch(:overflow_policy, :abort))
-      raise ArgumentError.new("#{fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICIES.include?(fallback_policy)
-
       defaults  = { idletime: DEFAULT_THREAD_IDLETIMEOUT }
       overrides = { min_threads:     0,
                     max_threads:     DEFAULT_MAX_POOL_SIZE,
-                    fallback_policy: fallback_policy,
                     max_queue:       DEFAULT_MAX_QUEUE_SIZE }
-
       super(defaults.merge(opts).merge(overrides))
     end
   end

lib/concurrent/executor/ruby_fixed_thread_pool.rb

@@ -15,19 +15,12 @@ class RubyFixedThreadPool < RubyThreadPoolExecutor
     # @raise [ArgumentError] if `num_threads` is less than or equal to zero
     # @raise [ArgumentError] if `fallback_policy` is not a known policy
     def initialize(num_threads, opts = {})
-      fallback_policy = opts.fetch(:fallback_policy, opts.fetch(:overflow_policy, :abort))
-      raise ArgumentError.new("#{fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICIES.include?(fallback_policy)
-
-      raise ArgumentError.new('number of threads must be greater than zero') if num_threads < 1
-
-      opts = {
-        min_threads: num_threads,
-        max_threads: num_threads,
-        fallback_policy: fallback_policy,
-        max_queue: DEFAULT_MAX_QUEUE_SIZE,
-        idletime: DEFAULT_THREAD_IDLETIMEOUT,
-      }.merge(opts)
-      super(opts)
+      raise ArgumentError.new('number of threads must be greater than zero') if num_threads.to_i < 1
+      defaults  = { max_queue:   DEFAULT_MAX_QUEUE_SIZE,
+                    idletime:    DEFAULT_THREAD_IDLETIMEOUT }
+      overrides = { min_threads: num_threads,
+                    max_threads: num_threads }
+      super(defaults.merge(opts).merge(overrides))
     end
   end
 end

lib/concurrent/executor/ruby_thread_pool_executor.rb

@@ -12,7 +12,7 @@ module Concurrent
   class RubyThreadPoolExecutor < RubyExecutorService
 
     # Default maximum number of threads that will be created in the pool.
-    DEFAULT_MAX_POOL_SIZE      = 2**13 # 8192
+    DEFAULT_MAX_POOL_SIZE      = 2_147_483_647 # java.lang.Integer::MAX_VALUE
 
     # Default minimum number of threads that will be retained in the pool.
     DEFAULT_MIN_POOL_SIZE      = 0
@@ -139,9 +139,10 @@ def ns_initialize(opts)
       raise ArgumentError.new("#{@fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICIES.include?(@fallback_policy)
       deprecated ':overflow_policy is deprecated terminology, please use :fallback_policy instead' if opts.has_key?(:overflow_policy)
 
-      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("`max_threads` cannot be less than #{DEFAULT_MIN_POOL_SIZE}") if @max_length < DEFAULT_MIN_POOL_SIZE
+      raise ArgumentError.new("`max_threads` cannot be greater than #{DEFAULT_MAX_POOL_SIZE}") if @max_length > DEFAULT_MAX_POOL_SIZE
+      raise ArgumentError.new("`min_threads` cannot be less than #{DEFAULT_MIN_POOL_SIZE}") if @min_length < DEFAULT_MIN_POOL_SIZE
+      raise ArgumentError.new("`min_threads` cannot be more than `max_threads`") if min_length > max_length
 
       self.auto_terminate = opts.fetch(:auto_terminate, true)
 
@@ -216,6 +217,9 @@ def ns_assign_worker(*args, &task)
       else
         false
       end
+    rescue ThreadError
+      # Raised when the operating system refuses to create the new thread
+      return false
     end
 
     # tries to enqueue task

spec/concurrent/executor/thread_pool_executor_shared.rb

@@ -60,16 +60,27 @@
       end
     end
 
+    it 'raises an exception if :max_threads is less than zero' do
+      expect {
+        described_class.new(max_threads: -1)
+      }.to raise_error(ArgumentError)
+    end
 
     it 'raises an exception if :min_threads is less than zero' do
       expect {
         described_class.new(min_threads: -1)
       }.to raise_error(ArgumentError)
     end
 
-    it 'raises an exception if :max_threads is not greater than zero' do
+    it 'raises an exception if :max_threads greater than the max allowable' do
+      expect {
+        described_class.new(max_threads: described_class::DEFAULT_MAX_POOL_SIZE+1)
+      }.to raise_error(ArgumentError)
+    end
+
+    it 'raises an exception if :max_threads is less than :min_threads' do
       expect {
-        described_class.new(max_threads: 0)
+        described_class.new(max_threads: 1, min_threads: 100)
       }.to raise_error(ArgumentError)
     end