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 thevalue
object every time the#value
method is called (default: false) -
:freeze_on_deref
When true the object will call the#freeze
method on thevalue
object every time the#value
method is called (default: false) -
:copy_on_deref
When given aProc
object theProc
will be run every time the#value
method is called. TheProc
will be given the currentvalue
as its only argument and the result returned by the block will be the return value of the#value
call. Whennil
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.
Class Method Summary
-
.all?(*promises) ⇒ Boolean
Aggregates a collection of promises and executes the #then condition if all aggregated promises succeed.
-
.any?(*promises) ⇒ Promise
Aggregates a collection of promises and executes the #then condition if any aggregated promises succeed.
-
.execute(opts = {}, &block) ⇒ Promise
Create a new
Promise
object with the given block, execute it, and return the:pending
object. -
.fulfill(value, opts = {}) ⇒ Promise
Create a new
Promise
and fulfill it immediately. -
.new(opts = {}) { ... } ⇒ Promise
constructor
Initialize a new
Promise
with the provided options. -
.reject(reason, opts = {}) ⇒ Promise
Create a new
Promise
and reject it immediately. -
.zip(*promises) ⇒ Promise<Array>
Builds a promise that produces the result of promises in an
Array
and fails if any of them fails.
IVar
- Inherited
Instance Attribute Summary
- #root? ⇒ Boolean readonly private Internal use only
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? | Alias for Concern::Obligation#fulfilled?. |
#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 |
Instance Method Summary
-
#catch(&block)
Alias for #rescue.
-
#execute ⇒ Promise
Execute an
:unscheduled
Promise
. - #fail(reason = StandardError.new) ⇒ IVar
-
#flat_map(&block) ⇒ Promise
Yield the successful result to the block that returns a promise.
-
#on_error(&block)
Alias for #rescue.
-
#on_success { ... } ⇒ Promise
Chain onto this promise an action to be undertaken on success (fulfillment).
-
#rescue { ... } ⇒ Promise
(also: #catch, #on_error)
Chain onto this promise an action to be undertaken on failure (rejection).
- #set(value = NULL) { ... } ⇒ IVar
-
#then(rescuer, executor, &block) ⇒ Promise
Chain a new promise off the current promise.
-
#zip(*promises) ⇒ Promise<Array>
Builds a promise that produces the result of self and others in an
Array
and fails if any of them fails. - #complete(success, value, reason) private
- #notify_child(child) private
- #on_fulfill(result) private
- #on_reject(reason) private
- #realize(task) private
- #set_pending private
- #set_state!(success, value, reason) private
- #synchronized_set_state!(success, value, reason) private
IVar
- Inherited
#add_observer | Add an observer on this object that will receive notification on update. |
#fail | |
#set | |
#try_set | Attempt to set the |
#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 |
#delete_observers | Remove all observers associated with this object. |
#with_observer | As |
Concern::Obligation
- Included
#exception, | |
#no_error! | Alias for Concern::Obligation#wait!. |
#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 |
#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 | Alias for Concern::Dereferenceable#value. |
#apply_deref_options, | |
#ns_set_deref_options |
|
Synchronization::LockableObject
- Inherited
Constructor Details
.new(opts = {}) { ... } ⇒ Promise
Initialize a new Promise
with the provided options.
# 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.
# 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.
# 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.
# 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.
# 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.
# 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
Builds a promise that produces the result of promises in an Array
and fails if any of them fails.
# 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)
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 528
def root? # :nodoc: @parent.nil? end
Instance Method Details
#catch(&block)
Alias for #rescue.
# 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
#execute ⇒ Promise
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
.
# 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
[ GitHub ]
#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.
# 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( (@value)) } if_state(:rejected) { child.on_reject(@reason) } end
#on_error(&block)
Alias for #rescue.
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 365
alias_method :on_error, :rescue
#on_fulfill(result) (private)
[ GitHub ]#on_reject(reason) (private)
[ GitHub ]
#on_success { ... } ⇒ Promise
Chain onto this promise an action to be undertaken on success (fulfillment).
#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).
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 360
def rescue(&block) self.then(block) end
#set(value = NULL) { ... } ⇒ IVar
[ 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
Promise
#then(rescuer, executor: executor, &block) ⇒ Promise
Chain a new promise off the current promise.
# 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( (@value)) if @state == :fulfilled child.on_reject(@reason) if @state == :rejected @children << child end child end
Builds a promise that produces the result of self and others in an Array
and fails if any of them fails.
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 440
def zip(*others) self.class.zip(self, *others) end