Merge pull request #840 from fzakaria/ISSUE-839

Introduce DaemonThreadFactory

Author: pitr-ch

lib/concurrent-ruby/concurrent/executor/cached_thread_pool.rb

@@ -51,11 +51,13 @@ def initialize(opts = {})
     def ns_initialize(opts)
       super(opts)
       if Concurrent.on_jruby?
+        self.auto_terminate = opts.fetch(:auto_terminate, true)
         @max_queue = 0
-        @executor = java.util.concurrent.Executors.newCachedThreadPool
+        @executor = java.util.concurrent.Executors.newCachedThreadPool(
+            Concurrent::DaemonThreadFactory.new(self.auto_terminate?)
+        )
         @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
   end

lib/concurrent-ruby/concurrent/executor/java_daemon_thread_factory.rb

@@ -0,0 +1,17 @@
+module Concurrent
+
+  class DaemonThreadFactory
+    include java.util.concurrent.ThreadFactory
+
+    def initialize(daemonize = true)
+      @daemonize = daemonize
+    end
+
+    def newThread(runnable)
+      thread = java.util.concurrent.Executors.defaultThreadFactory().newThread(runnable)
+      thread.setDaemon(@daemonize)
+      return thread
+    end
+  end
+
+end

lib/concurrent-ruby/concurrent/executor/java_thread_pool_executor.rb

@@ -108,15 +108,17 @@ def ns_initialize(opts)
           queue = java.util.concurrent.LinkedBlockingQueue.new(@max_queue)
         end
 
+        self.auto_terminate = opts.fetch(:auto_terminate, true)
+
         @executor = java.util.concurrent.ThreadPoolExecutor.new(
             min_length,
             max_length,
             idletime,
             java.util.concurrent.TimeUnit::SECONDS,
             queue,
+            Concurrent::DaemonThreadFactory.new(self.auto_terminate?),
             FALLBACK_POLICY_CLASSES[@fallback_policy].new)
 
-        self.auto_terminate = opts.fetch(:auto_terminate, true)
       end
     end
   end

lib/concurrent-ruby/concurrent/executor/thread_pool_executor.rb

@@ -5,6 +5,7 @@ module Concurrent
 
   if Concurrent.on_jruby?
     require 'concurrent/executor/java_thread_pool_executor'
+    require 'concurrent/executor/java_daemon_thread_factory'
   end
 
   ThreadPoolExecutorImplementation = case

spec/concurrent/executor/cached_thread_pool_spec.rb

@@ -45,14 +45,14 @@ module Concurrent
       end
 
       it 'returns zero while running' do
-        10.times{ subject.post{ nil } }
+        10.times { subject.post { nil } }
         subject.post { latch.count_down }
         latch.wait(0.1)
         expect(subject.min_length).to eq 0
       end
 
       it 'returns zero once shutdown' do
-        10.times{ subject.post{ nil } }
+        10.times { subject.post { nil } }
         subject.post { latch.count_down }
         latch.wait(0.1)
         subject.shutdown
@@ -68,14 +68,14 @@ module Concurrent
       end
 
       it 'returns :max_length while running' do
-        10.times{ subject.post{ nil } }
+        10.times { subject.post { nil } }
         subject.post { latch.count_down }
         latch.wait(0.1)
         expect(subject.max_length).to eq described_class::DEFAULT_MAX_POOL_SIZE
       end
 
       it 'returns :max_length once shutdown' do
-        10.times{ subject.post{ nil } }
+        10.times { subject.post { nil } }
         subject.post { latch.count_down }
         latch.wait(0.1)
         subject.shutdown
@@ -91,14 +91,14 @@ module Concurrent
       end
 
       it 'returns a non-zero number once tasks have been received' do
-        10.times{ subject.post{ sleep(0.1) } }
+        10.times { subject.post { sleep(0.1) } }
         subject.post { latch.count_down }
         latch.wait(0.1)
         expect(subject.largest_length).to be > 0
       end
 
       it 'returns a non-zero number after shutdown if tasks have been received' do
-        10.times{ subject.post{ sleep(0.1) } }
+        10.times { subject.post { sleep(0.1) } }
         subject.post { latch.count_down }
         latch.wait(0.1)
         subject.shutdown
@@ -109,7 +109,7 @@ module Concurrent
 
     context '#idletime' do
 
-      subject{ described_class.new(idletime: 42) }
+      subject { described_class.new(idletime: 42) }
 
       it 'returns the thread idletime' do
         expect(subject.idletime).to eq 42
@@ -123,7 +123,7 @@ module Concurrent
         context '#initialize' do
 
           it 'sets :fallback_policy correctly' do
-            clazz = java.util.concurrent.ThreadPoolExecutor::DiscardPolicy
+            clazz  = java.util.concurrent.ThreadPoolExecutor::DiscardPolicy
             policy = clazz.new
             expect(clazz).to receive(:new).at_least(:once).with(any_args).and_return(policy)
 
@@ -201,6 +201,26 @@ module Concurrent
         end
       end
 
+      context 'auto terminate' do
+
+        # https://github.com/ruby-concurrency/concurrent-ruby/issues/817
+        # https://github.com/ruby-concurrency/concurrent-ruby/issues/839
+        it 'does not stop shutdown ' do
+          Timeout.timeout(10) do
+            begin
+              test_file = File.join File.dirname(__FILE__), 'pool_quits.rb'
+              pid       = spawn RbConfig.ruby, test_file
+              Process.waitpid pid
+              expect($?.success?).to eq true
+            rescue Timeout::Error => e
+              Process.kill :KILL, pid
+              raise e
+            end
+          end
+        end
+
+      end
+
       context 'stress', notravis: true do
         configurations = [
             { min_threads:    2,

spec/concurrent/executor/pool_quits.rb

@@ -0,0 +1,14 @@
+
+lib = File.expand_path '../../../lib/concurrent-ruby/'
+$LOAD_PATH.push lib unless $LOAD_PATH.include? lib
+
+require 'concurrent-ruby'
+
+# the test relies on replicating that Minitest messed up the AtExit handling
+Concurrent.disable_at_exit_handlers!
+pool = Concurrent::CachedThreadPool.new
+pool.post do
+  sleep # sleep indefinitely
+end
+
+# the process main thread should quit out which should kill the daemon CachedThreadPool