Class: AtomicRuby::AtomicThreadPool

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

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

Constant Summary collapse

AUTOSCALE_IDLE_TIME =

Returns:

  • (::Integer) 
5
AUTOSCALE_INITIAL_WINDOW =

Returns:

  • (::Float) 
0.002
AUTOSCALE_MAX_WINDOW =

Returns:

  • (::Float) 
0.05
AUTOSCALE_SAMPLES_PER_WINDOW =

Returns:

  • (::Integer) 
5

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 Ruby execution is using the available CPU. Temporary workers remain available between blocking bursts, but retire when Ruby execution becomes the bottleneck or they remain idle. 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 })

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

  • size: (Integer)
  • max_size: (Integer, Float, nil) (defaults to: nil)
  • name: (String, nil) (defaults to: nil)
  • on_error: (Proc, nil) (defaults to: nil)

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



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
131
132
133
134
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 106

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
    @idle_trim_requested = AtomicBoolean.new(false)
    @ruby_cpu_trim_requested = AtomicBoolean.new(false)
    @thread_pool_monitor = ThreadPoolMonitor.new
    @thread_pool_monitor.start
  end

  start
end

Instance Method Details

#<<(work) ⇒ void

This method returns an undefined value.

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}" }

Parameters:

  • work (#call) —

    A callable object to be executed by a worker thread

Raises:



159
160
161
162
163
164
165
166
167
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 159

def <<(work)
  Thread.handle_interrupt(Exception => :never) do
    @idle_trim_requested&.make_false
    raise EnqueuedWorkAfterShutdownError unless @queue.push(work)

    @work_available.signal
    @autoscale_available&.signal
  end
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}"

Returns:

  • (Integer) —

    The number of threads actively processing work



237
238
239
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 237

def active_count
  @active_thread_count.value
end

#adaptive? ⇒ true, false

Returns whether the pool may grow beyond its baseline size.

Returns:

  • (true, false) 


309
310
311
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 309

def adaptive?
  @max_size > @size
end

#autoscale ⇒ void

This method returns an undefined value.

Scales temporary workers in response to workload pressure.

Sampling starts aggressively so the pool can react to short bursts, then backs off when more workers would not improve throughput.



399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 399

def autoscale
  previous_snapshot = @thread_pool_monitor.snapshot
  pressure_started_at = nil
  idle_started_at = nil
  phase_samples = [0, 0, 0]
  sampling_window = AUTOSCALE_INITIAL_WINDOW

  loop do
    @ruby_cpu_trim_requested.make_false if @alive_thread_count.value <= @size
    @autoscale_available.wait do
      @shutdown.true? || !@queue.empty? || @alive_thread_count.value > @size
    end
    break if @shutdown.true?

    now = Process.clock_gettime(Process::CLOCK_MONOTONIC)
    snapshot = @thread_pool_monitor.snapshot

    if @queue.empty?
      pressure_started_at = nil
      previous_snapshot = snapshot
      idle_started_at ||= now
      phase_samples.fill(0)
      sampling_window = AUTOSCALE_INITIAL_WINDOW

      if now - idle_started_at >= AUTOSCALE_IDLE_TIME
        @idle_trim_requested.make_true
        @work_available.broadcast
      end
    else
      @idle_trim_requested.make_false
      idle_started_at = nil

      if pressure_started_at.nil?
        previous_snapshot = snapshot
        pressure_started_at = now
        phase_samples.fill(0)
      elsif now - pressure_started_at >= sampling_window
        case scaling_direction(previous_snapshot, snapshot, phase_samples, now - pressure_started_at)
        when :up
          @ruby_cpu_trim_requested.make_false
          worker_count = @threads.length
          workers_to_add = [worker_count, @queue.size, @max_size - worker_count].min
          workers_to_add.times { spawn_worker(temporary: true) }
          sampling_window = AUTOSCALE_INITIAL_WINDOW
        when :down
          unless @ruby_cpu_trim_requested.true?
            @ruby_cpu_trim_requested.make_true
            @work_available.broadcast
          end
          sampling_window = [sampling_window * 2, AUTOSCALE_MAX_WINDOW].min
        else
          sampling_window = [sampling_window * 2, AUTOSCALE_MAX_WINDOW].min
        end
        previous_snapshot = snapshot
        pressure_started_at = now
        phase_samples.fill(0)
      else
        3.times { |index| phase_samples[index] += snapshot[index] }
      end
    end

    sleep((@queue.empty? ? AUTOSCALE_MAX_WINDOW : sampling_window) / AUTOSCALE_SAMPLES_PER_WINDOW)
  end
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

Returns:

  • (Integer) —

    The number of alive worker threads



184
185
186
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 184

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)

Returns:

  • (Integer) —

    The number of queued work items



206
207
208
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 206

def queue_length
  @queue.size
end

#scaling_direction(previous_snapshot, snapshot, phase_samples, elapsed_time) ⇒ :up, ...

Returns the direction in which the pool should scale.

Requiring workers to spend a majority of their time blocked outside the GVL recognizes blocking operations. The pool grows while Ruby execution uses less than half of one CPU and shrinks when Ruby execution becomes the bottleneck, without mistaking OS scheduling delays for GVL contention.

Parameters:

  • previous_snapshot (Array<Integer>) —

    previous GVL state snapshot

  • snapshot (Array<Integer>) —

    current GVL state snapshot

  • phase_samples (Array<Integer>) —

    sampled current GVL states

  • elapsed_time (Float) —

    seconds covered by the snapshots

Returns:

  • (:up, :down, nil) 


478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 478

def scaling_direction(previous_snapshot, snapshot, phase_samples, elapsed_time)
  @threads.select!(&:alive?)
  workers = @threads
  return if @active_thread_count.value < workers.length

  running_time = snapshot[3] - previous_snapshot[3]
  waiting_time = snapshot[4] - previous_snapshot[4]
  blocked_time = snapshot[5] - previous_snapshot[5]
  running_cpu_time = snapshot[6] - previous_snapshot[6]
  total_time = running_time + waiting_time + blocked_time

  minimum_measured_time = (elapsed_time * 1_000_000_000 * workers.length / 2).to_i
  if total_time < minimum_measured_time
    running_time, waiting_time, blocked_time = phase_samples
    total_time = running_time + waiting_time + blocked_time
  end

  ruby_cpu_bound = running_cpu_time * 2 >= elapsed_time * 1_000_000_000
  blocking = total_time.positive? && blocked_time * 2 > total_time && !ruby_cpu_bound

  return :up if blocking && workers.length < @max_size
  return :down if ruby_cpu_bound && workers.length > @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

Enqueues accepted before shutdown are processed before the workers 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

Raises:



266
267
268
269
270
271
272
273
274
275
276
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 266

def shutdown
  Thread.handle_interrupt(Exception => :never) do
    @queue.send(:_close)
    @shutdown.make_true
    @autoscale_available&.broadcast
    @work_available.broadcast
  end
  @autoscaler&.join
  @threads.each(&:join)
  @thread_pool_monitor&.stop
end

#spawn_worker(temporary: false) ⇒ Thread

Creates a worker thread.

Temporary workers retire when Ruby execution becomes the bottleneck or they remain idle.

Parameters:

  • temporary (true, false) (defaults to: false) —

    whether the worker belongs above the baseline

  • temporary: (Boolean) (defaults to: false)

Returns:

  • (Thread) 


322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 322

def spawn_worker(temporary: false)
  thread_number = @next_thread_number
  @next_thread_number += 1

  thread = Thread.new(thread_number) do |idx|
    thread_name = String.new("AtomicThreadPool thread #{idx}")
    thread_name << " for #{@name}" if @name
    Thread.current.name = thread_name

    @thread_pool_monitor&.register_worker
    @alive_thread_count.swap { |current_count| current_count + 1 }

    begin
      loop do
        work = nil
        should_exit = false

        @work_available.wait do
          should_exit = temporary && @ruby_cpu_trim_requested.true?
          unless should_exit
            work = @queue.pop
            unless work
              should_exit = @shutdown.true? && @queue.empty?
              should_exit ||= temporary && @idle_trim_requested.true? && @queue.empty?
            end
          end
          work || should_exit
        end

        break if should_exit

        @active_thread_count.swap { |current_count| current_count + 1 }
        if temporary
          work_started_at = Process.clock_gettime(Process::CLOCK_MONOTONIC)
          running_cpu_time_at_start = @thread_pool_monitor.snapshot[6]
        end
        begin
          @thread_pool_monitor&.start_work
          work.call
        rescue => err
          if @on_error
            @on_error.call(err)
          else
            warn "#{thread_name} rescued:"
            warn err.full_message
          end
        ensure
          @thread_pool_monitor&.stop_work
          @active_thread_count.swap { |current_count| current_count - 1 }
          if temporary && !@queue.empty?
            elapsed_time = Process.clock_gettime(Process::CLOCK_MONOTONIC) - work_started_at
            running_cpu_time = @thread_pool_monitor.snapshot[6] - running_cpu_time_at_start
            if running_cpu_time * 2 >= elapsed_time * 1_000_000_000
              @ruby_cpu_trim_requested.make_true
              @work_available.broadcast
            end
          end
        end
      end
    ensure
      @alive_thread_count.swap { |current_count| current_count - 1 }
      @thread_pool_monitor&.unregister_worker
    end
  end

  @threads << thread
  thread
end

#start ⇒ void

This method returns an undefined value.

Starts the worker threads for the thread pool.

This method is called automatically during initialization. It creates the specified number of worker threads and waits for all threads to be fully started before returning.



289
290
291
292
293
294
295
296
297
298
299
300
301
302
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 289

def start
  @size.times { spawn_worker }

  if adaptive?
    @autoscaler = Thread.new do
      thread_name = String.new("AtomicThreadPool autoscaler")
      thread_name << " for #{@name}" if @name
      Thread.current.name = thread_name
      autoscale
    end
  end

  Thread.pass until @alive_thread_count.value == @size
end