123456789_123456789_123456789_123456789_123456789_

Class: Concurrent::RubyExecutorService

Constant Summary

Concern::Logging - Included

SEV_LABEL

AbstractExecutorService - Inherited

FALLBACK_POLICIES

Class Method Summary

AbstractExecutorService - Inherited

.new

Create a new thread pool.

Instance Attribute Summary

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

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

.new(*args, &block) ⇒ RubyExecutorService

[ GitHub ]

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

def initialize(*args, &block)
  super
  @StopEvent    = Event.new
  @StoppedEvent = Event.new
end

Instance Attribute Details

#fallback_policySymbol (readonly)

Returns:

  • (Symbol)

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

[ GitHub ]

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

class RubyExecutorService < AbstractExecutorService
  safe_initialization!

  def initialize(*args, &block)
    super
    @StopEvent    = Event.new
    @StoppedEvent = Event.new
  end

  def post(*args, &task)
    raise ArgumentError.new('no block given') unless block_given?
    deferred_action = synchronize {
      if running?
        ns_execute(*args, &task)
      else
        fallback_action(*args, &task)
      end
    }
    if deferred_action
      deferred_action.call
    else
      true
    end
  end

  def shutdown
    synchronize do
      break unless running?
      stop_event.set
      ns_shutdown_execution
    end
    true
  end

  def kill
    synchronize do
      break if shutdown?
      stop_event.set
      ns_kill_execution
      stopped_event.set
    end
    true
  end

  def wait_for_termination(timeout = nil)
    stopped_event.wait(timeout)
  end

  private

  def stop_event
    @StopEvent
  end

  def stopped_event
    @StoppedEvent
  end

  def ns_shutdown_execution
    stopped_event.set
  end

  def ns_running?
    !stop_event.set?
  end

  def ns_shuttingdown?
    !(ns_running? || ns_shutdown?)
  end

  def ns_shutdown?
    stopped_event.set?
  end
end

#ns_running?Boolean (readonly, private)

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/executor/ruby_executor_service.rb', line 70

def ns_running?
  !stop_event.set?
end

#ns_shutdown?Boolean (readonly, private)

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/executor/ruby_executor_service.rb', line 78

def ns_shutdown?
  stopped_event.set?
end

#ns_shuttingdown?Boolean (readonly, private)

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/executor/ruby_executor_service.rb', line 74

def ns_shuttingdown?
  !(ns_running? || ns_shutdown?)
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_executor_service.rb', line 8

class RubyExecutorService < AbstractExecutorService
  safe_initialization!

  def initialize(*args, &block)
    super
    @StopEvent    = Event.new
    @StoppedEvent = Event.new
  end

  def post(*args, &task)
    raise ArgumentError.new('no block given') unless block_given?
    deferred_action = synchronize {
      if running?
        ns_execute(*args, &task)
      else
        fallback_action(*args, &task)
      end
    }
    if deferred_action
      deferred_action.call
    else
      true
    end
  end

  def shutdown
    synchronize do
      break unless running?
      stop_event.set
      ns_shutdown_execution
    end
    true
  end

  def kill
    synchronize do
      break if shutdown?
      stop_event.set
      ns_kill_execution
      stopped_event.set
    end
    true
  end

  def wait_for_termination(timeout = nil)
    stopped_event.wait(timeout)
  end

  private

  def stop_event
    @StopEvent
  end

  def stopped_event
    @StoppedEvent
  end

  def ns_shutdown_execution
    stopped_event.set
  end

  def ns_running?
    !stop_event.set?
  end

  def ns_shuttingdown?
    !(ns_running? || ns_shutdown?)
  end

  def ns_shutdown?
    stopped_event.set?
  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_executor_service.rb', line 8

class RubyExecutorService < AbstractExecutorService
  safe_initialization!

  def initialize(*args, &block)
    super
    @StopEvent    = Event.new
    @StoppedEvent = Event.new
  end

  def post(*args, &task)
    raise ArgumentError.new('no block given') unless block_given?
    deferred_action = synchronize {
      if running?
        ns_execute(*args, &task)
      else
        fallback_action(*args, &task)
      end
    }
    if deferred_action
      deferred_action.call
    else
      true
    end
  end

  def shutdown
    synchronize do
      break unless running?
      stop_event.set
      ns_shutdown_execution
    end
    true
  end

  def kill
    synchronize do
      break if shutdown?
      stop_event.set
      ns_kill_execution
      stopped_event.set
    end
    true
  end

  def wait_for_termination(timeout = nil)
    stopped_event.wait(timeout)
  end

  private

  def stop_event
    @StopEvent
  end

  def stopped_event
    @StoppedEvent
  end

  def ns_shutdown_execution
    stopped_event.set
  end

  def ns_running?
    !stop_event.set?
  end

  def ns_shuttingdown?
    !(ns_running? || ns_shutdown?)
  end

  def ns_shutdown?
    stopped_event.set?
  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_executor_service.rb', line 8

class RubyExecutorService < AbstractExecutorService
  safe_initialization!

  def initialize(*args, &block)
    super
    @StopEvent    = Event.new
    @StoppedEvent = Event.new
  end

  def post(*args, &task)
    raise ArgumentError.new('no block given') unless block_given?
    deferred_action = synchronize {
      if running?
        ns_execute(*args, &task)
      else
        fallback_action(*args, &task)
      end
    }
    if deferred_action
      deferred_action.call
    else
      true
    end
  end

  def shutdown
    synchronize do
      break unless running?
      stop_event.set
      ns_shutdown_execution
    end
    true
  end

  def kill
    synchronize do
      break if shutdown?
      stop_event.set
      ns_kill_execution
      stopped_event.set
    end
    true
  end

  def wait_for_termination(timeout = nil)
    stopped_event.wait(timeout)
  end

  private

  def stop_event
    @StopEvent
  end

  def stopped_event
    @StoppedEvent
  end

  def ns_shutdown_execution
    stopped_event.set
  end

  def ns_running?
    !stop_event.set?
  end

  def ns_shuttingdown?
    !(ns_running? || ns_shutdown?)
  end

  def ns_shutdown?
    stopped_event.set?
  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_executor_service.rb', line 8

class RubyExecutorService < AbstractExecutorService
  safe_initialization!

  def initialize(*args, &block)
    super
    @StopEvent    = Event.new
    @StoppedEvent = Event.new
  end

  def post(*args, &task)
    raise ArgumentError.new('no block given') unless block_given?
    deferred_action = synchronize {
      if running?
        ns_execute(*args, &task)
      else
        fallback_action(*args, &task)
      end
    }
    if deferred_action
      deferred_action.call
    else
      true
    end
  end

  def shutdown
    synchronize do
      break unless running?
      stop_event.set
      ns_shutdown_execution
    end
    true
  end

  def kill
    synchronize do
      break if shutdown?
      stop_event.set
      ns_kill_execution
      stopped_event.set
    end
    true
  end

  def wait_for_termination(timeout = nil)
    stopped_event.wait(timeout)
  end

  private

  def stop_event
    @StopEvent
  end

  def stopped_event
    @StoppedEvent
  end

  def ns_shutdown_execution
    stopped_event.set
  end

  def ns_running?
    !stop_event.set?
  end

  def ns_shuttingdown?
    !(ns_running? || ns_shutdown?)
  end

  def ns_shutdown?
    stopped_event.set?
  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_executor_service.rb', line 42

class RubyExecutorService < AbstractExecutorService
  safe_initialization!

  def initialize(*args, &block)
    super
    @StopEvent    = Event.new
    @StoppedEvent = Event.new
  end

  def post(*args, &task)
    raise ArgumentError.new('no block given') unless block_given?
    deferred_action = synchronize {
      if running?
        ns_execute(*args, &task)
      else
        fallback_action(*args, &task)
      end
    }
    if deferred_action
      deferred_action.call
    else
      true
    end
  end

  def shutdown
    synchronize do
      break unless running?
      stop_event.set
      ns_shutdown_execution
    end
    true
  end

  def kill
    synchronize do
      break if shutdown?
      stop_event.set
      ns_kill_execution
      stopped_event.set
    end
    true
  end

  def wait_for_termination(timeout = nil)
    stopped_event.wait(timeout)
  end

  private

  def stop_event
    @StopEvent
  end

  def stopped_event
    @StoppedEvent
  end

  def ns_shutdown_execution
    stopped_event.set
  end

  def ns_running?
    !stop_event.set?
  end

  def ns_shuttingdown?
    !(ns_running? || ns_shutdown?)
  end

  def ns_shutdown?
    stopped_event.set?
  end
end

#ns_shutdown_execution (private)

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/executor/ruby_executor_service.rb', line 66

def ns_shutdown_execution
  stopped_event.set
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_executor_service.rb', line 17

class RubyExecutorService < AbstractExecutorService
  safe_initialization!

  def initialize(*args, &block)
    super
    @StopEvent    = Event.new
    @StoppedEvent = Event.new
  end

  def post(*args, &task)
    raise ArgumentError.new('no block given') unless block_given?
    deferred_action = synchronize {
      if running?
        ns_execute(*args, &task)
      else
        fallback_action(*args, &task)
      end
    }
    if deferred_action
      deferred_action.call
    else
      true
    end
  end

  def shutdown
    synchronize do
      break unless running?
      stop_event.set
      ns_shutdown_execution
    end
    true
  end

  def kill
    synchronize do
      break if shutdown?
      stop_event.set
      ns_kill_execution
      stopped_event.set
    end
    true
  end

  def wait_for_termination(timeout = nil)
    stopped_event.wait(timeout)
  end

  private

  def stop_event
    @StopEvent
  end

  def stopped_event
    @StoppedEvent
  end

  def ns_shutdown_execution
    stopped_event.set
  end

  def ns_running?
    !stop_event.set?
  end

  def ns_shuttingdown?
    !(ns_running? || ns_shutdown?)
  end

  def ns_shutdown?
    stopped_event.set?
  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_executor_service.rb', line 8

class RubyExecutorService < AbstractExecutorService
  safe_initialization!

  def initialize(*args, &block)
    super
    @StopEvent    = Event.new
    @StoppedEvent = Event.new
  end

  def post(*args, &task)
    raise ArgumentError.new('no block given') unless block_given?
    deferred_action = synchronize {
      if running?
        ns_execute(*args, &task)
      else
        fallback_action(*args, &task)
      end
    }
    if deferred_action
      deferred_action.call
    else
      true
    end
  end

  def shutdown
    synchronize do
      break unless running?
      stop_event.set
      ns_shutdown_execution
    end
    true
  end

  def kill
    synchronize do
      break if shutdown?
      stop_event.set
      ns_kill_execution
      stopped_event.set
    end
    true
  end

  def wait_for_termination(timeout = nil)
    stopped_event.wait(timeout)
  end

  private

  def stop_event
    @StopEvent
  end

  def stopped_event
    @StoppedEvent
  end

  def ns_shutdown_execution
    stopped_event.set
  end

  def ns_running?
    !stop_event.set?
  end

  def ns_shuttingdown?
    !(ns_running? || ns_shutdown?)
  end

  def ns_shutdown?
    stopped_event.set?
  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_executor_service.rb', line 8

class RubyExecutorService < AbstractExecutorService
  safe_initialization!

  def initialize(*args, &block)
    super
    @StopEvent    = Event.new
    @StoppedEvent = Event.new
  end

  def post(*args, &task)
    raise ArgumentError.new('no block given') unless block_given?
    deferred_action = synchronize {
      if running?
        ns_execute(*args, &task)
      else
        fallback_action(*args, &task)
      end
    }
    if deferred_action
      deferred_action.call
    else
      true
    end
  end

  def shutdown
    synchronize do
      break unless running?
      stop_event.set
      ns_shutdown_execution
    end
    true
  end

  def kill
    synchronize do
      break if shutdown?
      stop_event.set
      ns_kill_execution
      stopped_event.set
    end
    true
  end

  def wait_for_termination(timeout = nil)
    stopped_event.wait(timeout)
  end

  private

  def stop_event
    @StopEvent
  end

  def stopped_event
    @StoppedEvent
  end

  def ns_shutdown_execution
    stopped_event.set
  end

  def ns_running?
    !stop_event.set?
  end

  def ns_shuttingdown?
    !(ns_running? || ns_shutdown?)
  end

  def ns_shutdown?
    stopped_event.set?
  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_executor_service.rb', line 33

class RubyExecutorService < AbstractExecutorService
  safe_initialization!

  def initialize(*args, &block)
    super
    @StopEvent    = Event.new
    @StoppedEvent = Event.new
  end

  def post(*args, &task)
    raise ArgumentError.new('no block given') unless block_given?
    deferred_action = synchronize {
      if running?
        ns_execute(*args, &task)
      else
        fallback_action(*args, &task)
      end
    }
    if deferred_action
      deferred_action.call
    else
      true
    end
  end

  def shutdown
    synchronize do
      break unless running?
      stop_event.set
      ns_shutdown_execution
    end
    true
  end

  def kill
    synchronize do
      break if shutdown?
      stop_event.set
      ns_kill_execution
      stopped_event.set
    end
    true
  end

  def wait_for_termination(timeout = nil)
    stopped_event.wait(timeout)
  end

  private

  def stop_event
    @StopEvent
  end

  def stopped_event
    @StoppedEvent
  end

  def ns_shutdown_execution
    stopped_event.set
  end

  def ns_running?
    !stop_event.set?
  end

  def ns_shuttingdown?
    !(ns_running? || ns_shutdown?)
  end

  def ns_shutdown?
    stopped_event.set?
  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_executor_service.rb', line 8

class RubyExecutorService < AbstractExecutorService
  safe_initialization!

  def initialize(*args, &block)
    super
    @StopEvent    = Event.new
    @StoppedEvent = Event.new
  end

  def post(*args, &task)
    raise ArgumentError.new('no block given') unless block_given?
    deferred_action = synchronize {
      if running?
        ns_execute(*args, &task)
      else
        fallback_action(*args, &task)
      end
    }
    if deferred_action
      deferred_action.call
    else
      true
    end
  end

  def shutdown
    synchronize do
      break unless running?
      stop_event.set
      ns_shutdown_execution
    end
    true
  end

  def kill
    synchronize do
      break if shutdown?
      stop_event.set
      ns_kill_execution
      stopped_event.set
    end
    true
  end

  def wait_for_termination(timeout = nil)
    stopped_event.wait(timeout)
  end

  private

  def stop_event
    @StopEvent
  end

  def stopped_event
    @StoppedEvent
  end

  def ns_shutdown_execution
    stopped_event.set
  end

  def ns_running?
    !stop_event.set?
  end

  def ns_shuttingdown?
    !(ns_running? || ns_shutdown?)
  end

  def ns_shutdown?
    stopped_event.set?
  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_executor_service.rb', line 8

class RubyExecutorService < AbstractExecutorService
  safe_initialization!

  def initialize(*args, &block)
    super
    @StopEvent    = Event.new
    @StoppedEvent = Event.new
  end

  def post(*args, &task)
    raise ArgumentError.new('no block given') unless block_given?
    deferred_action = synchronize {
      if running?
        ns_execute(*args, &task)
      else
        fallback_action(*args, &task)
      end
    }
    if deferred_action
      deferred_action.call
    else
      true
    end
  end

  def shutdown
    synchronize do
      break unless running?
      stop_event.set
      ns_shutdown_execution
    end
    true
  end

  def kill
    synchronize do
      break if shutdown?
      stop_event.set
      ns_kill_execution
      stopped_event.set
    end
    true
  end

  def wait_for_termination(timeout = nil)
    stopped_event.wait(timeout)
  end

  private

  def stop_event
    @StopEvent
  end

  def stopped_event
    @StoppedEvent
  end

  def ns_shutdown_execution
    stopped_event.set
  end

  def ns_running?
    !stop_event.set?
  end

  def ns_shuttingdown?
    !(ns_running? || ns_shutdown?)
  end

  def ns_shutdown?
    stopped_event.set?
  end
end

#stop_event (private)

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/executor/ruby_executor_service.rb', line 58

def stop_event
  @StopEvent
end

#stopped_event (private)

[ GitHub ]

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

def stopped_event
  @StoppedEvent
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_executor_service.rb', line 52

class RubyExecutorService < AbstractExecutorService
  safe_initialization!

  def initialize(*args, &block)
    super
    @StopEvent    = Event.new
    @StoppedEvent = Event.new
  end

  def post(*args, &task)
    raise ArgumentError.new('no block given') unless block_given?
    deferred_action = synchronize {
      if running?
        ns_execute(*args, &task)
      else
        fallback_action(*args, &task)
      end
    }
    if deferred_action
      deferred_action.call
    else
      true
    end
  end

  def shutdown
    synchronize do
      break unless running?
      stop_event.set
      ns_shutdown_execution
    end
    true
  end

  def kill
    synchronize do
      break if shutdown?
      stop_event.set
      ns_kill_execution
      stopped_event.set
    end
    true
  end

  def wait_for_termination(timeout = nil)
    stopped_event.wait(timeout)
  end

  private

  def stop_event
    @StopEvent
  end

  def stopped_event
    @StoppedEvent
  end

  def ns_shutdown_execution
    stopped_event.set
  end

  def ns_running?
    !stop_event.set?
  end

  def ns_shuttingdown?
    !(ns_running? || ns_shutdown?)
  end

  def ns_shutdown?
    stopped_event.set?
  end
end