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, timer_queue_provider: 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

  • timer_queue_provider (#call, nil) (defaults to: nil) —

    returns a TimerQueue-compatible object. Required when submit(timeout:) is used.



348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 348

def initialize(pool_size: 10, queue_size: 100, name: nil, logger: nil, timer_queue_provider: nil)
  @pool_size = pool_size
  @queue_size = queue_size
  @name = name
  @logger = logger
  @timer_queue_provider = timer_queue_provider
  @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



516
517
518
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 516

def name
  @name
end

#pool_size ⇒ Integer (readonly)

Returns configured maximum number of worker threads.

Returns:

  • (Integer) —

    configured maximum number of worker threads



510
511
512
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 510

def pool_size
  @pool_size
end

#queue_size ⇒ Integer (readonly)

Returns configured maximum queue depth.

Returns:

  • (Integer) —

    configured maximum queue depth



513
514
515
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 513

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 whose caller-facing timeout fired after worker execution had started.

Returns:

  • (Integer) —

    number of operations whose caller-facing timeout fired after worker execution had started



493
494
495
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 493

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



480
481
482
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 480

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 or skipped operations spent in the queue waiting for a worker. Returns 0.0 when none have been processed yet.

Returns:

  • (Float) 


501
502
503
504
505
506
507
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 501

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



486
487
488
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 486

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) 


469
470
471
472
473
474
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 469

def shutdown(drain_timeout: 30)
  @shutdown = true
  @queue.close
  @workers.each { |thread| thread.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.

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

A submit-time timeout is an operation-wide deadline measured from the start of this method, including queue wait. The timer settles the PendingOperation and notifies on_complete without forcibly interrupting a running worker. If the deadline fires before worker execution starts, the block is skipped and the operation is not counted as abandoned. If it fires after execution starts, the operation is marked abandoned and the eventual worker result is discarded.

Synchronous queue admission may still delay return from this method when on_full: :wait is used; resolving that requires interruptible admission.

Parameters:

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

    operation-wide deadline in seconds

  • cancellation_token (CancellationToken, nil) (defaults to: nil)
  • on_full (Symbol) (defaults to: :wait) —

    :wait, :raise, or :timeout

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

    queue-admission timeout for on_full: :timeout

Yields:

  • block containing the blocking call

Returns:

Raises:



390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
 
# File 'lib/phronomy/engine/concurrency/blocking_adapter_pool.rb', line 390

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

  submitted_at = Process.clock_gettime(Process::CLOCK_MONOTONIC)
  timer_queue = nil
  if timeout
    timer_queue = @timer_queue_provider&.call
    unless timer_queue
      raise Phronomy::ConfigurationError,
        "timer_queue is required when submit timeout is specified"
    end
  end

  op = PendingOperation.new(
    block,
    timeout: timeout,
    cancellation_token: cancellation_token,
    submitted_at: submitted_at,
    on_abandoned: timeout ? -> { @mutex.synchronize { @abandoned_count += 1 } } : nil
  )

  begin
    if timeout
      elapsed = Process.clock_gettime(Process::CLOCK_MONOTONIC) - submitted_at
      remaining = timeout.to_f - elapsed
      if remaining <= 0
        op.fire_timeout!
        return op
      end

      # Arm before queue admission so the deadline includes time spent waiting
      # for a queue slot.
      timer_queue.schedule(seconds: remaining) { op.fire_timeout! }
    end

    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 => e
    # Shutdown raced with this submit — preserve the existing public error.
    op.fail_submission!(e)
    raise Phronomy::PoolShutdownError, "pool has been shut down"
  rescue => e
    op.fail_submission!(e)
    raise
  end

  op
end