Class: Plushie::ThreadPool

Inherits:
Object
  • Object
show all
Defined in:
lib/plushie/thread_pool.rb,
sig/plushie/thread_pool.rbs

Overview

Simple bounded thread pool for background work.

Worker threads pull tasks from a shared queue. Sized to CPU count by default. Used for non-cancellable background operations.

Note: Command.async and Command.stream spawn dedicated threads (not from this pool) because they need individual cancel handles. The pool is used by the test framework and other non-cancellable work.

Examples:

pool = ThreadPool.new(size: 4)
pool.post { expensive_work() }
pool.shutdown

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(size: nil) ⇒ ThreadPool

Create a new thread pool.

Parameters:

  • size (Integer, nil) (defaults to: nil) —

    number of workers (default: CPU count)

  • size: (Integer, nil) (defaults to: nil) 


27
28
29
30
31
32
 
# File 'lib/plushie/thread_pool.rb', line 27

def initialize(size: nil)
  @size = size || default_size
  @queue = Thread::Queue.new
  @shutdown = false
  @workers = @size.times.map { spawn_worker }
end

Instance Attribute Details

#size ⇒ Integer (readonly)

Returns number of worker threads.

Returns:

  • (Integer) —

    number of worker threads



22
23
24
 
# File 'lib/plushie/thread_pool.rb', line 22

def size
  @size
end

Instance Method Details

#default_size ⇒ Integer

Returns:

  • (Integer) 


79
80
81
 
# File 'lib/plushie/thread_pool.rb', line 79

def default_size
  Etc.respond_to?(:nprocessors) ? [Etc.nprocessors, 2].max : 4
end

#post { ... }

This method returns an undefined value.

Queue a block for execution by a worker thread.

Yields:

  • the work to execute

Yield Returns:

  • (void)

Raises:

  • (RuntimeError) —

    if the pool has been shut down



38
39
40
41
 
# File 'lib/plushie/thread_pool.rb', line 38

def post(&block)
  raise "ThreadPool is shut down" if @shutdown
  @queue.push(block)
end

#shutdown(timeout: 5)

This method returns an undefined value.

Shut down the pool. Signals all workers to stop and waits for them to finish current work.

Parameters:

  • timeout (Numeric) (defaults to: 5) —

    max seconds to wait per worker (default: 5)

  • timeout: (Numeric) (defaults to: 5) 


47
48
49
50
51
52
53
54
55
56
 
# File 'lib/plushie/thread_pool.rb', line 47

def shutdown(timeout: 5)
  return if @shutdown
  @shutdown = true
  @size.times { @queue.push(:shutdown) }
  deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout
  @workers.each do |t|
    remaining = deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
    t.join([remaining, 0].max)
  end
end

#shutdown? ⇒ Boolean

Returns true if the pool has been shut down.

Returns:

  • (Boolean) —

    true if the pool has been shut down



59
60
61
 
# File 'lib/plushie/thread_pool.rb', line 59

def shutdown?
  @shutdown
end

#spawn_worker ⇒ Thread

Returns:

  • (Thread) 


65
66
67
68
69
70
71
72
73
74
75
76
77
 
# File 'lib/plushie/thread_pool.rb', line 65

def spawn_worker
  Thread.new do
    while (work = @queue.pop) != :shutdown
      begin
        work.call
      rescue => e
        # Worker exceptions are swallowed to keep the pool alive.
        # The caller is responsible for error handling in their block.
        warn "plushie: thread pool worker error: #{e.class}: #{e.message}"
      end
    end
  end
end