Class: FixedThreadPoolManager

Inherits:
Object
  • Object
show all
Includes:
Logging
Defined in:
lib/util/thread_pool_manager.rb

Constant Summary collapse

SHUTDOWN_SIGNAL = 
:shutdown

Constants included from Logging

Logging::DEBUG, Logging::NORMAL, Logging::QUIET, Logging::VERBOSE

Instance Method Summary collapse

Methods included from Logging

#log, #log_stdout, verbosity, verbosity=

Constructor Details

#initialize(percent_available_processors = 0.75) ⇒ FixedThreadPoolManager

Calculate the number of worker threads as 75% of available processors (less one for the monitor thread), with a minimum of 1.

Parameters:

  • percent_available_processors (Float) (defaults to: 0.75) —

    The percentage of available processors to use for worker threads.



12
13
14
15
16
17
18
19
20
21
22
23
24
25
 
# File 'lib/util/thread_pool_manager.rb', line 12

def initialize(percent_available_processors = 0.75)
  log Logging::VERBOSE, "FixedThreadPoolManager#initialize: verbosity is #{Logging.verbosity}"
  if percent_available_processors > 1 || percent_available_processors <= 0
    msg = <<~END_MSG
      Error: The allowable range for the ThreadPool.initialize percent_available_processors is between 0 and 1.
      You provided #{percent_available_processors}.
    END_MSG
    log Logging::QUIET, msg, :red
    exit! 1
  end
  @worker_count = [(Etc.nprocessors * percent_available_processors).floor, 1].max
  @main_work_queue = Queue.new
  @workers = []
end

Instance Method Details

#add_task(task) ⇒ Object

Adds a single task to the work queue. The pool must have been started with start first.



29
30
31
 
# File 'lib/util/thread_pool_manager.rb', line 29

def add_task(task)
  @main_work_queue.push(task)
end

#shutdown ⇒ Object

Signals the pool to shut down after all currently queued tasks are processed. This is a non-blocking method. When you call it, it simply places a special SHUTDOWN_SIGNAL message onto the main work queue. The method returns immediately, allowing your main thread to continue with other tasks. It's like telling the pool, "I'm not going to give you any more tasks, so start wrapping things up when you're done with what you have."



40
41
42
 
# File 'lib/util/thread_pool_manager.rb', line 40

def shutdown
  @main_work_queue.push(SHUTDOWN_SIGNAL)
end

#start ⇒ Object

Starts the workers and the monitor, but does not wait for them to complete. This is for "drip-feeding" tasks.

Parameters:

  • Block —

    of code to execute for each task.

Returns:

  • nil



48
49
50
 
# File 'lib/util/thread_pool_manager.rb', line 48

def start(&)
  initialize_workers(&)
end

#wait_for_completion ⇒ Object

This is the last method to call when using FixedPoolManager. This is a blocking method. It pauses the execution of your main thread and waits until the monitor and all worker threads have fully completed their work and terminated.



56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
 
# File 'lib/util/thread_pool_manager.rb', line 56

def wait_for_completion
  @worker_count.times { @main_work_queue.push(SHUTDOWN_SIGNAL) }

  last_active_count = -1
  loop do
    active_workers = @workers.count(&:alive?)
    break if active_workers.zero?

    if active_workers != last_active_count
      warn format("Waiting for %d worker threads to complete...", active_workers) + "\r" if Logging.verbosity > ::Logging::NORMAL
      last_active_count = active_workers
    end
    begin
      sleep 0.1
    rescue Interrupt
      # This can be interrupted by Ctrl-C. We catch it here to allow the main thread's
      # rescue block to handle the exit gracefully without a stack trace.
    end
  end

  warn (" " * 60) + "\r" # Clear the line
  log Logging::VERBOSE, "All work is complete.", :green
end