class Bundler::Worker
def abort_threads
def abort_threads @threads.each(&:exit) exit 1 end
def apply_func(obj, i)
def apply_func(obj, i) @func.call(obj, i) rescue Exception => e WrappedException.new(e) 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) @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 = Queue.new @response_queue = Queue.new @func = func @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 end trap("INT") { abort_threads } 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 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 @threads.each { @request_queue.enq POISON } @threads.each(&:join) end