Sha256: bc545cc17e0aab66fd066c34de09727aa865878d1c9f8329a35711c465004ced
Contents?: true
Size: 1.61 KB
Versions: 4
Compression:
Stored size: 1.61 KB
Contents
require 'rbbt/util/concurrency/processes/socket'
class RbbtProcessQueue
class RbbtProcessQueueWorker
attr_reader :pid, :queue, :callback_queue, :cleanup, :block
def initialize(queue, callback_queue = nil, cleanup = nil, &block)
@queue, @callback_queue, @cleanup, @block = queue, callback_queue, cleanup, block
@pid = Process.fork do
begin
@cleanup.call if @cleanup
@queue.close_write
if @callback_queue
Misc.purge_pipes(@callback_queue.swrite)
@callback_queue.close_read
else
Misc.purge_pipes
end
Signal.trap(:INT){ raise Aborted; }
loop do
p = @queue.pop
next if p.nil?
raise p if Exception === p
raise p.first if Exception === p.first
res = @block.call *p
@callback_queue.push res if @callback_queue
end
Kernel.exit! 0
rescue ClosedStream
rescue Aborted
Log.error "Worker #{Process.pid} aborted"
Kernel.exit! -1
rescue Exception
@callback_queue.push($!) if @callback_queue
Kernel.exit! -1
ensure
@callback_queue.close_write if @callback_queue
end
end
end
def join
begin
joined_pid = Process.waitpid @pid
rescue
end
end
def abort
begin
Process.kill :INT, @pid
rescue
end
end
def done?
begin
Process.waitpid @pid, Process::WNOHANG
rescue Errno::ECHILD
true
rescue
false
end
end
end
end
Version data entries
4 entries across 4 versions & 1 rubygems