Class: Protobuf::Nats::ThreadPool

Inherits:
Object
  • Object
show all
Defined in:
lib/protobuf/nats/thread_pool.rb

Instance Method Summary collapse

Constructor Details

#initialize(size, opts = {}) ⇒ ThreadPool

Returns a new instance of ThreadPool.



8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
 
# File 'lib/protobuf/nats/thread_pool.rb', line 8

def initialize(size, opts = {})
  @queue = ::Queue.new
  # Lock-free counter of in-flight work. Replaces a mutex-guarded integer so
  # that N workers running in true parallel (JRuby) don't serialize on every
  # task completion.
  @active_work = ::Concurrent::AtomicFixnum.new(0)

  # Callbacks
  @error_cb = lambda do |error|
    logger.error("Error in ThreadPool worker: #{error.message}
 #{error.backtrace.join("
")}")
  end

  # Synchronization
  @mutex = ::Mutex.new      # guards the @workers array only
  @cb_mutex = ::Mutex.new

  # Let's get this party started
  queue_size = opts[:max_queue].to_i || 0
  @max_size = size + queue_size
  @max_workers = size
  @shutting_down = ::Concurrent::AtomicBoolean.new(false)
  @workers = []
  supervise_workers
end

Instance Method Details

#enqueued_size ⇒ Object 



35
36
37
 
# File 'lib/protobuf/nats/thread_pool.rb', line 35

def enqueued_size
  @queue.size
end

#full? ⇒ Boolean

Thread-safe access to check if the pool is full.

Returns:

  • (Boolean) 


40
41
42
 
# File 'lib/protobuf/nats/thread_pool.rb', line 40

def full?
  @active_work.value >= @max_size
end

#kill ⇒ Object 



72
73
74
75
 
# File 'lib/protobuf/nats/thread_pool.rb', line 72

def kill
  @shutting_down.make_true
  @workers.map(&:kill)
end

#max_size ⇒ Object 



44
45
46
 
# File 'lib/protobuf/nats/thread_pool.rb', line 44

def max_size
  @max_size
end

#on_error(&cb) ⇒ Object

This callback is executed in a thread safe manner.



103
104
105
 
# File 'lib/protobuf/nats/thread_pool.rb', line 103

def on_error(&cb)
  @cb_mutex.synchronize { @error_cb = cb }
end

#push(&work_cb) ⇒ Object

This method is now thread-safe.



49
50
51
52
53
54
55
56
57
58
59
60
61
62
 
# File 'lib/protobuf/nats/thread_pool.rb', line 49

def push(&work_cb)
  return false if @shutting_down.true?

  # Optimistically claim a slot; back off if we exceeded the cap. This admits
  # work only while active_work < max_size, matching the original guard, but
  # without holding a mutex across the enqueue.
  if @active_work.increment > @max_size
    @active_work.decrement
    return false
  end

  @queue << [:work, work_cb]
  true
end

#replenish ⇒ Object

Top the pool back up to max_workers if workers have died (e.g. one was killed by a non-StandardError, which the per-task rescue can't catch). This is the ONLY respawn path after initialize -- #push deliberately does not supervise (a mutex acquisition plus an O(workers) alive? scan per request is contention on the hot enqueue path); the server's run loop calls this every second, so a dead worker is replaced within ~1s and its queued work is picked up then. No-op while shutting down so we don't resurrect workers mid-drain.



97
98
99
100
 
# File 'lib/protobuf/nats/thread_pool.rb', line 97

def replenish
  return if @shutting_down.true?
  supervise_workers
end

#shutdown ⇒ Object

This method is now thread-safe.



65
66
67
68
69
70
 
# File 'lib/protobuf/nats/thread_pool.rb', line 65

def shutdown
  # CAS ensures the poison pills are pushed exactly once.
  return unless @shutting_down.make_true

  @max_workers.times { @queue << [:stop, nil] }
end

#size ⇒ Object

Thread-safe access to the current active work size.



108
109
110
 
# File 'lib/protobuf/nats/thread_pool.rb', line 108

def size
  @active_work.value
end

#wait_for_termination(seconds = nil) ⇒ Object

Wait until all workers exit. Returns true if the pool drained, false if the timeout elapsed first. Prunes under the mutex (it mutates @workers).



79
80
81
82
83
84
85
86
87
 
# File 'lib/protobuf/nats/thread_pool.rb', line 79

def wait_for_termination(seconds = nil)
  deadline = seconds && (::Protobuf::Nats.monotonic_time + seconds)
  loop do
    @mutex.synchronize { prune_dead_workers }
    return true if @workers.empty?
    return false if deadline && ::Protobuf::Nats.monotonic_time >= deadline
    sleep 0.1
  end
end