RubySingleThreadExecutor < RubyThreadPoolExecutor

Author: jdantonio

CHANGELOG.md

@@ -1,3 +1,7 @@
+### Upcoming Release v1.0.0.pre2 (TBD)
+
+* Simplification of `RubySingleThreadExecutor`
+
 ## Current Release v1.0.0.pre1 (19 Aug 2015)
 
 * Merged in the `thread_safe` gem

lib/concurrent/executor/java_single_thread_executor.rb

@@ -6,7 +6,6 @@
   module Concurrent
 
     # @!macro single_thread_executor
-    # @!macro thread_pool_options
     # @!macro abstract_executor_service_public_api
     # @!visibility private
     class JavaSingleThreadExecutor < JavaExecutorService

lib/concurrent/executor/ruby_single_thread_executor.rb

@@ -1,80 +1,22 @@
-require 'thread'
-require 'concurrent/executor/ruby_executor_service'
-require 'concurrent/executor/serial_executor_service'
+require 'concurrent/executor/ruby_thread_pool_executor'
 
 module Concurrent
 
   # @!macro single_thread_executor
-  # @!macro thread_pool_options
   # @!macro abstract_executor_service_public_api
   # @!visibility private
-  class RubySingleThreadExecutor < RubyExecutorService
-    include SerialExecutorService
+  class RubySingleThreadExecutor < RubyThreadPoolExecutor
 
     # @!macro single_thread_executor_method_initialize
     def initialize(opts = {})
-      super
-    end
-
-    private
-
-    def ns_initialize(opts)
-      @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)
-      self.auto_terminate = opts.fetch(:auto_terminate, true)
-    end
-
-    # @!visibility private
-    def ns_execute(*args, &task)
-      supervise
-      @queue << [args, task]
-    end
-
-    # @!visibility private
-    def ns_shutdown_execution
-      @queue << :stop
-      stopped_event.set unless alive?
-    end
-
-    # @!visibility private
-    def ns_kill_execution
-      @queue.clear
-      @thread.kill if alive?
-    end
-
-    # @!visibility private
-    def alive?
-      @thread && @thread.alive?
-    end
-
-    # @!visibility private
-    def supervise
-      @thread = new_worker_thread unless alive?
-    end
-
-    # @!visibility private
-    def new_worker_thread
-      Thread.new do
-        Thread.current.abort_on_exception = false
-        work
-      end
-    end
-
-    # @!visibility private
-    def work
-      loop do
-        task = @queue.pop
-        break if task == :stop
-        begin
-          task.last.call(*task.first)
-        rescue => ex
-          # let it fail
-          log DEBUG, ex
-        end
-      end
-      stopped_event.set
+      super(
+        min_threads: 1,
+        max_threads: 1,
+        max_queue: 0,
+        idletime: DEFAULT_THREAD_IDLETIMEOUT,
+        fallback_policy: opts.fetch(:fallback_policy, :discard),
+        auto_terminate: opts.fetch(:auto_terminate, true)
+      )
     end
   end
 end

lib/concurrent/executor/single_thread_executor.rb

@@ -16,25 +16,35 @@ module Concurrent
 
   # @!macro [attach] single_thread_executor
   #
-  #   A thread pool with a set number of threads. The number of threads in the pool
-  #   is set on construction and remains constant. When all threads are busy new
-  #   tasks `#post` to the thread pool are enqueued until a thread becomes available.
-  #   Should a thread crash for any reason the thread will immediately be removed
-  #   from the pool and replaced.
+  #   A thread pool with a single thread an unlimited queue. Should the thread
+  #   die for any reason it will be removed and replaced, thus ensuring that
+  #   the executor will always remain viable and available to process jobs.
   #
-  #   The API and behavior of this class are based on Java's `SingleThreadExecutor`
+  #   A common pattern for background processing is to create a single thread
+  #   on which an infinite loop is run. The thread's loop blocks on an input
+  #   source (perhaps blocking I/O or a queue) and processes each input as it
+  #   is received. This pattern has several issues. The thread itself is highly
+  #   susceptible to errors during processing. Also, the thread itself must be
+  #   constantly monitored and restarted should it die. `SingleThreadExecutor`
+  #   encapsulates all these bahaviors. The task processor is highly resilient
+  #   to errors from within tasks. Also, should the thread die it will
+  #   automatically be restarted.
+  #
+  #   The API and behavior of this class are based on Java's `SingleThreadExecutor`.
   #
-  # @!macro thread_pool_options
   # @!macro abstract_executor_service_public_api
   class SingleThreadExecutor < SingleThreadExecutorImplementation
 
     # @!macro [new] single_thread_executor_method_initialize
     #
     #   Create a new thread pool.
     #
-    #   @option opts [Symbol] :fallback_policy (:discard) the policy for
-    #     handling new tasks that are received when the queue size has
-    #     reached `max_queue` or after the executor has shut down
+    #   @option opts [Symbol] :fallback_policy (:discard) the policy for handling new
+    #     tasks that are received when the queue size has reached
+    #     `max_queue` or the executor has shut down
+    #
+    #   @raise [ArgumentError] if `:fallback_policy` is not one of the values specified
+    #     in `FALLBACK_POLICIES`
     #
     #   @see http://docs.oracle.com/javase/tutorial/essential/concurrency/pools.html
     #   @see http://docs.oracle.com/javase/7/docs/api/java/util/concurrent/Executors.html