Class: Concurrent::Future

Inherits:
IVar
  • Object
show all
Defined in:
lib/concurrent-ruby/concurrent/future.rb

Overview

## 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 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.

See Also:

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from IVar

#add_observer, #fail, #try_set

Methods included from Concern::Observable

#add_observer, #count_observers, #delete_observer, #delete_observers, #with_observer

Methods included from Concern::Obligation

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

Methods included from Concern::Dereferenceable

#value

Constructor Details

#initialize(opts = {}) { ... } ⇒ Future

Create a new Future in the :unscheduled state.

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 Concern::Obligation#value

  • :freeze_on_deref (Boolean) — default: false —

    Call #freeze before returning the data from Concern::Obligation#value

  • :copy_on_deref (Proc) — default: nil —

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

  • :args (object, Array) —

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

Yields:

  • the asynchronous operation to perform

Raises:

  • (ArgumentError) —

    if no block is given



33
34
35
36
 
# File 'lib/concurrent-ruby/concurrent/future.rb', line 33

def initialize(opts = {}, &block)
  raise ArgumentError.new('no block given') unless block_given?
  super(NULL, opts.merge(__task_from_block__: block), &nil)
end

Class Method Details

.execute(opts = {}) { ... } ⇒ Future

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

Examples:

future = Concurrent::Future.execute{ sleep(1); 42 }
future.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 Concern::Obligation#value

  • :freeze_on_deref (Boolean) — default: false —

    Call #freeze before returning the data from Concern::Obligation#value

  • :copy_on_deref (Proc) — default: nil —

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

  • :args (object, Array) —

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

Yields:

  • the asynchronous operation to perform

Returns:

  • (Future) —

    the newly created Future in the :pending state

Raises:

  • (ArgumentError) —

    if no block is given



77
78
79
 
# File 'lib/concurrent-ruby/concurrent/future.rb', line 77

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

Instance Method Details

#cancel ⇒ Boolean

Attempt to cancel the operation if it has not already processed. The operation can only be cancelled while still pending. It cannot be cancelled once it has begun processing or has completed.

Returns:

  • (Boolean) —

    was the operation successfully cancelled.



99
100
101
102
103
104
105
106
 
# File 'lib/concurrent-ruby/concurrent/future.rb', line 99

def cancel
  if compare_and_set_state(:cancelled, :pending)
    complete(false, nil, CancelledOperationError.new)
    true
  else
    false
  end
end

#cancelled? ⇒ Boolean

Has the operation been successfully cancelled?

Returns:

  • (Boolean) 


111
112
113
 
# File 'lib/concurrent-ruby/concurrent/future.rb', line 111

def cancelled?
  state == :cancelled
end

#execute ⇒ Future

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

Examples:

Instance and execute in separate steps

future = Concurrent::Future.new{ sleep(1); 42 }
future.state #=> :unscheduled
future.execute
future.state #=> :pending

Instance and execute in one line

future = Concurrent::Future.new{ sleep(1); 42 }.execute
future.state #=> :pending

Returns:

  • (Future) —

    a reference to self



53
54
55
56
57
58
 
# File 'lib/concurrent-ruby/concurrent/future.rb', line 53

def execute
  if compare_and_set_state(:pending, :unscheduled)
    @executor.post{ safe_execute(@task, @args) }
    self
  end
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:



82
83
84
85
86
87
88
89
90
91
92
 
# File 'lib/concurrent-ruby/concurrent/future.rb', line 82

def set(value = NULL, &block)
  check_for_block_or_value!(block_given?, value)
  synchronize do
    if @state != :unscheduled
      raise MultipleAssignmentError
    else
      @task = block || Proc.new { value }
    end
  end
  execute
end

#wait_or_cancel(timeout) ⇒ Boolean

Wait the given number of seconds for the operation to complete. On timeout attempt to cancel the operation.

Parameters:

  • timeout (Numeric) —

    the maximum time in seconds to wait.

Returns:

  • (Boolean) —

    true if the operation completed before the timeout else false



121
122
123
124
125
126
127
128
129
 
# File 'lib/concurrent-ruby/concurrent/future.rb', line 121

def wait_or_cancel(timeout)
  wait(timeout)
  if complete?
    true
  else
    cancel
    false
  end
end