123456789_123456789_123456789_123456789_123456789_

Class: Concurrent::SimpleExecutorService

Overview

Note:

Intended for use primarily in testing and debugging.

An executor service in which every operation spawns a new, independently operating thread.

This is perhaps the most inefficient executor service in this library. It exists mainly for testing an debugging. Thread creation and management is expensive in Ruby and this executor performs no resource pooling. This can be very beneficial during testing and debugging because it decouples the using code from the underlying executor implementation. In production this executor will likely lead to suboptimal performance.

Constant Summary

Concern::Logging - Included

SEV_LABEL

AbstractExecutorService - Inherited

FALLBACK_POLICIES

Class Method Summary

RubyExecutorService - Inherited

AbstractExecutorService - Inherited

.new

Create a new thread pool.

Instance Attribute Summary

RubyExecutorService - Inherited

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

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

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

Constructor Details

This class inherits a constructor from Concurrent::RubyExecutorService

Class 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/simple_executor_service.rb', line 34

def self.<<(task)
  post(&task)
  self
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/simple_executor_service.rb', line 24

def self.post(*args)
  raise ArgumentError.new('no block given') unless block_given?
  Thread.new(*args) do
    Thread.current.abort_on_exception = false
    yield(*args)
  end
  true
end

Instance Attribute Details

#running?Boolean (readonly)

Is the executor running?

Returns:

  • (Boolean)

    true when running, false when shutting down or shutdown

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/executor/simple_executor_service.rb', line 62

def running?
  @running.true?
end

#shutdown?Boolean (readonly)

Is the executor shutdown?

Returns:

  • (Boolean)

    true when shutdown, false when shutting down or running

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/executor/simple_executor_service.rb', line 72

def shutdown?
  @stopped.set?
end

#shuttingdown?Boolean (readonly)

Is the executor shuttingdown?

Returns:

  • (Boolean)

    true when not running and not shutdown, else false

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/executor/simple_executor_service.rb', line 67

def shuttingdown?
  @running.false? && ! @stopped.set?
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/simple_executor_service.rb', line 56

def <<(task)
  post(&task)
  self
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/simple_executor_service.rb', line 84

def kill
  @running.make_false
  @stopped.set
  true
end

#ns_initialize(*args) (private)

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/executor/simple_executor_service.rb', line 97

def ns_initialize(*args)
  @running = Concurrent::AtomicBoolean.new(true)
  @stopped = Concurrent::Event.new
  @count = Concurrent::AtomicFixnum.new(0)
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/simple_executor_service.rb', line 40

def post(*args, &task)
  raise ArgumentError.new('no block given') unless block_given?
  return false unless running?
  @count.increment
  Thread.new(*args) do
    Thread.current.abort_on_exception = false
    begin
      yield(*args)
    ensure
      @count.decrement
      @stopped.set if @running.false? && @count.value == 0
    end
  end
end

#shutdown (readonly)

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/simple_executor_service.rb', line 77

def shutdown
  @running.make_false
  @stopped.set if @count.value == 0
  true
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/simple_executor_service.rb', line 91

def wait_for_termination(timeout = nil)
  @stopped.wait(timeout)
end