123456789_123456789_123456789_123456789_123456789_

Class: Concurrent::RubySingleThreadExecutor

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

Constant Summary

Concern::Logging - Included

SEV_LABEL

AbstractExecutorService - Inherited

FALLBACK_POLICIES

RubyThreadPoolExecutor - Inherited

DEFAULT_MAX_POOL_SIZE, DEFAULT_MAX_QUEUE_SIZE, DEFAULT_MIN_POOL_SIZE, DEFAULT_SYNCHRONOUS, DEFAULT_THREAD_IDLETIMEOUT

Class Method Summary

RubyThreadPoolExecutor - Inherited

.new

Create a new thread pool.

RubyExecutorService - Inherited

.new

AbstractExecutorService - Inherited

.new

Create a new thread pool.

Instance Attribute Summary

RubyThreadPoolExecutor - Inherited

#can_overflow?

Does the task queue have a maximum size?
#idletime

The number of seconds that a thread may be idle before being reclaimed.
#max_length

The maximum number of threads that may be created in the pool.
#max_queue

The maximum number of tasks that may be waiting in the work queue at any one time.
#min_length

The minimum number of threads that may be retained in the pool.
#synchronous

Whether or not a value of 0 for :max_queue option means the queue must perform direct hand-off or rather unbounded queue.
#ns_limited_queue?

RubyExecutorService - Inherited

#fallback_policy, #ns_running?, #ns_shutdown?, #ns_shuttingdown?

AbstractExecutorService - Inherited

#auto_terminate=

Set the auto-terminate behavior for this executor.
#auto_terminate?

Is the executor auto-terminate when the application exits?
#fallback_policy, #name,
#running?

Is the executor running?
#shutdown

Begin an orderly shutdown.
#shutdown?

Is the executor shutdown?
#shuttingdown?

Is the executor shuttingdown?
#ns_auto_terminate?

ExecutorService - Included

#can_overflow?

Does the task queue have a maximum size?
#serialized?

Does this executor guarantee serialization of its operations?

Instance Method Summary

RubyThreadPoolExecutor - Inherited

#active_count

The number of threads that are actively executing tasks.
#completed_task_count

The number of tasks that have been completed by the pool since construction.
#largest_length

The largest number of threads that have been created in the pool since construction.
#length

The number of threads currently in the pool.
#prune_pool

Prune the thread pool of unneeded threads.
#queue_length

The number of tasks in the queue awaiting execution.
#remaining_capacity

Number of tasks that may be enqueued before reaching max_queue and rejecting new tasks.
#scheduled_task_count

The number of tasks that have been scheduled for execution on the pool since construction.
#ns_add_busy_worker

creates new worker which has to receive work to do after it’s added.
#ns_assign_worker

tries to assign task to a worker, tries to get one from @ready or to create new one.
#ns_enqueue

tries to enqueue task.
#ns_execute, #ns_initialize, #ns_kill_execution,
#ns_prune_pool

try oldest worker if it is idle for enough time, it’s returned back at the start.
#ns_ready_worker

handle ready worker, giving it new job or assigning back to @ready.
#ns_remove_busy_worker

removes a worker which is not in not tracked in @ready.
#ns_reset_if_forked, #ns_shutdown_execution, #ns_worker_died, #ready_worker, #remove_busy_worker, #worker_died, #worker_task_completed

RubyExecutorService - Inherited

#<<

Submit a task to the executor for asynchronous processing.
#auto_terminate=

Set the auto-terminate behavior for this executor.
#auto_terminate?

Is the executor auto-terminate when the application exits?
#can_overflow?

Does the task queue have a maximum size?
#kill

Begin an immediate shutdown.
#post

Submit a task to the executor for asynchronous processing.
#running?

Is the executor running?
#serialized?

Does this executor guarantee serialization of its operations?
#shutdown

Begin an orderly shutdown.
#shutdown?

Is the executor shutdown?
#shuttingdown?

Is the executor shuttingdown?
#wait_for_termination

Block until executor shutdown is complete or until timeout seconds have passed.
#ns_shutdown_execution, #stop_event, #stopped_event

AbstractExecutorService - Inherited

#<<

Submit a task to the executor for asynchronous processing.
#can_overflow?

Does the task queue have a maximum size?
#kill

Begin an immediate shutdown.
#post

Submit a task to the executor for asynchronous processing.
#serialized?

Does this executor guarantee serialization of its operations?
#to_s,
#wait_for_termination

Block until executor shutdown is complete or until timeout seconds have passed.
#fallback_action

Returns an action which executes the #fallback_policy once the queue size reaches max_queue.
#ns_execute,
#ns_kill_execution

Callback method called when the executor has been killed.
#ns_shutdown_execution

Callback method called when an orderly shutdown has completed.

Concern::Deprecation - Included

#deprecated, #deprecated_method

ExecutorService - Included

#<<

Submit a task to the executor for asynchronous processing.
#post

Submit a task to the executor for asynchronous processing.

Concern::Logging - Included

#log

Logs through global_logger, it can be overridden by setting @logger.

Synchronization::LockableObject - Inherited

#new_condition

Constructor Details

.new(opts = {}) ⇒ RubySingleThreadExecutor

Create a new thread pool.

Parameters:

  • opts (Hash) (defaults to: {}) —

    a customizable set of options

Options Hash (opts):

  • :fallback_policy (Symbol) — default: :discard —

    the policy for handling new tasks that are received when the queue size has reached max_queue or the executor has shut down

Raises:

  • (ArgumentError) —

    if :fallback_policy is not one of the values specified in FALLBACK_POLICIES

See Also:

[ GitHub ] 

  


# File 'lib/concurrent-ruby/concurrent/executor/ruby_single_thread_executor.rb', line 11



def initialize(opts = {})
  super(
    min_threads: 1,
    max_threads: 1,
    max_queue: 0,
    idletime: DEFAULT_THREAD_IDLETIMEOUT,
    fallback_policy: opts.fetch(:fallback_policy, :discard),
  )
end


  







Instance Attribute Details



  

    #fallback_policy  ⇒ Symbol  (readonly)
  



  

    
  





  
Returns:


    
  
  • 
    (Symbol)
—    
    

    The fallback policy in effect. Either :abort, :discard, or :caller_runs.
    

    
  
    • 





  [ GitHub ]


  

  


# File 'lib/concurrent-ruby/concurrent/executor/ruby_single_thread_executor.rb', line 8



class RubySingleThreadExecutor < RubyThreadPoolExecutor

  # @!macro single_thread_executor_method_initialize
  def initialize(opts = {})
    super(
      min_threads: 1,
      max_threads: 1,
      max_queue: 0,
      idletime: DEFAULT_THREAD_IDLETIMEOUT,
      fallback_policy: opts.fetch(:fallback_policy, :discard),
    )
  end
end


  







Instance Method Details



  

    #<<(task)  ⇒ self 
  



  

    

Submit a task to the executor for asynchronous processing.


  





  
Parameters:


    
  
  • 
    task
    (Proc)
—    
    

    the asynchronous task to perform
    

    
  
    • 



Returns:


    
  
  • 
    (self)
—    
    

    returns itself
    

    
  
    • 





  [ GitHub ]


  

  


# File 'lib/concurrent-ruby/concurrent/executor/ruby_single_thread_executor.rb', line 8



class RubySingleThreadExecutor < RubyThreadPoolExecutor

  # @!macro single_thread_executor_method_initialize
  def initialize(opts = {})
    super(
      min_threads: 1,
      max_threads: 1,
      max_queue: 0,
      idletime: DEFAULT_THREAD_IDLETIMEOUT,
      fallback_policy: opts.fetch(:fallback_policy, :discard),
    )
  end
end


  








  

    #auto_terminate=(value)  ⇒ Boolean 
  



  

    
Deprecated. 


Has no effect





Set the auto-terminate behavior for this executor.


  





  
Parameters:


    
  
  • 
    value
    (Boolean)
—    
    

    The new auto-terminate value to set for this executor.
    

    
  
    • 



Returns:


    
  
  • 
    (Boolean)
—    
    

    true when auto-termination is enabled else false.
    

    
  
    • 





  [ GitHub ]


  

  


# File 'lib/concurrent-ruby/concurrent/executor/ruby_single_thread_executor.rb', line 8



class RubySingleThreadExecutor < RubyThreadPoolExecutor

  # @!macro single_thread_executor_method_initialize
  def initialize(opts = {})
    super(
      min_threads: 1,
      max_threads: 1,
      max_queue: 0,
      idletime: DEFAULT_THREAD_IDLETIMEOUT,
      fallback_policy: opts.fetch(:fallback_policy, :discard),
    )
  end
end


  








  

    #auto_terminate?  ⇒ Boolean 
  



  

    

Is the executor auto-terminate when the application exits?


  





  
Returns:


    
  
  • 
    (Boolean)
—    
    

    true when auto-termination is enabled else false.
    

    
  
    • 





  [ GitHub ]


  

  


# File 'lib/concurrent-ruby/concurrent/executor/ruby_single_thread_executor.rb', line 8



class RubySingleThreadExecutor < RubyThreadPoolExecutor

  # @!macro single_thread_executor_method_initialize
  def initialize(opts = {})
    super(
      min_threads: 1,
      max_threads: 1,
      max_queue: 0,
      idletime: DEFAULT_THREAD_IDLETIMEOUT,
      fallback_policy: opts.fetch(:fallback_policy, :discard),
    )
  end
end


  








  

    #can_overflow?  ⇒ Boolean 
  



  

    

Does the task queue have a maximum size?


  





  
Returns:


    
  
  • 
    (Boolean)
—    
    

    True if the task queue has a maximum size else false.
    

    
  
    • 





  [ GitHub ]


  

  


# File 'lib/concurrent-ruby/concurrent/executor/ruby_single_thread_executor.rb', line 8



class RubySingleThreadExecutor < RubyThreadPoolExecutor

  # @!macro single_thread_executor_method_initialize
  def initialize(opts = {})
    super(
      min_threads: 1,
      max_threads: 1,
      max_queue: 0,
      idletime: DEFAULT_THREAD_IDLETIMEOUT,
      fallback_policy: opts.fetch(:fallback_policy, :discard),
    )
  end
end


  








  

    #kill  
  



  

    

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.


  



  [ GitHub ]


  

  


# File 'lib/concurrent-ruby/concurrent/executor/ruby_single_thread_executor.rb', line 8



class RubySingleThreadExecutor < RubyThreadPoolExecutor

  # @!macro single_thread_executor_method_initialize
  def initialize(opts = {})
    super(
      min_threads: 1,
      max_threads: 1,
      max_queue: 0,
      idletime: DEFAULT_THREAD_IDLETIMEOUT,
      fallback_policy: opts.fetch(:fallback_policy, :discard),
    )
  end
end


  








  

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



  

    

Submit a task to the executor for asynchronous processing.


  





  
Parameters:


    
  
  • 
    args
    (Array)
—    
    

    zero or more arguments to be passed to the task
    

    
  
    • 



Yields:


    
  
  • 
    
    
    

    the asynchronous task to perform
    

    
  
    • 



Returns:


    
  
  • 
    (Boolean)
—    
    

    true if the task is queued, false if the executor is not running
    

    
  
    • 



Raises:


    
  
  • 
    (ArgumentError)
—    
    

    if no task is given
    

    
  
    • 





  [ GitHub ]


  

  


# File 'lib/concurrent-ruby/concurrent/executor/ruby_single_thread_executor.rb', line 8



class RubySingleThreadExecutor < RubyThreadPoolExecutor

  # @!macro single_thread_executor_method_initialize
  def initialize(opts = {})
    super(
      min_threads: 1,
      max_threads: 1,
      max_queue: 0,
      idletime: DEFAULT_THREAD_IDLETIMEOUT,
      fallback_policy: opts.fetch(:fallback_policy, :discard),
    )
  end
end


  








  

    #running?  ⇒ Boolean 
  



  

    

Is the executor running?


  





  
Returns:


    
  
  • 
    (Boolean)
—    
    

    true when running, false when shutting down or shutdown
    

    
  
    • 





  [ GitHub ]


  

  


# File 'lib/concurrent-ruby/concurrent/executor/ruby_single_thread_executor.rb', line 8



class RubySingleThreadExecutor < RubyThreadPoolExecutor

  # @!macro single_thread_executor_method_initialize
  def initialize(opts = {})
    super(
      min_threads: 1,
      max_threads: 1,
      max_queue: 0,
      idletime: DEFAULT_THREAD_IDLETIMEOUT,
      fallback_policy: opts.fetch(:fallback_policy, :discard),
    )
  end
end


  








  

    #serialized?  ⇒ Boolean 
  



  

    

Does this executor guarantee serialization of its operations?


  





  
Returns:


    
  
  • 
    (Boolean)
—    
    

    True if the executor guarantees that all operations will be post in the order they are received and no two operations may occur simultaneously. Else false.
    

    
  
    • 





  [ GitHub ]


  

  


# File 'lib/concurrent-ruby/concurrent/executor/ruby_single_thread_executor.rb', line 8



class RubySingleThreadExecutor < RubyThreadPoolExecutor

  # @!macro single_thread_executor_method_initialize
  def initialize(opts = {})
    super(
      min_threads: 1,
      max_threads: 1,
      max_queue: 0,
      idletime: DEFAULT_THREAD_IDLETIMEOUT,
      fallback_policy: opts.fetch(:fallback_policy, :discard),
    )
  end
end


  








  

    #shutdown  
  



  

    

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.


  



  [ GitHub ]


  

  


# File 'lib/concurrent-ruby/concurrent/executor/ruby_single_thread_executor.rb', line 8



class RubySingleThreadExecutor < RubyThreadPoolExecutor

  # @!macro single_thread_executor_method_initialize
  def initialize(opts = {})
    super(
      min_threads: 1,
      max_threads: 1,
      max_queue: 0,
      idletime: DEFAULT_THREAD_IDLETIMEOUT,
      fallback_policy: opts.fetch(:fallback_policy, :discard),
    )
  end
end


  








  

    #shutdown?  ⇒ Boolean 
  



  

    

Is the executor shutdown?


  





  
Returns:


    
  
  • 
    (Boolean)
—    
    

    true when shutdown, false when shutting down or running
    

    
  
    • 





  [ GitHub ]


  

  


# File 'lib/concurrent-ruby/concurrent/executor/ruby_single_thread_executor.rb', line 8



class RubySingleThreadExecutor < RubyThreadPoolExecutor

  # @!macro single_thread_executor_method_initialize
  def initialize(opts = {})
    super(
      min_threads: 1,
      max_threads: 1,
      max_queue: 0,
      idletime: DEFAULT_THREAD_IDLETIMEOUT,
      fallback_policy: opts.fetch(:fallback_policy, :discard),
    )
  end
end


  








  

    #shuttingdown?  ⇒ Boolean 
  



  

    

Is the executor shuttingdown?


  





  
Returns:


    
  
  • 
    (Boolean)
—    
    

    true when not running and not shutdown, else false
    

    
  
    • 





  [ GitHub ]


  

  


# File 'lib/concurrent-ruby/concurrent/executor/ruby_single_thread_executor.rb', line 8



class RubySingleThreadExecutor < RubyThreadPoolExecutor

  # @!macro single_thread_executor_method_initialize
  def initialize(opts = {})
    super(
      min_threads: 1,
      max_threads: 1,
      max_queue: 0,
      idletime: DEFAULT_THREAD_IDLETIMEOUT,
      fallback_policy: opts.fetch(:fallback_policy, :discard),
    )
  end
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.


  





  
Parameters:


    
  
  • 
    timeout
    (Integer)
    (defaults to: nil)
—    
    

    the maximum number of seconds to wait for shutdown to complete
    

    
  
    • 



Returns:


    
  
  • 
    (Boolean)
—    
    

    true if shutdown complete or false on timeout
    

    
  
    • 





  [ GitHub ]


  

  


# File 'lib/concurrent-ruby/concurrent/executor/ruby_single_thread_executor.rb', line 8



class RubySingleThreadExecutor < RubyThreadPoolExecutor

  # @!macro single_thread_executor_method_initialize
  def initialize(opts = {})
    super(
      min_threads: 1,
      max_threads: 1,
      max_queue: 0,
      idletime: DEFAULT_THREAD_IDLETIMEOUT,
      fallback_policy: opts.fetch(:fallback_policy, :discard),
    )
  end
end