Moved executor classes into own files.

Author: jdantonio

lib/concurrent/executor/abstract_executor_service.rb

@@ -0,0 +1,136 @@
+require 'concurrent/errors'
+require 'concurrent/executor/executor_service'
+require 'concurrent/synchronization/object'
+require 'concurrent/utility/at_exit'
+
+module Concurrent
+
+  # @!macro abstract_executor_service_public_api
+  # @!visibility private
+  class AbstractExecutorService < Synchronization::Object
+    include ExecutorService
+
+    # The set of possible fallback policies that may be set at thread pool creation.
+    FALLBACK_POLICIES = [:abort, :discard, :caller_runs].freeze
+
+    # @!macro executor_service_attr_reader_fallback_policy
+    attr_reader :fallback_policy
+
+    # Create a new thread pool.
+    def initialize(*args, &block)
+      super(&nil)
+      synchronize { ns_initialize(*args, &block) }
+    end
+
+    # @!macro executor_service_method_shutdown
+    def shutdown
+      raise NotImplementedError
+    end
+
+    # @!macro executor_service_method_kill
+    def kill
+      raise NotImplementedError
+    end
+
+    # @!macro executor_service_method_wait_for_termination
+    def wait_for_termination(timeout = nil)
+      raise NotImplementedError
+    end
+
+    # @!macro executor_service_method_running_question
+    def running?
+      synchronize { ns_running? }
+    end
+
+    # @!macro executor_service_method_shuttingdown_question
+    def shuttingdown?
+      synchronize { ns_shuttingdown? }
+    end
+
+    # @!macro executor_service_method_shutdown_question
+    def shutdown?
+      synchronize { ns_shutdown? }
+    end
+
+    # @!macro executor_service_method_auto_terminate_question
+    def auto_terminate?
+      synchronize { ns_auto_terminate? }
+    end
+
+    # @!macro executor_service_method_auto_terminate_setter
+    def auto_terminate=(value)
+      synchronize { self.ns_auto_terminate = value }
+    end
+
+    protected
+
+    # Handler which executes the `fallback_policy` once the queue size
+    # reaches `max_queue`.
+    #
+    # @param [Array] args the arguments to the task which is being handled.
+    #
+    # @!visibility private
+    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
+
+    # @!macro [attach] executor_service_method_shutdown_execution
+    #
+    #   Callback method called when an orderly shutdown has completed.
+    #   The default behavior is to signal all waiting threads.
+    def shutdown_execution
+      # do nothing
+    end
+
+    # @!macro [attach] executor_service_method_kill_execution
+    #
+    #   Callback method called when the executor has been killed.
+    #   The default behavior is to do nothing.
+    def kill_execution
+      # do nothing
+    end
+
+    protected
+
+    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
+end