Class: Karafka::Processing::WorkersPool

Inherits:
Object
  • Object
show all
Defined in:
lib/karafka/processing/workers_pool.rb

Overview

Dynamic thread pool that manages worker threads. Supports scaling at runtime via #scale.

All public methods that read or mutate @workers are synchronized via @mutex. @size is always updated under @mutex but can be read without locking for performance (integer assignment is atomic in MRI).

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ WorkersPool

Initializes an empty pool with zero workers. Workers are not started until #scale is called, allowing the pool to be created early (e.g. in Server.run) before the jobs queue exists.



33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/karafka/processing/workers_pool.rb', line 33

def initialize
  @jobs_queue = nil
  @workers = []
  @size = 0
  @mutex = Mutex.new
  # Monotonically increasing index for naming worker threads. Indices are never reused
  # after a worker exits, so thread names remain unique across the lifetime of the process
  # and make it easy to correlate log entries with specific worker generations.
  @next_index = 0
  # Number of downscale sentinels enqueued but not yet picked up by a worker. Workers
  # deregister asynchronously, so the live thread count alone overstates the pool between
  # a scale-down request and its completion. All sizing decisions use the committed size
  # (live workers minus pending shrinks) so repeated or overlapping scale-down requests do
  # not enqueue duplicate sentinels and drain the pool below the requested target (or
  # below the minimum of one worker).
  @pending_shrinks = 0
end

Instance Attribute Details

#jobs_queue=(value) ⇒ Object (writeonly)

Jobs queue reference, set by the Runner after both pool and queue are created. Must be assigned before calling #scale.



26
27
28
 
# File 'lib/karafka/processing/workers_pool.rb', line 26

def jobs_queue=(value)
  @jobs_queue = value
end

#size ⇒ Integer (readonly)

Returns current number of workers registered in the pool. Reflects the actual thread count, not a target. After a downscale request this value converges towards the target as workers pick up nil sentinels and deregister. Updated atomically under mutex, safe to read without locking.

Returns:

  • (Integer) —

    current number of workers registered in the pool. Reflects the actual thread count, not a target. After a downscale request this value converges towards the target as workers pick up nil sentinels and deregister. Updated atomically under mutex, safe to read without locking.



22
23
24
 
# File 'lib/karafka/processing/workers_pool.rb', line 22

def size
  @size
end

Instance Method Details

#alive ⇒ Array<Worker>

Returns workers that are still alive.

Returns:

  • (Array<Worker>) —

    workers that are still alive



96
97
98
 
# File 'lib/karafka/processing/workers_pool.rb', line 96

def alive
  snapshot.select(&:alive?)
end

#deregister(worker) ⇒ Object

Called by a worker when it exits (queue closed or pool downscaling). Thread-safe -- worker threads call this from their own thread.

Parameters:

  • worker (Worker) —

    worker to remove from the pool



114
115
116
117
118
119
120
121
122
123
124
 
# File 'lib/karafka/processing/workers_pool.rb', line 114

def deregister(worker)
  @mutex.synchronize do
    @workers.delete(worker)
    @size = @workers.size
    # A worker exits either because it picked up a downscale sentinel or because the jobs
    # queue was closed for shutdown. The worker cannot tell those apart, so pending
    # removals are settled here, with a floor because shutdown exits may have no pending
    # shrink to consume
    @pending_shrinks -= 1 if @pending_shrinks.positive?
  end
end

#join ⇒ Object

Wait for all current workers to finish.



106
107
108
 
# File 'lib/karafka/processing/workers_pool.rb', line 106

def join
  snapshot.each(&:join)
end

#scale(target) ⇒ Object

Scale pool towards target workers (minimum 1).

Scaling up is synchronous -- new worker threads are spawned and registered before this method returns. #size reflects the new count immediately.

Scaling down is asynchronous -- nil sentinels are enqueued and workers exit when they pick one up. #size decreases gradually as workers deregister themselves. Callers that need to know when downsizing is complete should poll #size or listen for the worker.scaling.down instrumentation event (whose :to payload reports the target, not the current count).

The entire read-decide-act cycle is synchronized to prevent stale reads. Instrumentation runs outside the mutex to avoid holding the lock during user callbacks.

Parameters:

  • target (Integer) —

    desired number of workers

Raises:



66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
 
# File 'lib/karafka/processing/workers_pool.rb', line 66

def scale(target)
  raise(Karafka::Errors::BaseError, "jobs_queue must be set before scaling") unless @jobs_queue

  target = [target, 1].max
  event = nil

  @mutex.synchronize do
    # Compute against the committed size, not the live thread count: sentinels already in
    # flight are removals this pool is committed to, just not yet executed by the workers
    current = @workers.size - @pending_shrinks
    delta = target - current

    if delta.positive?
      event = grow(delta)
    elsif delta.negative?
      event = shrink(delta.abs)
    end
  end

  return unless event

  monitor.instrument(*event)
end

#stopped? ⇒ Boolean

Returns true if all workers have stopped.

Returns:

  • (Boolean) —

    true if all workers have stopped



91
92
93
 
# File 'lib/karafka/processing/workers_pool.rb', line 91

def stopped?
  snapshot.none?(&:alive?)
end

#terminate ⇒ Object

Forcefully terminate all worker threads.



101
102
103
 
# File 'lib/karafka/processing/workers_pool.rb', line 101

def terminate
  snapshot.each(&:terminate)
end