Class: PromptObjects::Execution::BlockingExecutor

Inherits:
Object
  • Object
show all
Defined in:
lib/prompt_objects/execution/blocking_executor.rb

Overview

Bounded worker pool for synchronous adapters and capabilities. Completion is signalled through an IO so an Async fiber yields instead of blocking the server reactor while a worker is occupied.

Defined Under Namespace

Classes: Job, RejectedError

Constant Summary collapse

STOP = 
Object.new.freeze

Instance Method Summary collapse

Constructor Details

#initialize(size: 4, queue_capacity: size * 4) ⇒ BlockingExecutor

Returns a new instance of BlockingExecutor.

Raises:

  • (ArgumentError) 


14
15
16
17
18
19
20
21
22
 
# File 'lib/prompt_objects/execution/blocking_executor.rb', line 14

def initialize(size: 4, queue_capacity: size * 4)
  raise ArgumentError, "size must be positive" unless size.positive?
  raise ArgumentError, "queue_capacity must be positive" unless queue_capacity.positive?

  @jobs = SizedQueue.new(queue_capacity)
  @state_mutex = Mutex.new
  @closed = false
  @workers = Array.new(size) { start_worker }
end

Instance Method Details

#call(&work) ⇒ Object 



24
25
26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/prompt_objects/execution/blocking_executor.rb', line 24

def call(&work)
  raise ArgumentError, "a block is required" unless work

  reader, writer = IO.pipe
  result = {}
  enqueue(Job.new(work: work, completion: writer, result: result))
  reader.read(1)
  raise result[:error] if result.key?(:error)

  result[:value]
ensure
  reader&.close unless reader&.closed?
  writer&.close unless writer&.closed?
end

#close ⇒ Object 



39
40
41
42
43
44
45
46
47
48
 
# File 'lib/prompt_objects/execution/blocking_executor.rb', line 39

def close
  workers = @state_mutex.synchronize do
    return if @closed

    @closed = true
    @workers.dup
  end
  workers.size.times { @jobs.push(STOP) }
  workers.each(&:join)
end