ExecutorService extends Synchronization::Object

Author: jdantonio

lib/concurrent/executor/executor.rb

@@ -0,0 +1,237 @@
+require 'concurrent/logging'
+require 'concurrent/synchronization'
+require 'concurrent/executor/executor'
+
+module Concurrent
+
+  class AbstractExecutorService < Synchronization::Object
+    include Executor
+    include Logging
+
+    FALLBACK_POLICIES = [:abort, :discard, :caller_runs].freeze
+
+    attr_reader :fallback_policy
+
+    def initialize
+      super()
+    end
+
+    def running?
+      raise NotImplementedError
+    end
+
+    def shuttingdown?
+      raise NotImplementedError
+    end
+
+    def shutdown?
+      raise NotImplementedError
+    end
+
+    def shutdown
+      raise NotImplementedError
+    end
+
+    def kill
+      raise NotImplementedError
+    end
+
+    def wait_for_termination(timeout = nil)
+      raise NotImplementedError
+    end
+
+    def auto_terminate?
+      synchronize { ns_auto_terminate? }
+    end
+
+    def auto_terminate=(value)
+      synchronize { self.ns_auto_terminate = value }
+    end
+
+    protected
+
+    def handle_fallback(*args)
+      case fallback_policy
+      when :abort
+        raise RejectedExecutionError
+      when :discard
+        false
+      when :caller_runs
+        begin
+          yield(*args)
+        rescue => ex
+          # let it fail
+          log DEBUG, ex
+        end
+        true
+      else
+        fail "Unknown fallback policy #{fallback_policy}"
+      end
+    end
+
+    def execute(*args, &task)
+      raise NotImplementedError
+    end
+
+    def shutdown_execution
+    end
+
+    def kill_execution
+      # do nothing
+    end
+
+    private
+
+    def ns_auto_terminate?
+      !!@auto_terminate
+    end
+
+    def ns_auto_terminate=(value)
+      case value
+      when true
+        AtExit.add(self) { terminate_at_exit }
+        @auto_terminate = true
+      when false
+        AtExit.delete(self)
+        @auto_terminate = false
+      else
+        raise ArgumentError
+      end
+    end
+
+    def terminate_at_exit
+      kill # TODO be gentle first
+      wait_for_termination(10)
+    end
+  end
+
+  class RubyExecutorService < AbstractExecutorService
+
+    def initialize
+      super()
+      @stop_event    = Event.new
+      @stopped_event = Event.new
+    end
+
+    def post(*args, &task)
+      raise ArgumentError.new('no block given') unless block_given?
+      synchronize do
+        # If the executor is shut down, reject this task
+        return handle_fallback(*args, &task) unless running?
+        execute(*args, &task)
+        true
+      end
+    end
+
+    def running?
+      !stop_event.set?
+    end
+
+    def shuttingdown?
+      !(running? || shutdown?)
+    end
+
+    def shutdown?
+      stopped_event.set?
+    end
+
+    def shutdown
+      synchronize do
+        break unless running?
+        self.ns_auto_terminate = false
+        stop_event.set
+        shutdown_execution
+      end
+      true
+    end
+
+    def kill
+      synchronize do
+        break if shutdown?
+        self.ns_auto_terminate = false
+        stop_event.set
+        kill_execution
+        stopped_event.set
+      end
+      true
+    end
+
+    def wait_for_termination(timeout = nil)
+      stopped_event.wait(timeout)
+    end
+
+    protected
+
+    attr_reader :stop_event, :stopped_event
+
+    def shutdown_execution
+      stopped_event.set
+    end
+  end
+
+  if Concurrent.on_jruby?
+
+    class JavaExecutorService < AbstractExecutorService
+      include Executor
+      java_import 'java.lang.Runnable'
+
+      FALLBACK_POLICY_CLASSES = {
+        abort:       java.util.concurrent.ThreadPoolExecutor::AbortPolicy,
+        discard:     java.util.concurrent.ThreadPoolExecutor::DiscardPolicy,
+        caller_runs: java.util.concurrent.ThreadPoolExecutor::CallerRunsPolicy
+      }.freeze
+      private_constant :FALLBACK_POLICY_CLASSES
+
+      def initialize
+        super()
+      end
+
+      def post(*args, &task)
+        raise ArgumentError.new('no block given') unless block_given?
+        return handle_fallback(*args, &task) unless running?
+        executor_submit = @executor.java_method(:submit, [Runnable.java_class])
+        executor_submit.call { yield(*args) }
+        true
+      rescue Java::JavaUtilConcurrent::RejectedExecutionException
+        raise RejectedExecutionError
+      end
+
+      def running?
+        !(shuttingdown? || shutdown?)
+      end
+
+      def shuttingdown?
+        if @executor.respond_to? :isTerminating
+          @executor.isTerminating
+        else
+          false
+        end
+      end
+
+      def shutdown?
+        @executor.isShutdown || @executor.isTerminated
+      end
+
+      def wait_for_termination(timeout = nil)
+        if timeout.nil?
+          ok = @executor.awaitTermination(60, java.util.concurrent.TimeUnit::SECONDS) until ok
+          true
+        else
+          @executor.awaitTermination(1000 * timeout, java.util.concurrent.TimeUnit::MILLISECONDS)
+        end
+      end
+
+      def shutdown
+        self.ns_auto_terminate = false
+        @executor.shutdown
+        nil
+      end
+
+      def kill
+        self.ns_auto_terminate = false
+        @executor.shutdownNow
+        nil
+      end
+    end
+  end
+end

lib/concurrent/executor/executor_service.rb

@@ -17,14 +17,16 @@ class JavaCachedThreadPool < JavaThreadPoolExecutor
       #
       # @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/Executors.html#newCachedThreadPool--
       def initialize(opts = {})
+        super()
+
         @fallback_policy = opts.fetch(:fallback_policy, opts.fetch(:overflow_policy, :abort))
         warn '[DEPRECATED] :overflow_policy is deprecated terminology, please use :fallback_policy instead' if opts.has_key?(:overflow_policy)
         @max_queue = 0
 
-        raise ArgumentError.new("#{@fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICIES.keys.include?(@fallback_policy)
+        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_POLICIES[@fallback_policy].new)
+        @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)

lib/concurrent/executor/java_cached_thread_pool.rb

@@ -1,12 +1,11 @@
 if Concurrent.on_jruby?
-  require_relative 'executor'
+  require 'concurrent/executor/executor_service'
 
   module Concurrent
 
     # @!macro single_thread_executor
     # @!macro thread_pool_options
-    class JavaSingleThreadExecutor
-      include JavaExecutor
+    class JavaSingleThreadExecutor < JavaExecutorService
       include SerialExecutor
 
       # Create a new thread pool.
@@ -19,9 +18,10 @@ class JavaSingleThreadExecutor
       # @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Executors.html
       # @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html
       def initialize(opts = {})
+        super()
         @executor = java.util.concurrent.Executors.newSingleThreadExecutor
         @fallback_policy = opts.fetch(:fallback_policy, :discard)
-        raise ArgumentError.new("#{@fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICIES.keys.include?(@fallback_policy)
+        raise ArgumentError.new("#{@fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICY_CLASSES.keys.include?(@fallback_policy)
         self.auto_terminate = opts.fetch(:auto_terminate, true)
       end
     end

lib/concurrent/executor/java_single_thread_executor.rb

@@ -1,12 +1,11 @@
 if Concurrent.on_jruby?
-  require_relative 'executor'
+  require 'concurrent/executor/executor_service'
 
   module Concurrent
 
     # @!macro thread_pool_executor
     # @!macro thread_pool_options
-    class JavaThreadPoolExecutor
-      include JavaExecutor
+    class JavaThreadPoolExecutor < JavaExecutorService
 
       # Default maximum number of threads that will be created in the pool.
       DEFAULT_MAX_POOL_SIZE = java.lang.Integer::MAX_VALUE # 2147483647
@@ -53,6 +52,8 @@ class JavaThreadPoolExecutor
       #
       # @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ThreadPoolExecutor.html
       def initialize(opts = {})
+        super()
+
         min_length = opts.fetch(:min_threads, DEFAULT_MIN_POOL_SIZE).to_i
         max_length = opts.fetch(:max_threads, DEFAULT_MAX_POOL_SIZE).to_i
         idletime = opts.fetch(:idletime, DEFAULT_THREAD_IDLETIMEOUT).to_i
@@ -63,7 +64,7 @@ def initialize(opts = {})
         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("#{fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICIES.include?(@fallback_policy)
+        raise ArgumentError.new("#{fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICY_CLASSES.include?(@fallback_policy)
 
         if @max_queue == 0
           queue = java.util.concurrent.LinkedBlockingQueue.new
@@ -74,7 +75,7 @@ def initialize(opts = {})
         @executor = java.util.concurrent.ThreadPoolExecutor.new(
           min_length, max_length,
           idletime, java.util.concurrent.TimeUnit::SECONDS,
-          queue, FALLBACK_POLICIES[@fallback_policy].new)
+          queue, FALLBACK_POLICY_CLASSES[@fallback_policy].new)
 
         self.auto_terminate = opts.fetch(:auto_terminate, true)
       end

lib/concurrent/executor/java_thread_pool_executor.rb

@@ -1,4 +1,5 @@
 require 'concurrent/atomics'
+require 'concurrent/synchronization'
 require 'concurrent/executor/executor'
 
 module Concurrent
@@ -15,7 +16,7 @@ module Concurrent
   # lead to suboptimal performance.
   #
   # @note Intended for use primarily in testing and debugging.
-  class PerThreadExecutor
+  class PerThreadExecutor < Synchronization::Object
     include Executor
 
     # Creates a new executor

lib/concurrent/executor/per_thread_executor.rb

@@ -14,7 +14,6 @@ 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 }

lib/concurrent/executor/ruby_cached_thread_pool.rb

@@ -16,9 +16,9 @@ class RubyFixedThreadPool < RubyThreadPoolExecutor
     # @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
-      raise ArgumentError.new("#{fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICIES.include?(fallback_policy)
 
       opts = {
         min_threads: num_threads,

lib/concurrent/executor/ruby_fixed_thread_pool.rb

@@ -1,11 +1,11 @@
-require_relative 'executor'
+require 'concurrent/executor/executor'
+require 'concurrent/executor/executor_service'
 
 module Concurrent
 
   # @!macro single_thread_executor
   # @!macro thread_pool_options
-  class RubySingleThreadExecutor
-    include RubyExecutor
+  class RubySingleThreadExecutor < RubyExecutorService
     include SerialExecutor
 
     # Create a new thread pool.
@@ -18,11 +18,11 @@ class RubySingleThreadExecutor
     # @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Executors.html
     # @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html
     def initialize(opts = {})
+      super()
       @queue = Queue.new
       @thread = nil
       @fallback_policy = opts.fetch(:fallback_policy, :discard)
       raise ArgumentError.new("#{@fallback_policy} is not a valid fallback policy") unless FALLBACK_POLICIES.include?(@fallback_policy)
-      init_executor
       self.auto_terminate = opts.fetch(:auto_terminate, true)
     end