class Bundler::Worker
def abort_threads
def abort_threads Bundler.ui.debug("\n#{caller.join("\n")}") @threads.each(&:exit) exit 1 end
def add_interrupt_handler
def add_interrupt_handler @previous_interrupt_handler = trap("INT") { abort_threads } end
def apply_func(obj, i)
def apply_func(obj, i) @func.call(obj, i) rescue Exception => e # rubocop:disable Lint/RescueException WrappedException.new(e) end
def create_threads
def create_threads creation_errors = [] @threads = Array.new(@size) do |i| Thread.start { process_queue(i) }.tap do |thread| thread.name = "#{name} Worker ##{i}" if thread.respond_to?(:name=) end rescue ThreadError => e creation_errors << e nil end.compact add_interrupt_handler unless @threads.empty? return if creation_errors.empty? message = "Failed to create threads for the #{name} worker: #{creation_errors.map(&:to_s).uniq.join(", ")}" raise ThreadCreationError, message if @threads.empty? Bundler.ui.info message end
def deq
def deq result = @response_queue.deq raise result.exception if result.is_a?(WrappedException) result end
def enq(obj)
-
obj
(String
) -- mostly it is name of spec that should be downloaded
def enq(obj) create_threads unless @threads @request_queue.enq obj end
def initialize(size, name, func)
-
func
(Proc
) -- job to run in inside the worker pool -
name
(String
) -- name the name of the worker -
size
(Integer
) -- Size of pool
def initialize(size, name, func) @name = name @request_queue = Thread::Queue.new @response_queue = Thread::Queue.new @func = func @size = size @threads = nil @previous_interrupt_handler = nil end
def process_queue(i)
def process_queue(i) loop do obj = @request_queue.deq break if obj.equal? POISON @response_queue.enq apply_func(obj, i) end end
def remove_interrupt_handler
def remove_interrupt_handler return unless @previous_interrupt_handler trap "INT", @previous_interrupt_handler end
def stop
def stop stop_threads end
def stop_threads
Stop the worker threads by sending a poison object down the request queue
def stop_threads return unless @threads @threads.each { @request_queue.enq POISON } @threads.each(&:join) remove_interrupt_handler @threads = nil end