Class: AtomicRuby::AtomicThreadPool

Inherits:
Object
  • Object
show all
Defined in:
lib/atomic-ruby/atomic_thread_pool.rb

Overview

Note:

This class is NOT Ractor-safe as it contains mutable thread state that cannot be safely shared across ractors.

Provides a thread pool using atomic operations for work queuing.

AtomicThreadPool maintains a baseline number of worker threads that process work items from an AtomicQueue. When max_size is provided, it can temporarily add workers when work remains queued and its workers spend most of their time blocked outside the GVL. Both enqueueing and dequeueing are O(1) and lock-free, so concurrent producers and consumers never block one another.

Examples:

Basic usage

pool = AtomicThreadPool.new(size: 4)
pool << proc { puts "Hello from worker thread!" }
pool << proc { puts "Another work item" }
pool.shutdown

Processing work with results

results = []
pool = AtomicThreadPool.new(size: 2, name: "Calculator")

10.times do |index|
  pool << proc { results << index * 2 }
end

pool.shutdown
puts results.sort #=> [0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

Scaling for blocking work

pool = AtomicThreadPool.new(size: 2, max_size: 8)
20.times { pool << proc { Net::HTTP.get(uri) } }
pool.shutdown

Monitoring pool state

pool = AtomicThreadPool.new(size: 3)
puts pool.length        #=> 3
puts pool.queue_length  #=> 0
puts pool.active_count  #=> 0

5.times { pool << proc { sleep(1) } }
puts pool.queue_length  #=> 2 (3 workers busy, 2 queued)
puts pool.active_count  #=> 3 (3 workers processing)

Defined Under Namespace

Classes: EnqueuedWorkAfterShutdownError, Error

Instance Method Summary collapse

Constructor Details

#initialize(size:, max_size: nil, name: nil, on_error: nil) ⇒ AtomicThreadPool

Creates a new thread pool with the specified baseline size.

When max_size is greater than size, the pool adds temporary workers if work remains queued while its active workers spend most of their time blocked outside the GVL, but not when they are waiting for the GVL. Temporary workers leave after the queue remains empty, returning the pool to its baseline size. Omitting max_size creates a fixed-size pool.

Examples:

Create a basic pool

pool = AtomicThreadPool.new(size: 4)

Create a named pool

pool = AtomicThreadPool.new(size: 2, name: "Database Workers")

Create an adaptive pool

pool = AtomicThreadPool.new(size: 2, max_size: 8)

Create a pool with a custom error handler

errors = []
pool = AtomicThreadPool.new(size: 2, on_error: ->(err) { errors << err })

RBS:

  • (size: Integer, ?max_size: (Integer | Float)?, ?name: String?, ?on_error: Proc?) -> void

Parameters:

  • size (Integer) —

    The baseline number of worker threads (must be positive)

  • max_size (Integer, Float, nil) (defaults to: nil) —

    Maximum number of worker threads, Float::INFINITY for no limit, or nil for a fixed-size pool

  • name (String, nil) (defaults to: nil) —

    Optional name for the thread pool (used in thread names)

  • on_error (Proc, nil) (defaults to: nil) —

    Optional error handler called with the exception when a work item raises. Receives the exception as its argument. When nil, errors are printed to stderr

Raises:

  • (ArgumentError) —

    if size is not a positive integer

  • (ArgumentError) —

    if max_size is not an integer greater than or equal to size or Float::INFINITY

  • (ArgumentError) —

    if name is provided but not a string

  • (ArgumentError) —

    if on_error is provided but not a Proc



104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 104

def initialize(size:, max_size: nil, name: nil, on_error: nil)
  raise ArgumentError, "size must be a positive Integer" unless size.is_a?(Integer) && size > 0
  valid_max_size = max_size.nil? || max_size == Float::INFINITY || (max_size.is_a?(Integer) && max_size >= size)
  raise ArgumentError, "max_size must be an Integer greater than or equal to size or Float::INFINITY" unless valid_max_size
  raise ArgumentError, "name must be a String" unless name.nil? || name.is_a?(String)
  raise ArgumentError, "on_error must be a Proc" unless on_error.nil? || on_error.is_a?(Proc)

  @size = size
  @max_size = max_size || size
  @name = name
  @on_error = on_error

  @queue = AtomicQueue.new
  @shutdown = AtomicBoolean.new(false)
  @work_available = AtomicConditionVariable.new
  @alive_thread_count = Atom.new(0)
  @active_thread_count = Atom.new(0)
  @threads = []
  @next_thread_number = 0
  if adaptive?
    @autoscale_available = AtomicConditionVariable.new
    @trim_requested = AtomicBoolean.new(false)
    @thread_pool_monitor = ThreadPoolMonitor.new
  end

  start
end

Instance Method Details

#<<(work) ⇒ Object

Enqueues work to be executed by the thread pool.

The work item must respond to #call (typically a Proc or lambda). Work items are executed in FIFO order by available worker threads. If all workers are busy, the work is queued atomically. Enqueueing is O(1) regardless of current queue depth.

Examples:

Enqueue a simple task

pool << proc { puts "Hello World" }

Enqueue a lambda with parameters

calculator = ->(a, b) { puts a + b }
pool << proc { calculator.call(2, 3) }

Enqueue work that captures variables

name = "Alice"
pool << proc { puts "Processing #{name}" }

RBS:

  • (Proc work) -> void

Parameters:

  • work (#call) —

    A callable object to be executed by a worker thread

Raises:



155
156
157
158
159
160
161
162
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 155

def <<(work)
  raise EnqueuedWorkAfterShutdownError if @shutdown.true?

  @trim_requested&.make_false
  @queue.push(work)
  @work_available.signal
  @autoscale_available&.signal
end

#active_count ⇒ Integer

Returns the number of worker threads currently executing work.

This represents threads that have picked up a work item and are actively processing it. The count includes threads in the middle of executing work.call, but excludes threads that are idle or waiting for work.

Examples:

Monitor active workers

pool = AtomicThreadPool.new(size: 4)
puts pool.active_count #=> 0

5.times { pool << proc { sleep(1) } }
sleep(0.1) # Give threads time to pick up work
puts pool.active_count #=> 4 (all workers busy)
puts pool.queue_length #=> 1 (one item still queued)

Calculate total load

total_load = pool.active_count + pool.queue_length
puts "Total pending work: #{total_load}"

RBS:

  • () -> Integer

Returns:

  • (Integer) —

    The number of threads actively processing work



232
233
234
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 232

def active_count
  @active_thread_count.value
end

#length ⇒ Integer Also known as: size

Returns the number of currently alive worker threads.

This count decreases as the pool shuts down and threads terminate. An adaptive pool may report a value between the size and max_size parameters passed to the constructor.

Examples:

pool = AtomicThreadPool.new(size: 4)
puts pool.length #=> 4
pool.shutdown
puts pool.length #=> 0

RBS:

  • () -> Integer

Returns:

  • (Integer) —

    The number of alive worker threads



179
180
181
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 179

def length
  @alive_thread_count.value
end

#queue_length ⇒ Integer Also known as: queue_size

Returns the number of work items currently queued for execution.

This represents work that has been enqueued but not yet picked up by a worker thread. A high queue length indicates that work is being submitted faster than it can be processed.

Examples:

pool = AtomicThreadPool.new(size: 2)
5.times { pool << proc { sleep(1) } }
puts pool.queue_length #=> 3 (2 workers busy, 3 queued)

RBS:

  • () -> Integer

Returns:

  • (Integer) —

    The number of queued work items



201
202
203
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 201

def queue_length
  @queue.size
end

#shutdown ⇒ void

This method returns an undefined value.

Gracefully shuts down the thread pool.

This method:

  1. Marks the pool as shutdown (preventing new work from being enqueued)
  2. Waits for all currently queued work to complete
  3. Waits for all worker threads to terminate

After shutdown, all worker threads will be terminated and the pool cannot be restarted. Attempting to enqueue work after shutdown will raise an exception.

Examples:

pool = AtomicThreadPool.new(size: 4)
10.times { |index| pool << proc { puts index } }
pool.shutdown # waits for all work to complete
puts pool.length #=> 0

RBS:

  • () -> void

Raises:



258
259
260
261
262
263
264
265
266
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 258

def shutdown
  return if @shutdown.true?

  @shutdown.make_true
  @autoscale_available&.broadcast
  @work_available.broadcast
  @autoscaler&.join
  @threads.each(&:join)
end