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
AUTOSCALE_RUBY_CPU_SHARE =

Returns:

  • (::Float) 
0.75

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



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
135
136
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 107

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
  @capacity_changed = 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:



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

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



239
240
241
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 239

def active_count
  @active_thread_count.value
end

#adaptive? ⇒ true, false

Returns whether the pool may grow beyond its baseline size.

Returns:

  • (true, false) 


330
331
332
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 330

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.



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
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 422

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



186
187
188
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 186

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



208
209
210
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 208

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 three quarters 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) 


502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 502

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 >= elapsed_time * 1_000_000_000 * AUTOSCALE_RUBY_CPU_SHARE
  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:



287
288
289
290
291
292
293
294
295
296
297
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 287

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) 


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
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 343

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 }
        @capacity_changed.broadcast unless @capacity_changed.waiter_count.zero?
        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 }
          @capacity_changed.broadcast unless @capacity_changed.waiter_count.zero?
          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 >= elapsed_time * 1_000_000_000 * AUTOSCALE_RUBY_CPU_SHARE
              @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.



310
311
312
313
314
315
316
317
318
319
320
321
322
323
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 310

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

#wait_for_capacity { ... } ⇒ untyped

Blocks until the given block returns a truthy value, checking it again whenever a worker takes or finishes work.

Useful for applying backpressure, such as holding off on accepting more work until the pool has capacity for it.

Examples:

Wait until no work is queued

pool = AtomicThreadPool.new(size: 2)
5.times { pool << proc { sleep(0.1) } }
pool.wait_for_capacity { pool.queue_length.zero? } #=> true

Yields:

Yield Returns:

  • (untyped) —

    Truthy to stop waiting, falsy to keep waiting

Returns:

  • (untyped) —

    The first truthy value returned by the block



258
259
260
 
# File 'lib/atomic-ruby/atomic_thread_pool.rb', line 258

def wait_for_capacity(&block)
  @capacity_changed.wait(&block)
end