Class: AsyncFutures::Future

Inherits:
Object
  • Object
show all
Defined in:
lib/async_futures/future.rb

Overview

Class for async execution results.

Heavily inspired by Python's concurrent.futures.Future class.

Constant Summary collapse

FIRST_COMPLETED =

The Future.wait method will return when any future finishes or is cancelled.

:FIRST_COMPLETED
FIRST_EXCEPTION =

The Future.wait method will return when any future finishes by raising an exception. If no future raises an exception then it is equivalent to ALL_COMPLETED.

:FIRST_EXCEPTION
ALL_COMPLETED =

The Future.wait method will return when all futures finish or are cancelled.

:ALL_COMPLETED

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Future

Create a new Future instance in a pending state. Should generally only be called by Executor implementations.



191
192
193
194
195
196
197
198
199
200
 
# File 'lib/async_futures/future.rb', line 191

def initialize
  @mutex = Thread::Mutex.new
  @condition = Thread::ConditionVariable.new
  @state = PENDING
  @result = nil
  @exception = nil
  @done_callbacks = []
  @thread = nil
  @fiber = nil
end

Class Method Details

.as_completed(futures, timeout = nil) ⇒ Object

Returns an Enumerator over the Future instances (possibly created by different Executor instances) given by the Enumerable object futures that yields futures as they complete (finished or cancelled futures).

The returned Enumerator can only be enumerated over once. Subsequent enumeration attempts will raise RuntimeError.

Any futures given by futures that are duplicated will be returned once.

Any futures that completed before as_completed() is called will be yielded first.

The returned Enumerator raises a Timeout::Error if each or next() is called and the result isn’t available after timeout seconds from the original call to as_completed(). timeout can be an int or float. If timeout is not specified or nil, there is no limit to the wait time.

Raises:

  • (Timeout::Error) 


146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
 
# File 'lib/async_futures/future.rb', line 146

def as_completed(futures, timeout = nil) # rubocop:disable Metrics/AbcSize,Metrics/CyclomaticComplexity,Metrics/PerceivedComplexity
  clock_timeout = Time.now.to_f + timeout if timeout
  mtx = Thread::Mutex.new
  queue = Thread::Queue.new
  has_enumerated = false

  fs_ary = futures.to_a.uniq
  fs_sze = fs_ary.size
  fs_cnt = fs_sze

  cb_timeout = timeout && (clock_timeout - Time.now.to_f)
  raise Timeout::Error unless cb_timeout.nil? || cb_timeout.positive?

  Timeout.timeout(cb_timeout) do
    fs_ary.each do |future|
      future.add_done_callback do |done_future|
        queue.push done_future

        mtx.synchronize do
          fs_cnt -= 1
          queue.close if fs_cnt.zero?
        end
      end
    end
  end

  Enumerator.new(fs_sze) do |yielder|
    raise 'Enumerator already consumed' if mtx.synchronize { has_enumerated }

    enum_timeout = timeout && (clock_timeout - Time.now.to_f)
    raise Timeout::Error unless enum_timeout.nil? || enum_timeout.positive?

    Timeout.timeout(enum_timeout) do
      while (done_future = queue.pop)
        yielder.yield done_future
      end
    end
  ensure
    mtx.synchronize { has_enumerated = true }
  end
end

.wait(futures, timeout = nil, return_when = ALL_COMPLETED) ⇒ Object

Wait for the Future instances (possibly created by different Executor instances) given by Enumerable object futures to complete. Duplicate futures given to futures are removed and will be returned only once.

Returns a Hash of sets. The first set, keyed to :done, contains the futures that completed (finished or cancelled futures) before the wait completed. The second set, keyed to :not_done, contains the futures that did not complete (pending or running futures).

timeout can be used to control the maximum number of seconds to wait before returning. timeout can be an int or float. If timeout is not specified or nil, there is no limit to the wait time.

A negative value for timeout is allowed and will just return immediately. Already completed futures are still included in this case. In this circumstance, all return_when values behave identically.

return_when indicates when this function should return. See constant descriptions for details.



54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
 
# File 'lib/async_futures/future.rb', line 54

def wait(futures, timeout = nil, return_when = ALL_COMPLETED) # rubocop:disable Metrics/AbcSize,Metrics/CyclomaticComplexity,Metrics/PerceivedComplexity
  clock_timeout = Time.now.to_f + timeout if timeout
  mtx = Thread::Mutex.new
  queue = Thread::Queue.new

  fs_ary = futures.to_a.uniq
  fs_cnt = fs_ary.size

  done_set = Set.new
  not_done_set = Set.new(fs_ary)

  return { done: done_set, not_done: not_done_set } if fs_ary.empty?

  case return_when
  when FIRST_COMPLETED
    fs_ary.each do |future|
      future.add_done_callback do |ftr|
        mtx.synchronize do
          queue.push(ftr)
          queue.close
        rescue ClosedQueueError
          # Do nothing
        end
      end
    end
  when FIRST_EXCEPTION
    fs_ary.each do |future|
      future.add_done_callback do |ftr|
        mtx.synchronize do
          queue.push(ftr)
          queue.close if !ftr.cancelled? && ftr.exception
        rescue ClosedQueueError
          # Do nothing
        end
      end
    end
  when ALL_COMPLETED
    fs_ary.each do |future|
      future.add_done_callback do |ftr|
        queue.push(ftr)

        mtx.synchronize do
          fs_cnt -= 1
          queue.close if fs_cnt.zero?
        end
      rescue ClosedQueueError
        # Do nothing
      end
    end
  else
    raise ArgumentError.new("Unknown 'return_when' value '#{return_when}'")
  end

  begin
    cb_timeout = timeout && (clock_timeout - Time.now.to_f)
    raise Timeout::Error unless cb_timeout.nil? || cb_timeout.positive?

    Timeout.timeout(cb_timeout) do
      while (dn_ftr = queue.pop)
        done_set.add(dn_ftr)
      end
    end
  rescue Timeout::Error
    queue.close
    while (dn_ftr = queue.pop)
      done_set.add(dn_ftr)
    end
  end

  done_set.merge(fs_ary.lazy.filter(&:done?))
  { done: done_set, not_done: not_done_set.difference(done_set) }
end

Instance Method Details

#add_done_callback(&block) ⇒ Object

Attaches a block that will be called when the future finishes.

The block will be called with this future as its only argument when the future completes or is cancelled. The block will always be called by a Thread in the same Ractor in which it was added. If the future has already completed or been cancelled then the block will be called immediately. These blocks are called in the order that they were added.

Raises:

  • (ArgumentError) 


387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
 
# File 'lib/async_futures/future.rb', line 387

def add_done_callback(&block)
  raise ArgumentError.new('No block given') unless block

  @mutex.synchronize do
    unless lockless_done?
      @done_callbacks.append(block)
      return
    end
  end

  # If we reached here, the future already ended, just call the block immediately.
  begin
    block.call(self)
  rescue Exception # rubocop:disable Lint/RescueException
    logger&.error { "Exception calling callback for #{self}" }
  end
end

#cancel ⇒ Object

Attempt to cancel the call. If the call is currently being executed or finished running and cannot be cancelled then the method will return False, otherwise the call will be cancelled and the method will return True.



271
272
273
274
275
276
277
278
279
280
281
282
283
 
# File 'lib/async_futures/future.rb', line 271

def cancel # rubocop:disable Naming/PredicateMethod
  @mutex.synchronize do
    return true if lockless_cancelled?
    return false if lockless_running? || lockless_finished?

    # The only other state left is PENDING, so we can safely cancel.
    @state = CANCELLED
    @condition.broadcast
  end

  invoke_callbacks
  true
end

#cancelled? ⇒ Boolean

Return True if the call was successfully cancelled.

Returns:

  • (Boolean) 


300
301
302
 
# File 'lib/async_futures/future.rb', line 300

def cancelled?
  @mutex.synchronize { lockless_cancelled? }
end

#complete(*args, **kwargs, &block) ⇒ Object

Convenience method to complete the future with the given block, args, and kwargs.

This method will only run the given block if the future is not already running, canceled, or completed.

It will return true if the block was run by this call and false if it was not run by this call.

Raises:

  • (ArgumentError) 


219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
 
# File 'lib/async_futures/future.rb', line 219

def complete(*args, **kwargs, &block) # rubocop:disable Style/ArgumentsForwarding,Naming/PredicateMethod
  raise ArgumentError.new('No block given') unless block

  begin
    return false unless set_running_or_notify_cancel(set_context: true)
  rescue InvalidStateError
    # RUNNING, CANCELLED_AND_NOTIFIED, or FINISHED states.
    return false
  end

  begin
    result = block.call(*args, **kwargs) # rubocop:disable Style/ArgumentsForwarding
  rescue Exception => e # rubocop:disable Lint/RescueException
    set_exception(e)
  else
    set_result(result)
  end
  true
end

#done? ⇒ Boolean

Return True if the call was successfully cancelled or finished running.

Returns:

  • (Boolean) 


310
311
312
 
# File 'lib/async_futures/future.rb', line 310

def done?
  @mutex.synchronize { lockless_done? }
end

#exception(timeout = nil) ⇒ Object

Return the exception raised by the call. If the call hasn't yet completed then this method will wait up to timeout seconds. If the call hasn't completed in timeout seconds, then a Timeout::Error will be raised. timeout can be an int or float. If timeout is not specified or nil, there is no limit to the wait time.

If the future is cancelled before completing then CancelledError will be raised.

If the call completed without raising, nil is returned.



344
345
346
347
348
349
350
 
# File 'lib/async_futures/future.rb', line 344

def exception(timeout = nil)
  private_join(timeout) do
    raise CancelledError if lockless_cancelled?

    @exception
  end
end

#fiber ⇒ Object

The Fiber that owns the work for this Future. Used to detect deadlocks. Not for direct use. Should only be used by Future and Executor implementations.



243
244
245
 
# File 'lib/async_futures/future.rb', line 243

def fiber
  @mutex.synchronize { @fiber }
end

#fiber=(value) ⇒ Object

Set fiber attribute. Should only be used by Future and Executor implementations.



249
250
251
 
# File 'lib/async_futures/future.rb', line 249

def fiber=(value)
  @mutex.synchronize { @fiber = value }
end

#finished? ⇒ Boolean

Return True if the call finished running and was not cancelled.

Not present on Python concurrent.futures.Future class.

Returns:

  • (Boolean) 


295
296
297
 
# File 'lib/async_futures/future.rb', line 295

def finished?
  @mutex.synchronize { lockless_finished? }
end

#freeze ⇒ Object

The future can’t be frozen, so this method raises an exception:

AsyncFutures::Future.new.freeze # Raises TypeError (cannot freeze #<AsyncFutures::Future:0x...>)

Raises:

  • (TypeError) 


207
208
209
 
# File 'lib/async_futures/future.rb', line 207

def freeze
  raise TypeError.new("cannot freeze #{self}")
end

#join(timeout = nil) ⇒ Object

Wait for future to be done? (through regular completion, exception, or cancellation), then return self. If the call hasn't yet completed then this method will wait up to timeout seconds. If the call hasn't completed in timeout seconds, then nil will be returned. timeout can be an int or float. If timeout is not specified or nil, there is no limit to the wait time.

Calling join with a timeout value of zero will return immediately. This is effectively equivalent to calling done?.

Not present on Python's concurrent.futures.Future class.



368
369
370
371
372
373
374
375
376
 
# File 'lib/async_futures/future.rb', line 368

def join(timeout = nil)
  return (done? && self) || nil if timeout&.zero?

  private_join(timeout) do
    self
  end
rescue Timeout::Error
  nil
end

#pending? ⇒ Boolean

Return True if the call has not yet started.

Not present on Python concurrent.futures.Future class.

Returns:

  • (Boolean) 


288
289
290
 
# File 'lib/async_futures/future.rb', line 288

def pending?
  @mutex.synchronize { lockless_pending? }
end

#result(timeout = nil) ⇒ Object

Return the value returned by the call. If the call hasn't yet completed then this method will wait up to timeout seconds. If the call hasn't completed in timeout seconds, then a Timeout::Error will be raised. timeout can be an int or float. If timeout is not specified or nil, there is no limit to the wait time.

If the future is cancelled before completing then CancelledError will be raised.

If the call raised an exception, this method will raise the same exception.



325
326
327
328
329
330
331
332
 
# File 'lib/async_futures/future.rb', line 325

def result(timeout = nil)
  private_join(timeout) do
    raise CancelledError if lockless_cancelled?
    raise @exception if @exception

    @result
  end
end

#running? ⇒ Boolean

Return True if the call is currently being executed and cannot be cancelled.

Returns:

  • (Boolean) 


305
306
307
 
# File 'lib/async_futures/future.rb', line 305

def running?
  @mutex.synchronize { lockless_running? }
end

#set_exception(exception) ⇒ Object

Sets the result of the work associated with the Future to the Exception exception.

This method should only be used by Executor implementations and unit tests.



466
467
468
469
470
471
472
473
474
475
476
 
# File 'lib/async_futures/future.rb', line 466

def set_exception(exception) # rubocop:disable Naming/AccessorMethodName
  @mutex.synchronize do
    raise InvalidStateError.new(self, @state) if lockless_done?
    raise ArgumentError.new("Not an Exception: #{exception.inspect}") unless exception.is_a?(Exception)

    @exception = exception
    @state = FINISHED
    @condition.broadcast
  end
  invoke_callbacks
end

#set_result(result) ⇒ Object

Sets the result of the work associated with the Future to result.

This method should only be used by Executor implementations and unit tests.



452
453
454
455
456
457
458
459
460
461
 
# File 'lib/async_futures/future.rb', line 452

def set_result(result) # rubocop:disable Naming/AccessorMethodName
  @mutex.synchronize do
    raise InvalidStateError.new(self, @state) if lockless_done?

    @result = result
    @state = FINISHED
    @condition.broadcast
  end
  invoke_callbacks
end

#set_running_or_notify_cancel(set_context: false) ⇒ Object

This method should only be called by Executor implementations before executing the work associated with the Future and by unit tests.

If the method returns false then the Future was cancelled, i.e. Future.cancel was called and returned true. Any threads waiting on the Future completing (i.e. through Future.as_completed() or Future.wait()) will be woken up.

If the method returns true then the Future was not cancelled and has been put in the running state, i.e. calls to Future.running? will return true.

This method should only be called once. If it is called more than once, then it will raise an InvalidStateError exception. If it is called after Future.set_result() or Future.set_exception() have been called then it will raise an InvalidStateError exception. Thus, this is why it is more of an implementation detail for Executor implementations (or similar).



427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
 
# File 'lib/async_futures/future.rb', line 427

def set_running_or_notify_cancel(set_context: false)
  @mutex.synchronize do
    case @state
    when CANCELLED
      @state = CANCELLED_AND_NOTIFIED
      @condition.broadcast
      return false
    when PENDING
      @state = RUNNING
      @condition.broadcast
      if set_context
        @thread = Thread.current
        @fiber = Fiber.current
      end
      return true
    else
      # raised for RUNNING, CANCELLED_AND_NOTIFIED, and FINISHED states.
      raise InvalidStateError.new(self, @state)
    end
  end
end

#thread ⇒ Object

The Thread that owns the work for this Future. Used to detect deadlocks. Not for direct use. Should only be used by Future and Executor implementations.



257
258
259
 
# File 'lib/async_futures/future.rb', line 257

def thread
  @mutex.synchronize { @thread }
end

#thread=(value) ⇒ Object

Set thread attribute. Should only be used by Future and Executor implementations.



263
264
265
 
# File 'lib/async_futures/future.rb', line 263

def thread=(value)
  @mutex.synchronize { @thread = value }
end