123456789_123456789_123456789_123456789_123456789_

Class: Concurrent::Promise

Relationships & Source Files
Super Chains via Extension / Inclusion / Inheritance
Class Chain:
Instance Chain:
Inherits: Concurrent::IVar
Defined in: lib/concurrent-ruby/concurrent/promise.rb

Overview

Promises are inspired by the JavaScript Promises/A and / Promises/A+ specifications.

> A promise represents the eventual value returned from the single > completion of an operation.

Promises are similar to futures and share many of the same behaviours. Promises are far more robust, however. Promises can be chained in a tree structure where each promise may have zero or more children. Promises are chained using the #then method. The result of a call to #then is always another promise. Promises are resolved asynchronously (with respect to the main thread) but in a strict order: parents are guaranteed to be resolved before their children, children before their younger siblings. The #then method takes two parameters: an optional block to be executed upon parent resolution and an optional callable to be executed upon parent failure. The result of each promise is passed to each of its children upon resolution. When a promise is rejected all its children will be summarily rejected and will receive the reason.

Promises have several possible states: :unscheduled, :pending, :processing, :rejected, or :fulfilled. These are also aggregated as #incomplete? and #complete?. When a Promise is created it is set to :unscheduled. Once the #execute method is called the state becomes :pending. Once a job is pulled from the thread pool’s queue and is given to a thread for processing (often immediately upon #post) the state becomes :processing. The future will remain in this state until processing is complete. A future that is in the :unscheduled, :pending, or :processing is considered #incomplete?. A #complete? Promise is either :rejected, indicating that an exception was thrown during processing, or :fulfilled, indicating success. If a Promise is :fulfilled its #value will be updated to reflect the result of the operation. If :rejected the reason will be updated with a reference to the thrown exception. The predicate methods #unscheduled?, #pending?, #rejected?, and #fulfilled? can be called at any time to obtain the state of the Promise, as can the #state method, which returns a symbol.

Retrieving the value of a promise is done through the value (alias: deref) method. Obtaining the value of a promise is a potentially blocking operation. When a promise is rejected a call to value will return nil immediately. When a promise is fulfilled a call to value will immediately return the current value. When a promise is pending a call to value will block until the promise is either rejected or fulfilled. A timeout value can be passed to value to limit how long the call will block. If nil the call will block indefinitely. If 0 the call will not block. Any other integer or float value will indicate the maximum number of seconds to block.

Promises run on the global thread pool.

## Copy Options

::Object references in Ruby are mutable. This can lead to serious problems when the #value of an object is a mutable reference. Which is always the case unless the value is a Fixnum, Symbol, or similar “primitive” data type. Each instance can be configured with a few options that can help protect the program from potentially dangerous operations. Each of these options can be optionally set when the object instance is created:

  • :dup_on_deref When true the object will call the #dup method on the value object every time the #value method is called (default: false)

  • :freeze_on_deref When true the object will call the #freeze method on the value object every time the #value method is called (default: false)

  • :copy_on_deref When given a Proc object the Proc will be run every time the #value method is called. The Proc will be given the current value as its only argument and the result returned by the block will be the return value of the #value call. When nil this option will be ignored (default: nil)

When multiple deref options are set the order of operations is strictly defined. The order of deref operations is:

  • :copy_on_deref

  • :dup_on_deref

  • :freeze_on_deref

Because of this ordering there is no need to #freeze an object created by a provided :copy_on_deref block. Simply set :freeze_on_deref to true. Setting both :dup_on_deref to true and :freeze_on_deref to true is as close to the behavior of a “pure” functional language (like Erlang, Clojure, or Haskell) as we are likely to get in Ruby.

Examples

Start by requiring promises

“‘ruby require ’concurrent/promise’ “‘

Then create one

“‘ruby p = .execute do

  # do something
  42
end

“‘

Promises can be chained using the #then method. The #then method accepts a block and an executor, to be executed on fulfillment, and a callable argument to be executed on rejection. The result of the each promise is passed as the block argument to chained promises.

“‘ruby p = Promise.new10.then{|x| x * 2}.then{|result| result - 10 }.execute “`

And so on, and so on, and so on…

“‘ruby p = Promise.fulfill(20).

then{|result| result - 10 }.
then{|result| result * 3 }.
then(executor: different_executor){|result| result % 5 }.execute

“‘

The initial state of a newly created Promise depends on the state of its parent:

  • if parent is unscheduled the child will be unscheduled

  • if parent is pending the child will be pending

  • if parent is fulfilled the child will be pending

  • if parent is rejected the child will be pending (but will ultimately be rejected)

Promises are executed asynchronously from the main thread. By the time a child Promise finishes intialization it may be in a different state than its parent (by the time a child is created its parent may have completed execution and changed state). Despite being asynchronous, however, the order of execution of Promise objects in a chain (or tree) is strictly defined.

There are multiple ways to create and execute a new Promise. Both ways provide identical behavior:

“‘ruby

create, operate, then execute

p1 = Promise.new{ “Hello World!” } p1.state #=> :unscheduled p1.execute

create and immediately execute

p2 = Promise.new{ “Hello World!” }.execute

execute during creation

p3 = Promise.execute{ “Hello World!” } “‘

Once the .execute method is called a Promise becomes pending:

“‘ruby p = Promise.execute{ “Hello, world!” } p.state #=> :pending p.pending? #=> true “`

Wait a little bit, and the promise will resolve and provide a value:

“‘ruby p = Promise.execute{ “Hello, world!” } sleep(0.1)

p.state #=> :fulfilled p.fulfilled? #=> true p.value #=> “Hello, world!” “‘

If an exception occurs, the promise will be rejected and will provide a reason for the rejection:

“‘ruby p = Promise.execute{ raise StandardError.new(“Here comes the Boom!”) } sleep(0.1)

p.state #=> :rejected p.rejected? #=> true p.reason #=> “#<StandardError: Here comes the Boom!>” “‘

Rejection

When a promise is rejected all its children will be rejected and will receive the rejection reason as the rejection callable parameter:

“‘ruby p = .execute { Thread.pass; raise StandardError }

c1 = p.then(-> reason { 42 }) c2 = p.then(-> reason { raise ‘Boom!’ })

c1.wait.state #=> :fulfilled c1.value #=> 45 c2.wait.state #=> :rejected c2.reason #=> #<RuntimeError: Boom!> “‘

Once a promise is rejected it will continue to accept children that will receive immediately rejection (they will be executed asynchronously).

Aliases

The #then method is the most generic alias: it accepts a block to be executed upon parent fulfillment and a callable to be executed upon parent rejection. At least one of them should be passed. The default block is ‘{ |result| result }` that fulfills the child with the parent value. The default callable is `{ |reason| raise reason }` that rejects the child with the parent reason.

  • on_success { |result| ... } is the same as then {|result| ... }

  • rescue { |reason| ... } is the same as then(Proc.new { |reason| ... } )

  • #rescue is aliased by #catch and #on_error

Class Method Summary

IVar - Inherited

.new

Create a new IVar in the :pending state with the (optional) initial value.

Instance Attribute Summary

Concern::Obligation - Included

#complete?

Has the obligation completed processing?

#fulfilled?

Has the obligation been fulfilled?

#incomplete?

Is the obligation still awaiting completion of processing?

#pending?

Is obligation completion still pending?

#realized?
#rejected?

Has the obligation been rejected?

#state

The current state of the obligation.

#unscheduled?

Is the obligation still unscheduled?

#state=

Concern::Dereferenceable - Included

#value

Return the value this object represents after applying the options specified by the #set_deref_options method.

Instance Method Summary

IVar - Inherited

#add_observer

Add an observer on this object that will receive notification on update.

#fail

Set the IVar to failed due to some error and wake or notify all threads waiting on it.

#set

Set the IVar to a value and wake or notify all threads waiting on it.

#try_set

Attempt to set the IVar with the given value or block.

#complete, #complete_without_notification, #notify_observers, #ns_complete_without_notification, #ns_initialize, #safe_execute, #check_for_block_or_value!

Concern::Observable - Included

#add_observer

Adds an observer to this set.

#count_observers

Return the number of observers associated with this object.

#delete_observer

Remove observer as an observer on this object so that it will no longer receive notifications.

#delete_observers

Remove all observers associated with this object.

#with_observer

As #add_observer but can be used for chaining.

Concern::Obligation - Included

#exception,
#no_error!
#reason

If an exception was raised during processing this will return the exception object.

#value

The current value of the obligation.

#value!

The current value of the obligation.

#wait

Wait until obligation is complete or the timeout has been reached.

#wait!

Wait until obligation is complete or the timeout is reached.

#compare_and_set_state

Atomic compare and set operation State is set to next_state only if current state == expected_current.

#event, #get_arguments_from,
#if_state

Executes the block within mutex if current state is included in expected_states.

#init_obligation,
#ns_check_state?

Am I in the current state?

#ns_set_state, #set_state

Concern::Dereferenceable - Included

#deref
#apply_deref_options,
#ns_set_deref_options

Set the options which define the operations #value performs before returning data to the caller (dereferencing).

Synchronization::LockableObject - Inherited

Constructor Details

.new(opts = {}) { ... } ⇒ Promise

Initialize a new Promise with the provided options.

Parameters:

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

    the options used to define the behavior at update and deref and to specify the executor on which to perform actions

Options Hash (opts):

  • :executor (Executor)

    when set use the given Executor instance. Three special values are also supported: :io returns the global pool for long, blocking (IO) tasks, :fast returns the global pool for short, fast operations, and :immediate returns the global ImmediateExecutor object.

  • :dup_on_deref (Boolean) — default: false

    Call #dup before returning the data from #value

  • :freeze_on_deref (Boolean) — default: false

    Call #freeze before returning the data from #value

  • :copy_on_deref (Proc) — default: nil

    When calling the #value method, call the given proc passing the internal value as the sole argument then return the new value returned from the proc.

Yields:

  • The block operation to be performed asynchronously.

Raises:

  • (ArgumentError)

    if no block is given

See Also:

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 210

def initialize(opts = {}, &block)
  opts.delete_if { |k, v| v.nil? }
  super(NULL, opts.merge(__promise_body_from_block__: block), &nil)
end

Class Method Details

.all?(*promises) ⇒ Boolean

Aggregates a collection of promises and executes the #then condition if all aggregated promises succeed. Executes the #rescue handler with a PromiseExecutionError if any of the aggregated promises fail. Upon execution will execute any of the aggregate promises that were not already executed.

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 464

def self.all?(*promises)
  aggregate(:all?, *promises)
end

.any?(*promises) ⇒ Promise

Aggregates a collection of promises and executes the #then condition if any aggregated promises succeed. Executes the #rescue handler with a PromiseExecutionError if any of the aggregated promises fail. Upon execution will execute any of the aggregate promises that were not already executed.

The returned promise will not yet have been executed. Additional #then and #rescue handlers may still be provided. Once the returned promise is execute the aggregate promises will be also be executed (if they have not been executed already). The results of the aggregate promises will be checked upon completion. The necessary #then and #rescue blocks on the aggregating promise will then be executed as appropriate. If the #rescue handlers are executed the raises exception will be PromiseExecutionError.

Parameters:

  • promises (Array)

    Zero or more promises to aggregate

Returns:

  • (Promise)

    an unscheduled (not executed) promise that aggregates the promises given as arguments

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 475

def self.any?(*promises)
  aggregate(:any?, *promises)
end

.execute(opts = {}, &block) ⇒ Promise

Create a new Promise object with the given block, execute it, and return the :pending object.

Examples:

promise = Concurrent::Promise.execute{ sleep(1); 42 }
promise.state #=> :pending

Parameters:

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

    the options used to define the behavior at update and deref and to specify the executor on which to perform actions

Options Hash (opts):

  • :executor (Executor)

    when set use the given Executor instance. Three special values are also supported: :io returns the global pool for long, blocking (IO) tasks, :fast returns the global pool for short, fast operations, and :immediate returns the global ImmediateExecutor object.

  • :dup_on_deref (Boolean) — default: false

    Call #dup before returning the data from #value

  • :freeze_on_deref (Boolean) — default: false

    Call #freeze before returning the data from #value

  • :copy_on_deref (Proc) — default: nil

    When calling the #value method, call the given proc passing the internal value as the sole argument then return the new value returned from the proc.

  • :parent (Promise)

    the parent Promise when building a chain/tree

  • :on_fulfill (Proc)

    fulfillment handler

  • :on_reject (Proc)

    rejection handler

  • :args (object, Array)

    zero or more arguments to be passed the task block on execution

Returns:

  • (Promise)

    the newly created Promise in the :pending state

Raises:

  • (ArgumentError)

    if no block is given

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 296

def self.execute(opts = {}, &block)
  new(opts, &block).execute
end

.fulfill(value, opts = {}) ⇒ Promise

Create a new Promise and fulfill it immediately.

Parameters:

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

    the options used to define the behavior at update and deref and to specify the executor on which to perform actions

Options Hash (opts):

  • :executor (Executor)

    when set use the given Executor instance. Three special values are also supported: :io returns the global pool for long, blocking (IO) tasks, :fast returns the global pool for short, fast operations, and :immediate returns the global ImmediateExecutor object.

  • :dup_on_deref (Boolean) — default: false

    Call #dup before returning the data from #value

  • :freeze_on_deref (Boolean) — default: false

    Call #freeze before returning the data from #value

  • :copy_on_deref (Proc) — default: nil

    When calling the #value method, call the given proc passing the internal value as the sole argument then return the new value returned from the proc.

  • :parent (Promise)

    the parent Promise when building a chain/tree

  • :on_fulfill (Proc)

    fulfillment handler

  • :on_reject (Proc)

    rejection handler

  • :args (object, Array)

    zero or more arguments to be passed the task block on execution

Returns:

  • (Promise)

    the newly created Promise

Raises:

  • (ArgumentError)

    if no block is given

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 224

def self.fulfill(value, opts = {})
  Promise.new(opts).tap { |p| p.send(:synchronized_set_state!, true, value, nil) }
end

.reject(reason, opts = {}) ⇒ Promise

Create a new Promise and reject it immediately.

Parameters:

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

    the options used to define the behavior at update and deref and to specify the executor on which to perform actions

Options Hash (opts):

  • :executor (Executor)

    when set use the given Executor instance. Three special values are also supported: :io returns the global pool for long, blocking (IO) tasks, :fast returns the global pool for short, fast operations, and :immediate returns the global ImmediateExecutor object.

  • :dup_on_deref (Boolean) — default: false

    Call #dup before returning the data from #value

  • :freeze_on_deref (Boolean) — default: false

    Call #freeze before returning the data from #value

  • :copy_on_deref (Proc) — default: nil

    When calling the #value method, call the given proc passing the internal value as the sole argument then return the new value returned from the proc.

  • :parent (Promise)

    the parent Promise when building a chain/tree

  • :on_fulfill (Proc)

    fulfillment handler

  • :on_reject (Proc)

    rejection handler

  • :args (object, Array)

    zero or more arguments to be passed the task block on execution

Returns:

  • (Promise)

    the newly created Promise

Raises:

  • (ArgumentError)

    if no block is given

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 237

def self.reject(reason, opts = {})
  Promise.new(opts).tap { |p| p.send(:synchronized_set_state!, false, nil, reason) }
end

.zip(*promises) ⇒ Promise<Array> .zip(*promises, opts) ⇒ Promise<Array>

Builds a promise that produces the result of promises in an Array and fails if any of them fails.

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 409

def self.zip(*promises)
  opts = promises.last.is_a?(::Hash) ? promises.pop.dup : {}
  opts[:executor] ||= ImmediateExecutor.new
  zero = if !opts.key?(:execute) || opts.delete(:execute)
    fulfill([], opts)
  else
    Promise.new(opts) { [] }
  end

  promises.reduce(zero) do |p1, p2|
    p1.flat_map do |results|
      p2.then do |next_result|
        results << next_result
      end
    end
  end
end

Instance Attribute Details

#root?Boolean (readonly, private)

This method is for internal use only.
[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 528

def root? # :nodoc:
  @parent.nil?
end

Instance Method Details

#catch(&block)

Alias for #rescue.

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 364

alias_method :catch, :rescue

#complete(success, value, reason) (private)

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 551

def complete(success, value, reason)
  children_to_notify = synchronize do
    set_state!(success, value, reason)
    @children.dup
  end

  children_to_notify.each { |child| notify_child(child) }
  observers.notify_and_delete_observers{ [Time.now, self.value, reason] }
end

#executePromise

Execute an :unscheduled Promise. Immediately sets the state to :pending and passes the block to a new thread/thread pool for eventual execution. Does nothing if the Promise is in any state other than :unscheduled.

Returns:

  • (Promise)

    a reference to self

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 246

def execute
  if root?
    if compare_and_set_state(:pending, :unscheduled)
      set_pending
      realize(@promise_body)
    end
  else
    compare_and_set_state(:pending, :unscheduled)
    @parent.execute
  end
  self
end

#fail(reason = StandardError.new) ⇒ IVar

Set the IVar to failed due to some error and wake or notify all threads waiting on it.

Parameters:

  • reason (Object) (defaults to: StandardError.new)

    for the failure

Returns:

Raises:

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 278

def fail(reason = StandardError.new)
  set { raise reason }
end

#flat_map(&block) ⇒ Promise

Yield the successful result to the block that returns a promise. If that promise is also successful the result is the result of the yielded promise. If either part fails the whole also fails.

Examples:

Promise.execute { 1 }.flat_map { |v| Promise.execute { v + 2 } }.value! #=> 3
[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 375

def flat_map(&block)
  child = Promise.new(
    parent: self,
    executor: ImmediateExecutor.new,
  )

  on_error { |e| child.on_reject(e) }
  on_success do |result1|
    begin
      inner = block.call(result1)
      inner.execute
      inner.on_success { |result2| child.on_fulfill(result2) }
      inner.on_error { |e| child.on_reject(e) }
    rescue => e
      child.on_reject(e)
    end
  end

  child
end

#notify_child(child) (private)

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 545

def notify_child(child)
  if_state(:fulfilled) { child.on_fulfill(apply_deref_options(@value)) }
  if_state(:rejected) { child.on_reject(@reason) }
end

#on_error(&block)

Alias for #rescue.

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 365

alias_method :on_error, :rescue

#on_fulfill(result) (private)

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 533

def on_fulfill(result)
  realize Proc.new { @on_fulfill.call(result) }
  nil
end

#on_reject(reason) (private)

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 539

def on_reject(reason)
  realize Proc.new { @on_reject.call(reason) }
  nil
end

#on_success { ... } ⇒ Promise

Chain onto this promise an action to be undertaken on success (fulfillment).

Yields:

  • The block to execute

Returns:

  • (Promise)

    self

Raises:

  • (ArgumentError)
[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 349

def on_success(&block)
  raise ArgumentError.new('no block given') unless block_given?
  self.then(&block)
end

#realize(task) (private)

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 562

def realize(task)
  @executor.post do
    success, value, reason = SafeTaskExecutor.new(task, rescue_exception: true).execute(*@args)
    complete(success, value, reason)
  end
end

#rescue { ... } ⇒ Promise Also known as: #catch, #on_error

Chain onto this promise an action to be undertaken on failure (rejection).

Yields:

  • The block to execute

Returns:

  • (Promise)

    self

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 360

def rescue(&block)
  self.then(block)
end

#set(value = NULL) { ... } ⇒ IVar

Set the IVar to a value and wake or notify all threads waiting on it.

Parameters:

  • value (Object) (defaults to: NULL)

    the value to store in the IVar

Yields:

  • A block operation to use for setting the value

Returns:

Raises:

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 262

def set(value = NULL, &block)
  raise PromiseExecutionError.new('supported only on root promise') unless root?
  check_for_block_or_value!(block_given?, value)
  synchronize do
    if @state != :unscheduled
      raise MultipleAssignmentError
    else
      @promise_body = block || Proc.new { |result| value }
    end
  end
  execute
end

#set_pending (private)

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 520

def set_pending
  synchronize do
    @state = :pending
    @children.each { |c| c.set_pending }
  end
end

#set_state!(success, value, reason) (private)

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 570

def set_state!(success, value, reason)
  set_state(success, value, reason)
  event.set
end

#synchronized_set_state!(success, value, reason) (private)

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 576

def synchronized_set_state!(success, value, reason)
  synchronize { set_state!(success, value, reason) }
end

#then(rescuer, executor, &block) ⇒ Promise #then(rescuer, executor: executor, &block) ⇒ Promise

Chain a new promise off the current promise.

Yields:

  • The block operation to be performed asynchronously.

Returns:

  • (Promise)

    the new promise

Raises:

  • (ArgumentError)
[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 314

def then(*args, &block)
  if args.last.is_a?(::Hash)
    executor = args.pop[:executor]
    rescuer = args.first
  else
    rescuer, executor = args
  end

  executor ||= @executor

  raise ArgumentError.new('rescuers and block are both missing') if rescuer.nil? && !block_given?
  block = Proc.new { |result| result } unless block_given?
  child = Promise.new(
    parent: self,
    executor: executor,
    on_fulfill: block,
    on_reject: rescuer
  )

  synchronize do
    child.state = :pending if @state == :pending
    child.on_fulfill(apply_deref_options(@value)) if @state == :fulfilled
    child.on_reject(@reason) if @state == :rejected
    @children << child
  end

  child
end

#zip(*promises) ⇒ Promise<Array> #zip(*promises, opts) ⇒ Promise<Array>

Builds a promise that produces the result of self and others in an Array and fails if any of them fails.

[ GitHub ]

  
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 440

def zip(*others)
  self.class.zip(self, *others)
end