Class: Concurrent::Promise
- 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 Concern::Obligation#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
require 'concurrent'
Then create one
p = Concurrent::Promise.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.
p = Concurrent::Promise.new{10}.then{|x| x * 2}.then{|result| result - 10 }.execute
And so on, and so on, and so on...
p = Concurrent::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:
# create, operate, then execute
p1 = Concurrent::Promise.new{ "Hello World!" }
p1.state #=> :unscheduled
p1.execute
# create and immediately execute
p2 = Concurrent::Promise.new{ "Hello World!" }.execute
# execute during creation
p3 = Concurrent::Promise.execute{ "Hello World!" }
Once the execute
method is called a Promise
becomes pending
:
p = Concurrent::Promise.execute{ "Hello, world!" }
p.state #=> :pending
p.pending? #=> true
Wait a little bit, and the promise will resolve and provide a value:
p = Concurrent::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:
p = Concurrent::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:
p = Concurrent::Promise.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 asthen {|result| ... }
rescue { |reason| ... }
is the same asthen(Proc.new { |reason| ... } )
rescue
is aliased bycatch
andon_error
Class Method Summary collapse
-
.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. -
.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.
Instance Method Summary collapse
-
#execute ⇒ Promise
Execute an
:unscheduled
Promise
. -
#fail(reason = StandardError.new) ⇒ IVar
Set the
IVar
to failed due to some error and wake or notify all threads waiting on it. -
#flat_map(&block) ⇒ Promise
Yield the successful result to the block that returns a promise.
-
#initialize(opts = {}) { ... } ⇒ Promise
constructor
Initialize a new Promise with the provided options.
-
#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
Set the
IVar
to a value and wake or notify all threads waiting on it. -
#then(*args) { ... } ⇒ Promise
Chain a new promise off the current promise.
-
#zip(*others) ⇒ Promise<Array>
Builds a promise that produces the result of self and others in an Array and fails if any of them fails.
Constructor Details
#initialize(opts = {}) { ... } ⇒ Promise
Initialize a new Promise with the provided options.
210 211 212 213 |
# 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 Concurrent::PromiseExecutionError
if any of the aggregated promises
fail. Upon execution will execute any of the aggregate promises that
were not already executed.
463 464 465 |
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 463 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 Concurrent::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
Concurrent::PromiseExecutionError
.
474 475 476 |
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 474 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.
295 296 297 |
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 295 def self.execute(opts = {}, &block) new(opts, &block).execute end |
.fulfill(value, opts = {}) ⇒ Promise
Create a new Promise
and fulfill it immediately.
224 225 226 |
# 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.
237 238 239 |
# 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.
408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 |
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 408 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 Method Details
#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
.
246 247 248 249 250 251 252 253 254 255 256 |
# 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 @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.
277 278 279 |
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 277 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.
374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 |
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 374 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 |
#on_success { ... } ⇒ Promise
Chain onto this promise an action to be undertaken on success (fulfillment).
348 349 350 351 |
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 348 def on_success(&block) raise ArgumentError.new('no block given') unless block_given? self.then(&block) end |
#rescue { ... } ⇒ Promise Also known as: catch, on_error
Chain onto this promise an action to be undertaken on failure (rejection).
359 360 361 |
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 359 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.
261 262 263 264 265 266 267 268 269 270 271 272 |
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 261 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 |
#then(rescuer, executor, &block) ⇒ Promise #then(rescuer, executor: executor, &block) ⇒ Promise
Chain a new promise off the current promise.
313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 |
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 313 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 |
#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.
439 440 441 |
# File 'lib/concurrent-ruby/concurrent/promise.rb', line 439 def zip(*others) self.class.zip(self, *others) end |