Class: Concurrent::Promise

Inherits:
Object
  • Object
show all
Includes:
Obligation
Defined in:
lib/concurrent/promise.rb

Overview

Promises are inspired by the JavaScript [Promises/A](wiki.commonjs.org/wiki/Promises/A) and [Promises/A+](promises-aplus.github.io/promises-spec/) 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 four possible states: unscheduled, pending, rejected, and fulfilled. A Promise created using .new will be unscheduled. It is scheduled by calling the execute method. Upon execution the Promise and all its children will be set to pending. When a promise is pending it will remain in that state until processing is complete. A completed Promise is either rejected, indicating that an exception was thrown during processing, or fulfilled, indicating it succeeded. 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. A Promise created using .execute will be pending, a Promise created using ‘.fulfill(value)` will be fulfilled with the given value and a Promise created using `.reject(reason)` will be rejected with the given reason.

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.

### Examples

Start by requiring promises

“‘ruby require ’concurrent’ “‘

Then create one

“‘ruby p = Concurrent::Promise.execute do

  # do something
  42
end

“‘

Promises can be chained using the then method. The then method accepts a block, 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 = Concurrent::Promise.new10.then{|x| x * 2}.then{|result| result - 10 }.execute “`

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

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

then{|result| result - 10 }.
then{|result| result * 3 }.
then{|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 nitialization it may be in a different state that 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 = 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:

“‘ruby 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:

“‘ruby 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:

“‘ruby 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:

“‘ruby p = [ Concurrent::Promise.execute{ Thread.pass; raise StandardError } ]

c1 = p.then(Proc.new{ |reason| 42 }) c2 = p.then(Proc.new{ |reason| raise ‘Boom!’ })

sleep(0.1)

c1.state #=> :rejected c2.state #=> :rejected “‘

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 collapse

Instance Method Summary collapse

Methods included from Obligation

#completed?, #exception, #fulfilled?, #incomplete?, #no_error!, #pending?, #reason, #rejected?, #state, #unscheduled?, #value, #value!, #wait

Methods included from Dereferenceable

#value

Constructor Details

#initialize(opts = {}, &block) ⇒ Promise

Initialize a new Promise with the provided options.

Options Hash (opts):

  • :parent (Promise) —

    the parent Promise when building a chain/tree

  • :on_fulfill (Proc) —

    fulfillment handler

  • :on_reject (Proc) —

    rejection handler

  • :operation (Boolean) — default: false —

    when true will execute the future on the global operation pool (for long-running operations), when false will execute the future on the global task pool (for short-running tasks)

  • :executor (object) —

    when provided will run all operations on this executor rather than the global thread pool (overrides :operation)

  • :args (object, Array) —

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

  • :dup_on_deref (String) — default: false —

    call #dup before returning the data

  • :freeze_on_deref (String) — default: false —

    call #freeze before returning the data

  • :copy_on_deref (String) — default: nil —

    call the given Proc passing the internal value and returning the value returned from the proc

See Also:



197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
 
# File 'lib/concurrent/promise.rb', line 197

def initialize(opts = {}, &block)
  opts.delete_if { |k, v| v.nil? }

  @executor = OptionsParser::get_executor_from(opts) || Concurrent.configuration.global_operation_pool
  @args = OptionsParser::get_arguments_from(opts)

  @parent = opts.fetch(:parent) { nil }
  @on_fulfill = opts.fetch(:on_fulfill) { Proc.new { |result| result } }
  @on_reject = opts.fetch(:on_reject) { Proc.new { |reason| raise reason } }

  @promise_body = block || Proc.new { |result| result }
  @state = :unscheduled
  @children = []

  init_obligation
  set_deref_options(opts)
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.



367
368
369
 
# File 'lib/concurrent/promise.rb', line 367

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.



378
379
380
 
# File 'lib/concurrent/promise.rb', line 378

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

Options Hash (opts):

  • :parent (Promise) —

    the parent Promise when building a chain/tree

  • :on_fulfill (Proc) —

    fulfillment handler

  • :on_reject (Proc) —

    rejection handler

  • :operation (Boolean) — default: false —

    when true will execute the future on the global operation pool (for long-running operations), when false will execute the future on the global task pool (for short-running tasks)

  • :executor (object) —

    when provided will run all operations on this executor rather than the global thread pool (overrides :operation)

  • :args (object, Array) —

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

  • :dup_on_deref (String) — default: false —

    call #dup before returning the data

  • :freeze_on_deref (String) — default: false —

    call #freeze before returning the data

  • :copy_on_deref (String) — default: nil —

    call the given Proc passing the internal value and returning the value returned from the proc

Raises:

  • (ArgumentError) —

    if no block is given



251
252
253
 
# File 'lib/concurrent/promise.rb', line 251

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

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



216
217
218
 
# File 'lib/concurrent/promise.rb', line 216

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

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



222
223
224
 
# File 'lib/concurrent/promise.rb', line 222

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

.zip(*promises) ⇒ Promise<Array>

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



325
326
327
328
329
330
331
332
333
334
335
 
# File 'lib/concurrent/promise.rb', line 325

def self.zip(*promises)
  zero = fulfill([], executor: ImmediateExecutor.new)

  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 



227
228
229
230
231
232
233
234
235
236
237
 
# File 'lib/concurrent/promise.rb', line 227

def execute
  if root?
    if compare_and_set_state(:pending, :unscheduled)
      set_pending
      realize(@promise_body)
    end
  else
    @parent.execute
  end
  self
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



298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
 
# File 'lib/concurrent/promise.rb', line 298

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(&block) ⇒ Promise

Raises:

  • (ArgumentError) 


277
278
279
280
 
# File 'lib/concurrent/promise.rb', line 277

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

#rescue(&block) ⇒ Promise Also known as: catch, on_error 



283
284
285
 
# File 'lib/concurrent/promise.rb', line 283

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

#then(rescuer = nil, &block) ⇒ Promise

Returns the new promise.

Raises:

  • (ArgumentError) 


256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
 
# File 'lib/concurrent/promise.rb', line 256

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

  mutex.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(*others) ⇒ Promise<Array>

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



343
344
345
 
# File 'lib/concurrent/promise.rb', line 343

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