Class: Phronomy::Concurrency::BlockingAdapterPool

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

Overview

A bounded, observable thread pool for blocking I/O operations.

Architectural boundary

BlockingAdapterPool is the only place in Phronomy that uses raw OS threads for I/O. All third-party gem calls whose internal I/O Phronomy cannot control — including RubyLLM, ActiveRecord, Redis, Faraday, and MCP stdio transport — must route through this pool (or a named pool obtained via Runtime#pool). Custom non-blocking HTTP/selector runtimes are intentionally out of scope; the pool + cooperative scheduler combination satisfies all current concurrency requirements without that complexity. (See ADR-010.)

All blocking calls (LLM HTTP, MCP stdio, ActiveRecord, Redis, etc.) must be submitted through this pool so that:

  1. The total number of OS threads is capped.
  2. Queue depth is bounded (backpressure when the pool is saturated).
  3. Per-operation timeouts are enforced consistently.
  4. Abandoned (timed-out) operations are tracked and logged.
  5. Metrics (active count, queue depth, abandoned count, avg wait time) are observable at runtime.

Examples:

Submitting a blocking LLM call

op = runtime.blocking_io.submit(timeout: 30) { chat.ask(message) }
result = op.blocking_wait   # blocks the calling thread until done

With cancellation

token = Phronomy::Concurrency::CancellationToken.timeout_after(60)
op = pool.submit(timeout: 30, cancellation_token: token) { expensive_call }
result = op.blocking_wait

Defined Under Namespace

Classes: PendingOperation

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(pool_size: 10, queue_size: 100, name: nil, logger: nil) ⇒ BlockingAdapterPool

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

Parameters:

  • pool_size (Integer) (defaults to: 10) —

    maximum number of worker threads

  • queue_size (Integer) (defaults to: 100) —

    maximum pending operations waiting for a worker

  • name (String, Symbol, nil) (defaults to: nil) —

    optional pool name used in thread labels

  • logger (Logger, nil) (defaults to: nil) —

    optional logger for warnings



282
283
284
285
286
287
288
289
290
291
292
293
294
295
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 282

def initialize(pool_size: 10, queue_size: 100, name: nil, logger: nil)
  @pool_size = pool_size
  @queue_size = queue_size
  @name = name
  @logger = logger
  @queue = SizedQueue.new(queue_size)
  @active_count = 0
  @abandoned_count = 0
  @total_wait_ns = 0
  @completed_count = 0
  @mutex = Mutex.new
  @shutdown = false
  @workers = Array.new(pool_size) { |i| spawn_worker(i) }
end

Instance Attribute Details

#name ⇒ String, ... (readonly)

Returns pool name used in thread labels.

Returns:

  • (String, Symbol, nil) —

    pool name used in thread labels



404
405
406
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 404

def name
  @name
end

#pool_size ⇒ Integer (readonly)

Returns configured maximum number of worker threads.

Returns:

  • (Integer) —

    configured maximum number of worker threads



398
399
400
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 398

def pool_size
  @pool_size
end

#queue_size ⇒ Integer (readonly)

Returns configured maximum queue depth.

Returns:

  • (Integer) —

    configured maximum queue depth



401
402
403
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 401

def queue_size
  @queue_size
end

Instance Method Details

#abandoned_count ⇒ Integer

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 number of operations that were abandoned due to timeout.

Returns:

  • (Integer) —

    number of operations that were abandoned due to timeout



381
382
383
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 381

def abandoned_count
  @mutex.synchronize { @abandoned_count }
end

#active_count ⇒ Integer

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 number of operations currently executing on workers.

Returns:

  • (Integer) —

    number of operations currently executing on workers



369
370
371
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 369

def active_count
  @mutex.synchronize { @active_count }
end

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

Average time (in seconds) that completed operations spent in the queue waiting for a worker. Returns 0.0 when no operations have completed yet.

Returns:

  • (Float) 


389
390
391
392
393
394
395
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 389

def average_wait_seconds
  @mutex.synchronize do
    return 0.0 if @completed_count.zero?

    @total_wait_ns / @completed_count.to_f / 1_000_000_000.0
  end
end

#queue_depth ⇒ Integer

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 number of operations waiting in the queue.

Returns:

  • (Integer) —

    number of operations waiting in the queue



375
376
377
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 375

def queue_depth
  @queue.size
end

#shutdown(drain_timeout: 30) ⇒ 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.

Gracefully drains the pool and terminates all worker threads. Waits up to drain_timeout seconds for in-flight operations to finish.

Closing the underlying SizedQueue signals workers to exit after draining remaining items, without blocking on a full-queue push.

Parameters:

  • drain_timeout (Numeric) (defaults to: 30) —

    seconds to wait for workers to finish

Returns:

  • (self) 


358
359
360
361
362
363
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 358

def shutdown(drain_timeout: 30)
  @shutdown = true
  @queue.close
  @workers.each { |t| t.join(drain_timeout) }
  self
end

#submit(timeout: nil, cancellation_token: nil, on_full: :wait, full_timeout: 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.

Note:

Cooperative callers: if you are running under the :fiber backend (i.e. inside a DeterministicScheduler Fiber), set timeout: here rather than on PendingOperation#await. The await-time timeout is not enforced on the cooperative path (the Fiber cannot preempt a running worker thread). A submit-time timeout triggers on the worker side and marks the operation abandoned, which unblocks the waiting Fiber via the normal on-complete callback.

Submits a blocking operation to the pool. Returns a PendingOperation immediately; the block runs on a worker thread.

Parameters:

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

    seconds before the operation is abandoned

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

Yields:

  • block containing the blocking call

Returns:

Raises:



315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 315

def submit(timeout: nil, cancellation_token: nil, on_full: :wait, full_timeout: nil, &block)
  raise Phronomy::PoolShutdownError, "pool has been shut down" if @shutdown

  op = PendingOperation.new(block, timeout: timeout, cancellation_token: cancellation_token,
    on_abandoned: timeout ? -> { @mutex.synchronize { @abandoned_count += 1 } } : nil)
  begin
    case on_full
    when :raise
      begin
        @queue.push(op, true)
      rescue ThreadError
        raise Phronomy::BackpressureError, "BlockingAdapterPool queue is full (depth: #{@queue_size})"
      end
    when :timeout
      deadline = full_timeout ? (Process.clock_gettime(Process::CLOCK_MONOTONIC) + full_timeout) : nil
      loop do
        @queue.push(op, true)
        break
      rescue ThreadError
        if deadline && Process.clock_gettime(Process::CLOCK_MONOTONIC) >= deadline
          raise Phronomy::TimeoutError, "timed out waiting for a free slot in BlockingAdapterPool"
        end
        sleep(0.005)
      end
    else # :wait (default)
      @queue.push(op)
    end
  rescue ClosedQueueError
    # Shutdown raced with this submit — treat as if @shutdown was already set.
    raise Phronomy::PoolShutdownError, "pool has been shut down"
  end
  op
end