Class: Phronomy::Concurrency::OffloadPool::PendingOperation

Inherits:
Object
  • Object
show all
Defined in:
lib/phronomy/engine/concurrency/offload_pool.rb

Overview

Represents the pending result of submitted offloaded work. Returned immediately by #submit; call #blocking_wait to synchronously wait from a non-EventLoop caller such as a low-level test.

Instance Method Summary collapse

Constructor Details

#initialize(block, timeout: nil, cancellation_token: nil, on_abandoned: nil, submitted_at: nil) ⇒ PendingOperation

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a new instance of PendingOperation.



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
187
188
189
190
 
# File 'lib/phronomy/engine/concurrency/offload_pool.rb', line 160

def initialize(
  block,
  timeout: nil,
  cancellation_token: nil,
  on_abandoned: nil,
  submitted_at: nil
)
  @block = block
  @timeout = timeout
  @cancellation_token = cancellation_token
  @on_abandoned = on_abandoned
  @value = nil
  @error = nil
  @done = false
  @timed_out = false
  @cancelled = false
  @started = false
  @abandoned = false
  @wait_time = nil
  @submitted_at = submitted_at ||
    Process.clock_gettime(Process::CLOCK_MONOTONIC)
  @mutex = Mutex.new
  @cond = ConditionVariable.new

  # Explicit submit cancellation is operation-wide. Deadline-only tokens are
  # promoted to cancel! by OffloadPool#submit using the Runtime timer queue.
  @cancellation_callback = if @cancellation_token
    -> { fire_cancellation! }
  end
  @cancellation_token&.on_cancel(&@cancellation_callback)
end

Instance Method Details

#abandoned? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns true when timeout/cancellation settled the caller-facing operation after worker execution had started. The worker is not forcibly interrupted and its eventual result is discarded.

Returns:

  • (Boolean) —

    true when timeout/cancellation settled the caller-facing operation after worker execution had started. The worker is not forcibly interrupted and its eventual result is discarded.



71
72
73
 
# File 'lib/phronomy/engine/concurrency/offload_pool.rb', line 71

def abandoned?
  @mutex.synchronize { @abandoned }
end

#blocking_wait(timeout: nil) ⇒ Object Also known as: wait_result

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Blocks the calling thread until the operation settles and returns its value.

A timeout passed here is local to this synchronous waiter. When it expires, TimeoutError is raised to this caller, but the operation is not settled, marked abandoned, or otherwise changed. The worker continues, and another waiter or an on_complete callback may receive the eventual result unless the submit-time deadline or submit cancellation settles the operation first.

Operation-wide cancellation belongs exclusively to the cancellation_token: passed to Phronomy::Concurrency::OffloadPool#submit. PendingOperation does not define a separate waiter-local cancellation-token lifecycle.

Parameters:

  • timeout (Numeric, nil) (defaults to: nil) —

    maximum seconds this waiter will block

Returns:

  • (Object)

Raises:



99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
 
# File 'lib/phronomy/engine/concurrency/offload_pool.rb', line 99

def blocking_wait(timeout: nil)
  deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout if timeout
  value, error = @mutex.synchronize do
    until @done
      if deadline
        remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
        if remaining <= 0
          raise Phronomy::TimeoutError,
            "timed out waiting for offloaded operation after #{timeout}s"
        end
        @cond.wait(@mutex, remaining)
      else
        @cond.wait(@mutex)
      end
    end

    [@value, @error]
  end

  raise error if error

  value
end

#cancelled? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns true when submit cancellation settled the operation.

Returns:

  • (Boolean) —

    true when submit cancellation settled the operation



63
64
65
 
# File 'lib/phronomy/engine/concurrency/offload_pool.rb', line 63

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

#done? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns true when the caller-facing result has settled (success, failure, cancellation, or submit-time timeout).

Returns:

  • (Boolean) —

    true when the caller-facing result has settled (success, failure, cancellation, or submit-time timeout)



51
52
53
 
# File 'lib/phronomy/engine/concurrency/offload_pool.rb', line 51

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

#execute! ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Executes the operation on a pool worker.



253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
 
# File 'lib/phronomy/engine/concurrency/offload_pool.rb', line 253

def execute!
  @wait_time = Process.clock_gettime(Process::CLOCK_MONOTONIC) - @submitted_at

  # A monotonic deadline can make cancelled? true before the timer callback
  # runs. Promote that state to explicit cancellation so callbacks and all
  # observers see the same operation-wide settlement.
  if @cancellation_token&.cancelled?
    @cancellation_token.cancel!
  end

  should_run = @mutex.synchronize do
    if @done
      false
    else
      # Linearization point: after this assignment, a concurrent timeout or
      # cancellation is classified as in-flight abandonment and the block
      # itself is allowed to finish without Thread#raise.
      @started = true
      true
    end
  end
  return unless should_run

  # Do NOT use Timeout.timeout here — it delivers an async Thread#raise
  # that can corrupt library/application state (mutexes, C extensions, etc.).
  # I/O libraries should set native connection/read timeouts. CPU-heavy work
  # that needs hard termination should use a future process-offload facility.
  begin
    complete_with_value!(@block.call)
  rescue Exception => e # rubocop:disable Lint/RescueException
    # Rescue all Exception subclasses so non-StandardError raises still
    # settle the operation and unblock waiters.
    complete_with_error!(e)
    raise if e.is_a?(SignalException) || e.is_a?(SystemExit)
  end
end

#fail_submission!(error = nil) ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Marks an operation that could not be admitted to the pool as settled, so a previously armed submit-time timer becomes a harmless no-op.

Parameters:

  • error (Exception, nil) (defaults to: nil)

Returns:

  • (Boolean) —

    true when this call changed the state



233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
 
# File 'lib/phronomy/engine/concurrency/offload_pool.rb', line 233

def fail_submission!(error = nil)
  callbacks = nil
  changed = @mutex.synchronize do
    next false if @done

    @done = true
    @error = error if error
    @cond.broadcast
    callbacks = @callbacks
    @callbacks = nil
    true
  end

  detach_submit_cancellation if changed
  deliver_completion_callbacks(callbacks, nil, error) if changed
  changed
end

#fire_cancellation! ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Settles the operation because its submit cancellation token was cancelled.

Cancellation is caller-facing settlement, not asynchronous worker interruption. If execution has already started, the operation is marked abandoned and the worker continues until the synchronous call returns.

Returns:

  • (Boolean) —

    true when this call settled the operation



216
217
218
219
220
221
222
223
224
225
 
# File 'lib/phronomy/engine/concurrency/offload_pool.rb', line 216

def fire_cancellation!
  settle_early!(cancelled: true) do |started|
    message = if started
      "offloaded operation cancelled during execution"
    else
      "offloaded operation cancelled before execution"
    end
    CancellationError.new(message)
  end
end

#fire_timeout! ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Settles the operation with a submit-time timeout.

The worker is not interrupted. If execution has already started, the operation is marked abandoned and the worker's eventual result is discarded.

Returns:

  • (Boolean) —

    true when this call settled the operation, false when the operation had already settled



200
201
202
203
204
205
206
 
# File 'lib/phronomy/engine/concurrency/offload_pool.rb', line 200

def fire_timeout!
  settle_early!(timed_out: true) do
    Phronomy::TimeoutError.new(
      "offloaded operation timed out after #{@timeout}s"
    )
  end
end

#on_complete {|result, error| ... } ⇒ self

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Registers an independent callback to be called when the operation settles.

If the operation has already settled, the callback is invoked immediately on the calling thread. Otherwise it may be invoked on a pool worker thread, on the EventLoop thread when a timer fires, or on the thread that explicitly cancels the submit cancellation token. The execution thread is not guaranteed; callbacks must be thread-safe and should complete quickly.

Completion callback failures are logged and isolated. One callback cannot suppress delivery to later callbacks or change the operation's settled result.

The callback receives result and error (one of them will be nil).

Yields:

  • (result, error)

Returns:

  • (self)

Raises:

  • (ArgumentError) 


143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
 
# File 'lib/phronomy/engine/concurrency/offload_pool.rb', line 143

def on_complete(&callback)
  raise ArgumentError, "on_complete requires a block" unless callback

  fire_args = nil
  @mutex.synchronize do
    if @done
      fire_args = [@value, @error]
    else
      @callbacks ||= []
      @callbacks << callback
    end
  end
  deliver_completion_callback(callback, *fire_args) if fire_args
  self
end

#timed_out? ⇒ Boolean

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns true when the submit-time deadline settled the operation.

Returns:

  • (Boolean) —

    true when the submit-time deadline settled the operation



57
58
59
 
# File 'lib/phronomy/engine/concurrency/offload_pool.rb', line 57

def timed_out?
  @mutex.synchronize { @timed_out }
end

#wait_time ⇒ Float

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns seconds spent in the queue before execution started.

Returns:

  • (Float) —

    seconds spent in the queue before execution started



77
78
79
 
# File 'lib/phronomy/engine/concurrency/offload_pool.rb', line 77

def wait_time
  @wait_time || 0.0
end