Class: Concurrent::SingleThreadExecutor

Inherits:
SingleThreadExecutorImplementation
  • Object
show all
Defined in:
lib/concurrent-ruby/concurrent/executor/single_thread_executor.rb

Overview

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.

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 behaviors. 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`.

Instance Attribute Summary collapse

Instance Method Summary collapse

Instance Attribute Details

#fallback_policy ⇒ Symbol (readonly) 



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/concurrent-ruby/concurrent/executor/single_thread_executor.rb', line 37

class SingleThreadExecutor < SingleThreadExecutorImplementation

  # @!macro 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 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
  #   @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html

  # @!method initialize(opts = {})
  #   @!macro single_thread_executor_method_initialize
end

Instance Method Details

#<<(task) ⇒ self

Submit a task to the executor for asynchronous processing.



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/concurrent-ruby/concurrent/executor/single_thread_executor.rb', line 37

class SingleThreadExecutor < SingleThreadExecutorImplementation

  # @!macro 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 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
  #   @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html

  # @!method initialize(opts = {})
  #   @!macro single_thread_executor_method_initialize
end

#auto_terminate=(value) ⇒ Boolean

Deprecated.

Has no effect

Set the auto-terminate behavior for this executor.



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/concurrent-ruby/concurrent/executor/single_thread_executor.rb', line 37

class SingleThreadExecutor < SingleThreadExecutorImplementation

  # @!macro 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 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
  #   @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html

  # @!method initialize(opts = {})
  #   @!macro single_thread_executor_method_initialize
end

#auto_terminate? ⇒ Boolean

Is the executor auto-terminate when the application exits?



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/concurrent-ruby/concurrent/executor/single_thread_executor.rb', line 37

class SingleThreadExecutor < SingleThreadExecutorImplementation

  # @!macro 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 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
  #   @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html

  # @!method initialize(opts = {})
  #   @!macro single_thread_executor_method_initialize
end

#can_overflow? ⇒ Boolean

Does the task queue have a maximum size?



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/concurrent-ruby/concurrent/executor/single_thread_executor.rb', line 37

class SingleThreadExecutor < SingleThreadExecutorImplementation

  # @!macro 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 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
  #   @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html

  # @!method initialize(opts = {})
  #   @!macro single_thread_executor_method_initialize
end

#kill ⇒ Object

Begin an immediate shutdown. In-progress tasks will be allowed to complete but enqueued tasks will be dismissed and no new tasks will be accepted. Has no additional effect if the thread pool is not running.



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/concurrent-ruby/concurrent/executor/single_thread_executor.rb', line 37

class SingleThreadExecutor < SingleThreadExecutorImplementation

  # @!macro 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 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
  #   @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html

  # @!method initialize(opts = {})
  #   @!macro single_thread_executor_method_initialize
end

#post(*args) { ... } ⇒ Boolean

Submit a task to the executor for asynchronous processing.

Yields:

  • the asynchronous task to perform

Raises:

  • (ArgumentError) —

    if no task is given



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/concurrent-ruby/concurrent/executor/single_thread_executor.rb', line 37

class SingleThreadExecutor < SingleThreadExecutorImplementation

  # @!macro 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 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
  #   @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html

  # @!method initialize(opts = {})
  #   @!macro single_thread_executor_method_initialize
end

#running? ⇒ Boolean

Is the executor running?



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/concurrent-ruby/concurrent/executor/single_thread_executor.rb', line 37

class SingleThreadExecutor < SingleThreadExecutorImplementation

  # @!macro 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 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
  #   @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html

  # @!method initialize(opts = {})
  #   @!macro single_thread_executor_method_initialize
end

#serialized? ⇒ Boolean

Does this executor guarantee serialization of its operations?



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/concurrent-ruby/concurrent/executor/single_thread_executor.rb', line 37

class SingleThreadExecutor < SingleThreadExecutorImplementation

  # @!macro 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 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
  #   @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html

  # @!method initialize(opts = {})
  #   @!macro single_thread_executor_method_initialize
end

#shutdown ⇒ Object

Begin an orderly shutdown. Tasks already in the queue will be executed, but no new tasks will be accepted. Has no additional effect if the thread pool is not running.



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/concurrent-ruby/concurrent/executor/single_thread_executor.rb', line 37

class SingleThreadExecutor < SingleThreadExecutorImplementation

  # @!macro 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 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
  #   @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html

  # @!method initialize(opts = {})
  #   @!macro single_thread_executor_method_initialize
end

#shutdown? ⇒ Boolean

Is the executor shutdown?



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/concurrent-ruby/concurrent/executor/single_thread_executor.rb', line 37

class SingleThreadExecutor < SingleThreadExecutorImplementation

  # @!macro 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 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
  #   @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html

  # @!method initialize(opts = {})
  #   @!macro single_thread_executor_method_initialize
end

#shuttingdown? ⇒ Boolean

Is the executor shuttingdown?



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/concurrent-ruby/concurrent/executor/single_thread_executor.rb', line 37

class SingleThreadExecutor < SingleThreadExecutorImplementation

  # @!macro 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 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
  #   @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html

  # @!method initialize(opts = {})
  #   @!macro single_thread_executor_method_initialize
end

#wait_for_termination(timeout = nil) ⇒ Boolean

Note:

Does not initiate shutdown or termination. Either ‘shutdown` or `kill` must be called before this method (or on another thread).

Block until executor shutdown is complete or until ‘timeout` seconds have passed.



37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
 
# File 'lib/concurrent-ruby/concurrent/executor/single_thread_executor.rb', line 37

class SingleThreadExecutor < SingleThreadExecutorImplementation

  # @!macro 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 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
  #   @see http://docs.oracle.com/javase/8/docs/api/java/util/concurrent/ExecutorService.html

  # @!method initialize(opts = {})
  #   @!macro single_thread_executor_method_initialize
end