Class: Phronomy::Concurrency::BlockingAdapterPool::PendingOperation

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

Overview

Represents the pending result of a submitted blocking operation. Returned immediately by #submit; call #blocking_wait to wait for the result.

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.



184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 184

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
  @started = false
  @abandoned = false
  @wait_time = nil
  @submitted_at = submitted_at || Process.clock_gettime(Process::CLOCK_MONOTONIC)
  @mutex = Mutex.new
  @cond = ConditionVariable.new
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 a submit-time timeout occurred after worker execution had started. The worker is not forcibly interrupted.

Returns:

  • (Boolean) —

    true when a submit-time timeout occurred after worker execution had started. The worker is not forcibly interrupted.



56
57
58
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 56

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

#blocking_wait(timeout: nil, cancellation_token: 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 until the operation completes and returns its value.

A timeout passed here is local to this 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 cancellation settles the operation first.

A submit-time timeout passed to Phronomy::Concurrency::BlockingAdapterPool#submit is enforced by the runtime timer queue independently of this method and is therefore not re-read here.

An optional cancellation_token may be passed here (or at submit time). If the token is cancelled while waiting, Phronomy::CancellationError is raised without interrupting the worker.

Cooperative path (:fiber / DeterministicScheduler): When called from a Fiber managed by DeterministicScheduler, the calling Fiber suspends cooperatively via Fiber.yield rather than blocking the OS thread. The Fiber is resumed through on_complete when the operation settles. A waiter-local timeout: is not enforced on this path; use the submit-time timeout for an operation-wide deadline.

Parameters:

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

    maximum seconds this waiter will block (thread path only; ignored on the cooperative/fiber path)

  • cancellation_token (CancellationToken, nil) (defaults to: nil)

Returns:

  • (Object)

Raises:



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
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 97

def blocking_wait(timeout: nil, cancellation_token: nil)
  effective_token = cancellation_token || @cancellation_token

  raise CancellationError, "blocking operation cancelled" if effective_token&.cancelled?

  # Cooperative context: suspend the calling Fiber rather than blocking
  # the OS thread so that DeterministicScheduler can continue dispatching
  # other tasks while waiting for the blocking worker or submit-time timer.
  scheduler = Thread.current.thread_variable_get(:phronomy_deterministic_scheduler)
  in_managed_fiber = !Fiber.respond_to?(:main) || Fiber.current != Fiber.main
  if scheduler && in_managed_fiber
    unless done?
      scheduler.track_blocking_await
      waiting_fiber = Fiber.current
      on_complete do |_result, _error|
        scheduler.complete_blocking_await
        scheduler.enqueue_fiber(-> { waiting_fiber.resume })
      end
      Fiber.yield(:cooperative_suspend)
    end

    raise CancellationError, "blocking operation cancelled" if effective_token&.cancelled?

    value, error = @mutex.synchronize { [@value, @error] }
    raise error if error

    return value
  end

  # Wake up the waiting thread whenever the token is cancelled so we can
  # propagate cancellation without sleeping until the operation completes.
  effective_token&.on_cancel { @mutex.synchronize { @cond.broadcast } }

  deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout if timeout
  value, error = @mutex.synchronize do
    until @done
      raise CancellationError, "blocking operation cancelled" if effective_token&.cancelled?

      if deadline
        remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
        if remaining <= 0
          raise Phronomy::TimeoutError, "timed out waiting for blocking operation after #{timeout}s"
        end
        @cond.wait(@mutex, remaining)
      else
        @cond.wait(@mutex)
      end
    end

    [@value, @error]
  end

  raise error if error

  value
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)



43
44
45
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 43

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.



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
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 264

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

  cancellation_error = nil
  callbacks = nil
  should_run = @mutex.synchronize do
    if @done
      false
    elsif @cancellation_token&.cancelled?
      cancellation_error = CancellationError.new("operation cancelled before execution")
      @done = true
      @error = cancellation_error
      @cond.broadcast
      callbacks = @callbacks
      @callbacks = nil
      false
    else
      # Linearization point: after this assignment, a concurrent timeout is
      # classified as an in-flight abandonment and the block will run.
      @started = true
      true
    end
  end

  if cancellation_error
    callbacks&.each { |callback| callback.call(nil, cancellation_error) }
    return
  end

  return unless should_run

  # Do NOT use Timeout.timeout here — it delivers an async Thread#raise
  # that can corrupt external library state (mutexes, C extensions, etc.).
  # Each blocking library should set its own native connection/read timeout.
  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



251
252
253
254
255
256
257
258
259
260
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 251

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

    @done = true
    @error = error if error
    @cond.broadcast
  end
  true
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



209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 209

def fire_timeout!
  error = Phronomy::TimeoutError.new(
    "blocking operation timed out after #{@timeout}s"
  )
  callbacks = nil
  abandoned_now = false

  @mutex.synchronize do
    return false if @done

    @done = true
    @timed_out = true
    @error = error
    @abandoned = @started
    abandoned_now = @abandoned
    @cond.broadcast
    callbacks = @callbacks
    @callbacks = nil
  end

  # Internal bookkeeping is completed before user callbacks run. A metrics
  # callback must never suppress delivery of TimeoutError to on_complete.
  if abandoned_now
    begin
      @on_abandoned&.call
    rescue => e
      Phronomy.configuration.logger&.error {
        "BlockingAdapterPool abandoned callback failed: #{e.class}: #{e.message}"
      }
    end
  end

  callbacks&.each { |callback| callback.call(nil, error) }
  true
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 a 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 or on the runtime timer thread. The execution thread is not guaranteed; callbacks must be thread-safe and should complete quickly.

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

Yields:

  • (result, error)

Returns:

  • (self) 


169
170
171
172
173
174
175
176
177
178
179
180
181
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 169

def on_complete(&callback)
  fire_args = nil
  @mutex.synchronize do
    if @done
      fire_args = [@value, @error]
    else
      @callbacks ||= []
      @callbacks << callback
    end
  end
  callback.call(*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



49
50
51
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 49

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



62
63
64
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 62

def wait_time
  @wait_time || 0.0
end