Class: Zeridion::Flare::Worker::Pool

Inherits:
Object
  • Object
show all
Defined in:
lib/zeridion_flare/worker/pool.rb

Overview

Bounded-concurrency job pool: concurrency long-lived worker threads draining a work queue, gated by a permit SizedQueue so the runtime can read free slots in O(1) for the poll capacity.

Contract: the caller MUST #acquire a permit BEFORE #submit-ting a job, and the pool releases that permit when the job's block returns (success or error). This keeps free_slots honest: it never schedules more than concurrency jobs, and capacity reflects true free slots.

Instance Method Summary collapse

Constructor Details

#initialize(concurrency:) ⇒ Pool

Returns a new instance of Pool.



15
16
17
18
19
20
21
22
23
 
# File 'lib/zeridion_flare/worker/pool.rb', line 15

def initialize(concurrency:)
  @concurrency = concurrency
  @work = Thread::Queue.new
  # Pre-fill the permit queue: `concurrency` available slots.
  @permits = Thread::SizedQueue.new(concurrency)
  concurrency.times { @permits.push(:permit) }
  @threads = []
  @started = false
end

Instance Method Details

#all_threads_dead? ⇒ Boolean

Are all worker threads dead? Used by tests to assert clean teardown.

Returns:

  • (Boolean) 


99
100
101
 
# File 'lib/zeridion_flare/worker/pool.rb', line 99

def all_threads_dead?
  @threads.none?(&:alive?)
end

#free_slots ⇒ Integer

Returns current free slots (O(1)).

Returns:

  • (Integer) —

    current free slots (O(1)).



51
52
53
 
# File 'lib/zeridion_flare/worker/pool.rb', line 51

def free_slots
  @permits.size
end

#in_flight ⇒ Integer

Returns number of jobs currently executing.

Returns:

  • (Integer) —

    number of jobs currently executing.



74
75
76
 
# File 'lib/zeridion_flare/worker/pool.rb', line 74

def in_flight
  @concurrency - free_slots
end

#shutdown(timeout:) ⇒ Object

Drain: stop accepting new work and wait up to timeout seconds for in-flight jobs to finish, then tear down the threads. Returns true if all threads exited within the timeout, false if some were still running (the caller decides what to do — we do NOT kill them, since a Thread#kill mid-Net::HTTP can corrupt the socket).



83
84
85
86
87
88
89
90
91
92
93
94
95
96
 
# File 'lib/zeridion_flare/worker/pool.rb', line 83

def shutdown(timeout:)
  deadline = monotonic_now + timeout.to_f
  # Tell each thread to exit once it has finished its current job and
  # drained any queued work.
  @concurrency.times { @work.push(:__stop__) }

  all_joined = true
  @threads.each do |t|
    remaining = deadline - monotonic_now
    remaining = 0 if remaining.negative?
    all_joined = false if t.join(remaining).nil?
  end
  all_joined
end

#start ⇒ Object

Spawn the worker threads. Idempotent.



26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
 
# File 'lib/zeridion_flare/worker/pool.rb', line 26

def start
  return self if @started

  @started = true
  @concurrency.times do |i|
    @threads << Thread.new do
      if Thread.current.respond_to?(:name=)
        Thread.current.name = "flare-job-#{i}"
      end
      loop do
        task = @work.pop
        break if task == :__stop__

        begin
          task.call
        ensure
          release
        end
      end
    end
  end
  self
end

#submit(&block) ⇒ Object

Enqueue a job block onto a worker thread. A permit must already be held (via #try_acquire). The pool releases the permit after the block runs.



68
69
70
71
 
# File 'lib/zeridion_flare/worker/pool.rb', line 68

def submit(&block)
  @work.push(block)
  nil
end

#try_acquire ⇒ Object

Try to reserve a slot without blocking. Returns true if a permit was taken (the caller must then #submit and the pool will release it), or false if the pool is saturated.



58
59
60
61
62
63
 
# File 'lib/zeridion_flare/worker/pool.rb', line 58

def try_acquire
  @permits.pop(true)
  true
rescue ThreadError
  false
end