Merge pull request #727 from ruby-concurrency/pitr-ch/pool

Fix #717: Reenable java implementations, fix global IO executor on JRuby, bunch of other bug fixes

Author: pitr-ch

Gemfile

@@ -10,8 +10,9 @@ group :development do
   gem 'rake-compiler-dock', '~> 0.6.0'
   gem 'gem-compiler', '~> 0.3.0'
   gem 'benchmark-ips', '~> 2.7'
+end
 
-  # documentation
+group :documentation do
   gem 'countloc', '~> 0.4.0', :platforms => :mri, :require => false
   # TODO (pitr-ch 04-May-2018): update to remove: [DEPRECATION] `last_comment` is deprecated.  Please use `last_description` instead.
   gem 'yard', '~> 0.8.0', :require => false

concurrent-ruby-edge.gemspec

@@ -1,8 +1,5 @@
-$:.push File.join(File.dirname(__FILE__), 'lib')
-$:.push File.join(File.dirname(__FILE__), 'support')
-
-require 'concurrent/version'
-require 'file_map'
+require_relative 'lib/concurrent/version'
+require_relative 'support/file_map'
 
 Gem::Specification.new do |s|
   git_files = `git ls-files`.split("\n")

concurrent-ruby-ext.gemspec

@@ -1,6 +1,4 @@
-$:.push File.join(File.dirname(__FILE__), 'lib')
-
-require 'concurrent/version'
+require_relative 'lib/concurrent/version'
 
 Gem::Specification.new do |s|
   s.name        = 'concurrent-ruby-ext'

concurrent-ruby.gemspec

@@ -1,33 +1,26 @@
-$:.push File.join(File.dirname(__FILE__), 'lib')
-$:.push File.join(File.dirname(__FILE__), 'support')
-
-require 'concurrent/version'
-require 'file_map'
+require_relative 'lib/concurrent/version'
+require_relative 'lib/concurrent/utility/engine'
+require_relative 'support/file_map'
 
 Gem::Specification.new do |s|
-  git_files = `git ls-files`.split("\n")
-
   s.name             = 'concurrent-ruby'
   s.version          = Concurrent::VERSION
-  s.platform         = Gem::Platform::RUBY
+  s.platform         = Concurrent.on_jruby? ? Gem::Platform::JAVA : Gem::Platform::RUBY
   s.authors          = ["Jerry D'Antonio", 'Petr Chalupa', 'The Ruby Concurrency Team']
   s.email            = '[email protected]'
   s.homepage         = 'http://www.concurrent-ruby.com'
   s.summary          = 'Modern concurrency tools for Ruby. Inspired by Erlang, Clojure, Scala, Haskell, F#, C#, Java, and classic concurrency patterns.'
   s.license          = 'MIT'
   s.date             = Time.now.strftime('%Y-%m-%d')
-  s.files            = FileMap::MAP.fetch(:core)
+  s.files            = [*FileMap::MAP.fetch(:core),
+                        *FileMap::MAP.fetch(:spec),
+                        *(Dir['lib/**/*.jar'] if Concurrent.on_jruby?)]
   s.extra_rdoc_files = Dir['README*', 'LICENSE*', 'CHANGELOG*']
   s.require_paths    = ['lib']
-  s.description      = <<-EOF
-Modern concurrency tools including agents, futures, promises, thread pools, actors, supervisors, and more.
-Inspired by Erlang, Clojure, Go, JavaScript, actors, and classic concurrency patterns.
-  EOF
-
-  if defined?(JRUBY_VERSION)
-    s.files    += Dir['lib/**/*.jar']
-    s.platform = 'java'
-  end
+  s.description      = <<-TXT.gsub(/^ +/, '')
+    Modern concurrency tools including agents, futures, promises, thread pools, actors, supervisors, and more.
+    Inspired by Erlang, Clojure, Go, JavaScript, actors, and classic concurrency patterns.
+  TXT
 
   s.required_ruby_version = '>= 1.9.3'
 end

ext/com/concurrent_ruby/ext/SynchronizationLibrary.java

@@ -5,6 +5,7 @@
 import org.jruby.RubyClass;
 import org.jruby.RubyModule;
 import org.jruby.RubyObject;
+import org.jruby.RubyThread;
 import org.jruby.anno.JRubyClass;
 import org.jruby.anno.JRubyMethod;
 import org.jruby.runtime.Block;
@@ -96,6 +97,14 @@ public void load(Ruby runtime, boolean wrap) throws IOException {
 
         defineClass(runtime, synchronizationModule, "AbstractLockableObject", "JRubyLockableObject",
                 JRubyLockableObject.class, JRUBY_LOCKABLE_OBJECT_ALLOCATOR);
+
+        defineClass(runtime, synchronizationModule, "Object", "JRuby",
+                JRuby.class, new ObjectAllocator() {
+                    @Override
+                    public IRubyObject allocate(Ruby runtime, RubyClass klazz) {
+                        return new JRuby(runtime, klazz);
+                    }
+                });
     }
 
     private RubyClass defineClass(
@@ -267,4 +276,29 @@ public IRubyObject nsBroadcast(ThreadContext context) {
             return this;
         }
     }
+
+    @JRubyClass(name = "JRuby")
+    public static class JRuby extends RubyObject {
+        public JRuby(Ruby runtime, RubyClass metaClass) {
+            super(runtime, metaClass);
+        }
+
+        @JRubyMethod(name = "sleep_interruptibly", visibility = Visibility.PUBLIC, module = true)
+        public static IRubyObject sleepInterruptibly(ThreadContext context, IRubyObject receiver, Block block) {
+            try {
+                return context.getThread().executeTask(context, block,
+                        new RubyThread.Task<Block, IRubyObject>() {
+                            public IRubyObject run(ThreadContext context, Block block1) throws InterruptedException {
+                                return block1.call(context);
+                            }
+
+                            public void wakeup(RubyThread thread, Block block1) {
+                                thread.getNativeThread().interrupt();
+                            }
+                        });
+            } catch (InterruptedException e) {
+                throw context.runtime.newThreadError("interrupted in Concurrent::Synchronization::JRuby.sleep_interruptibly");
+            }
+        }
+    }
 }

lib/concurrent/atomic/count_down_latch.rb

@@ -55,7 +55,7 @@ module Concurrent
   # @!macro internal_implementation_note
   CountDownLatchImplementation = case
                                  when Concurrent.on_jruby?
-                                   MutexCountDownLatch
+                                   JavaCountDownLatch
                                  else
                                    MutexCountDownLatch
                                  end

lib/concurrent/atomic/java_count_down_latch.rb

@@ -9,19 +9,20 @@ class JavaCountDownLatch
 
       # @!macro count_down_latch_method_initialize
       def initialize(count = 1)
-        unless count.is_a?(Fixnum) && count >= 0
-          raise ArgumentError.new('count must be in integer greater than or equal zero')
-        end
+        Utility::NativeInteger.ensure_integer_and_bounds(count)
+        Utility::NativeInteger.ensure_positive(count)
         @latch = java.util.concurrent.CountDownLatch.new(count)
       end
 
       # @!macro count_down_latch_method_wait
       def wait(timeout = nil)
         if timeout.nil?
-          @latch.await
+          Synchronization::JRuby.sleep_interruptibly { @latch.await }
           true
         else
-          @latch.await(1000 * timeout, java.util.concurrent.TimeUnit::MILLISECONDS)
+          Synchronization::JRuby.sleep_interruptibly do
+            @latch.await(1000 * timeout, java.util.concurrent.TimeUnit::MILLISECONDS)
+          end
         end
       end
 

lib/concurrent/collection/map/atomic_reference_map_backend.rb

@@ -831,7 +831,7 @@ def rebuild(table)
             # no lock needed (or available) if bin >= 0, because we're not popping values from locked_indexes until we've run through the whole table
             redo unless (bin >= 0 ? table.cas(i, nil, forwarder) : lock_and_clean_up_reverse_forwarders(table, old_table_size, new_table, i, forwarder))
           elsif Node.locked_hash?(node_hash = node.hash)
-            locked_indexes ||= Array.new
+            locked_indexes ||= ::Array.new
             if bin < 0 && locked_arr_idx > 0
               locked_arr_idx -= 1
               i, locked_indexes[locked_arr_idx] = locked_indexes[locked_arr_idx], i # swap with another bin

lib/concurrent/configuration.rb

@@ -175,13 +175,8 @@ def self.new_fast_executor(opts = {})
   end
 
   def self.new_io_executor(opts = {})
-    ThreadPoolExecutor.new(
-        min_threads:     [2, Concurrent.processor_count].max,
-        max_threads:     ThreadPoolExecutor::DEFAULT_MAX_POOL_SIZE,
-        # max_threads:     1000,
+    CachedThreadPool.new(
         auto_terminate:  opts.fetch(:auto_terminate, true),
-        idletime:        60, # 1 minute
-        max_queue:       0, # unlimited
         fallback_policy: :abort # shouldn't matter -- 0 max queue
     )
   end

lib/concurrent/edge/promises.rb

@@ -1797,8 +1797,8 @@ def process_on_blocker_resolution(future, index)
 
       def on_resolvable(resolved_future, index)
         all_fulfilled = true
-        values        = Array.new(@Resolutions.size)
-        reasons       = Array.new(@Resolutions.size)
+        values        = ::Array.new(@Resolutions.size)
+        reasons       = ::Array.new(@Resolutions.size)
 
         @Resolutions.each_with_index do |internal_state, i|
           fulfilled, values[i], reasons[i] = internal_state.result