Class: BoltRb::WorkerPool

Inherits:
Object
  • Object
show all
Defined in:
lib/bolt_rb/worker_pool.rb

Overview

Fixed-size pool of threads that run handler jobs.

The Socket Mode client reads frames on one thread. Handlers that wait on the network must not run on that thread, or pings and later events stall behind them. The App posts each event to this pool and the socket thread returns to reading at once.

Examples:

pool = BoltRb::WorkerPool.new(size: 4, logger: BoltRb.logger)
pool.start
pool.post { do_slow_work }
pool.shutdown

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(size:, logger:, shutdown_timeout: 30) ⇒ WorkerPool

Creates a new pool. Call #start to spawn the threads.

Parameters:

  • size (Integer) —

    Number of worker threads

  • logger (Logger) —

    Logger for job errors

  • shutdown_timeout (Numeric) (defaults to: 30) —

    Seconds to wait for each worker on shutdown



25
26
27
28
29
30
31
32
 
# File 'lib/bolt_rb/worker_pool.rb', line 25

def initialize(size:, logger:, shutdown_timeout: 30)
  @size = size
  @logger = logger
  @shutdown_timeout = shutdown_timeout
  @queue = nil
  @threads = []
  @running = false
end

Instance Attribute Details

#size ⇒ Integer (readonly)

Returns Number of worker threads.

Returns:

  • (Integer) —

    Number of worker threads



18
19
20
 
# File 'lib/bolt_rb/worker_pool.rb', line 18

def size
  @size
end

Instance Method Details

#post { ... } ⇒ void

This method returns an undefined value.

Queues a job for a worker thread

If the pool is not running, the job runs on the calling thread.

Yields:

  • The job to run



51
52
53
54
55
56
57
 
# File 'lib/bolt_rb/worker_pool.rb', line 51

def post(&job)
  if @running
    @queue << job
  else
    run_job(job)
  end
end

#queue_size ⇒ Integer

Returns Number of jobs waiting for a worker.

Returns:

  • (Integer) —

    Number of jobs waiting for a worker



84
85
86
 
# File 'lib/bolt_rb/worker_pool.rb', line 84

def queue_size
  @queue ? @queue.size : 0
end

#running? ⇒ Boolean

Returns Whether the pool accepts jobs.

Returns:

  • (Boolean) —

    Whether the pool accepts jobs



79
80
81
 
# File 'lib/bolt_rb/worker_pool.rb', line 79

def running?
  @running
end

#shutdown ⇒ void

This method returns an undefined value.

Stops accepting jobs, finishes queued jobs, and joins the workers

Workers that do not finish inside the shutdown timeout are left to exit on their own. The pool reports not running either way.



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

def shutdown
  return unless @running

  @running = false
  @queue.close
  @threads.each do |thread|
    next if thread.join(@shutdown_timeout)

    @logger.warn "[WorkerPool] #{thread.name} did not finish within #{@shutdown_timeout}s"
  end
  @threads = []
end

#start ⇒ void

This method returns an undefined value.

Spawns the worker threads



37
38
39
40
41
42
43
 
# File 'lib/bolt_rb/worker_pool.rb', line 37

def start
  return if @running

  @queue = Queue.new
  @running = true
  @threads = Array.new(size) { |index| spawn_worker(index) }
end