Add Concurrent.executor as an easy access point to global executors

Author: pitr-ch

.gitignore

@@ -35,4 +35,3 @@ ext/**/*.bundle
 ext/**/*.so
 ext/**/*.jar
 pkg
-*.gem

lib/concurrent/agent.rb

@@ -2,7 +2,6 @@
 
 require 'concurrent/dereferenceable'
 require 'concurrent/observable'
-require 'concurrent/executor/executor_options'
 require 'concurrent/utility/timeout'
 require 'concurrent/logging'
 
@@ -15,7 +14,6 @@ module Concurrent
   class Agent
     include Dereferenceable
     include Observable
-    include ExecutorOptions
     include Logging
 
     attr_reader :timeout, :io_executor, :fast_executor
@@ -42,8 +40,8 @@ def initialize(initial, opts = {})
       @validator            = Proc.new { |result| true }
       self.observers        = CopyOnWriteObserverSet.new
       @serialized_execution = SerializedExecution.new
-      @io_executor          = get_executor_from(opts) || Concurrent.global_io_executor
-      @fast_executor        = get_executor_from(opts) || Concurrent.global_fast_executor
+      @io_executor          = Executor.executor_from_options(opts) || Concurrent.global_io_executor
+      @fast_executor        = Executor.executor_from_options(opts) || Concurrent.global_fast_executor
       init_mutex
       set_deref_options(opts)
     end

lib/concurrent/configuration.rb

@@ -105,6 +105,16 @@ def self.global_timer_set
     GLOBAL_TIMER_SET.value
   end
 
+  # General access point to global executors.
+  # @param [Symbol, Executor] maps symbols:
+  #   - :fast - {Concurrent.global_fast_executor}
+  #   - :io - {Concurrent.global_io_executor}
+  #   - :immediate - {Concurrent.global_immediate_executor}
+  # @return [Executor]
+  def self.executor(executor_identifier)
+    Executor.executor(executor_identifier)
+  end
+
   def self.new_fast_executor(opts = {})
     FixedThreadPool.new(
         [2, Concurrent.processor_count].max,

lib/concurrent/delay.rb

@@ -1,7 +1,6 @@
 require 'thread'
 require 'concurrent/configuration'
 require 'concurrent/obligation'
-require 'concurrent/executor/executor_options'
 require 'concurrent/executor/immediate_executor'
 
 module Concurrent
@@ -39,7 +38,6 @@ module Concurrent
   # @see Concurrent::Dereferenceable
   class Delay
     include Obligation
-    include ExecutorOptions
 
     # NOTE: Because the global thread pools are lazy-loaded with these objects
     # there is a performance hit every time we post a new task to one of these
@@ -61,7 +59,7 @@ def initialize(opts = {}, &block)
 
       init_obligation
       set_deref_options(opts)
-      @task_executor = get_executor_from(opts)
+      @task_executor = Executor.executor_from_options(opts)
 
       @task      = block
       @state     = :pending

lib/concurrent/executor/executor.rb

@@ -6,6 +6,65 @@
 module Concurrent
 
   module Executor
+    include Logging
+
+    # Get the requested `Executor` based on the values set in the options hash.
+    #
+    # @param [Hash] opts the options defining the requested executor
+    # @option opts [Executor] :executor when set use the given `Executor` instance.
+    #   Three special values are also supported: `:fast` returns the global fast executor,
+    #   `:io` returns the global io executor, and `:immediate` returns a new
+    #   `ImmediateExecutor` object.
+    #
+    # @return [Executor, nil] the requested thread pool, or nil when no option specified
+    #
+    # @!visibility private
+    def self.executor_from_options(opts = {}) # :nodoc:
+      case
+      when opts.key?(:executor)
+        if opts[:executor].nil?
+          nil
+        else
+          executor(opts[:executor])
+        end
+      when opts.key?(:operation) || opts.key?(:task)
+        if opts[:operation] == true || opts[:task] == false
+          Kernel.warn '[DEPRECATED] use `executor: :fast` instead'
+          return Concurrent.global_fast_executor
+        end
+
+        if opts[:operation] == false || opts[:task] == true
+          Kernel.warn '[DEPRECATED] use `executor: :io` instead'
+          return Concurrent.global_io_executor
+        end
+
+        raise ArgumentError.new("executor '#{opts[:executor]}' not recognized")
+      else
+        nil
+      end
+    end
+
+    def self.executor(executor_identifier)
+      case executor_identifier
+      when :fast
+        Concurrent.global_fast_executor
+      when :io
+        Concurrent.global_io_executor
+      when :immediate
+        Concurrent.global_immediate_executor
+      when :operation
+        Kernel.warn '[DEPRECATED] use `executor: :fast` instead'
+        Concurrent.global_fast_executor
+      when :task
+        Kernel.warn '[DEPRECATED] use `executor: :io` instead'
+        Concurrent.global_io_executor
+      when Executor
+        executor_identifier
+      else
+        raise ArgumentError, "executor not recognized by '#{executor_identifier}'"
+      end
+    end
+
     # The policy defining how rejected tasks (tasks received once the
     # queue size reaches the configured `max_queue`, or after the
     # executor has shut down) are handled. Must be one of the values
@@ -363,4 +422,6 @@ def synchronize
       end
     end
   end
+
+
 end

lib/concurrent/executor/executor_options.rb

@@ -187,13 +187,11 @@ def ns_shutdown_execution
 
     # @api private
     def ns_kill_execution
-      ns_shutdown_execution
-      unless stopped_event.wait(1)
-        @pool.each &:kill
-        @pool.clear
-        @ready.clear
-        # TODO log out unprocessed tasks in queue
-      end
+      # TODO log out unprocessed tasks in queue
+      # TODO try to shutdown first?
+      @pool.each &:kill
+      @pool.clear
+      @ready.clear
     end
 
     alias_method :kill_execution, :ns_kill_execution

lib/concurrent/executor/ruby_thread_pool_executor.rb

@@ -4,7 +4,6 @@
 require 'concurrent/executor/executor'
 require 'concurrent/executor/single_thread_executor'
 require 'concurrent/utility/monotonic_time'
-require 'concurrent/executor/executor_options'
 
 module Concurrent
 
@@ -15,7 +14,6 @@ module Concurrent
   # @!macro monotonic_clock_warning
   class TimerSet
     include RubyExecutor
-    include ExecutorOptions
 
     # Create a new set of timed tasks.
     #
@@ -28,7 +26,7 @@ class TimerSet
     #     `ImmediateExecutor` object.
     def initialize(opts = {})
       @queue          = PriorityQueue.new(order: :min)
-      @task_executor  = get_executor_from(opts) || Concurrent.global_io_executor
+      @task_executor  = Executor.executor_from_options(opts) || Concurrent.global_io_executor
       @timer_executor = SingleThreadExecutor.new
       @condition      = Condition.new
       init_executor

lib/concurrent/executor/timer_set.rb

@@ -5,7 +5,8 @@ module Concurrent
   edge_lib_files = Dir['lib/concurrent/actor.rb',
                        'lib/concurrent/actor/**/*.rb',
                        'lib/concurrent/channel.rb',
-                       'lib/concurrent/channel/**/*.rb'] & git_files
+                       'lib/concurrent/channel/**/*.rb',
+                       'lib/concurrent/next.rb'] & git_files
   core_lib_files = all_lib_files - edge_lib_files
 
   FILE_MAP = {

lib/concurrent/file_map.rb

@@ -2,7 +2,6 @@
 
 require 'concurrent/ivar'
 require 'concurrent/executor/safe_task_executor'
-require 'concurrent/executor/executor_options'
 
 module Concurrent
 
@@ -12,7 +11,6 @@ module Concurrent
   # @see http://clojuredocs.org/clojure_core/clojure.core/future Clojure's future function
   # @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Future.html java.util.concurrent.Future
   class Future < IVar
-    include ExecutorOptions
 
     # Create a new `Future` in the `:unscheduled` state.
     #
@@ -29,7 +27,7 @@ def initialize(opts = {}, &block)
       super(IVar::NO_VALUE, opts)
       @state = :unscheduled
       @task = block
-      @executor = get_executor_from(opts) || Concurrent.global_io_executor
+      @executor = Executor.executor_from_options(opts) || Concurrent.global_io_executor
       @args = get_arguments_from(opts)
     end