@@ -19,3 +19,9 @@ end def debug(message) puts '[PID %d debug] %s' % [$$, message] end + +# bind a signal handler to multiple signals +def traps(*args, &cmd) + cmd = args.pop unless block_given? + args.each { |signal| trap signal, cmd } +end sequement.rb @@ -10,53 +10,75 @@ module Sequement SOCKET_TIMEOUT = 2 SOCKET_BACKLOG = 10 + # host, port: listen for TCP connection + # dir: directory to read/write sequences + # concurrency: number of workers to fork def initialize(host, port, dir, concurrency) @host, @port = host, port @dir = dir @concurrency = concurrency - @pipes_in, @pipes_out = {}, {} - @sequences = {} - @pipe_sig_read, @pipe_sig_write = IO.pipe - @pipe_writer_read, @pipe_writer_write = IO.pipe - @writer = Sequement::Writer.new(@pipe_writer_read) - signal_init end + # Starts the server: + # * forks writer and workers + # * monitors IPC pipes until interrupted. def start - #debug 'creating listening socket' + @writer = Sequement::Writer.new.start + + @sequences = {} + @pipes_in, @pipes_out = {}, {} + @pipe_sig = Pipe.new + + traps :INT, :TERM do + traps :INT, :TERM, 'DEFAULT' + puts "PID #$$ shutting down..." + @pipe_sig.writer.putc 0 + end + @socket = create_listen_socket - trap('EXIT') { @socket.close } # applies to parent & worker processes + trap(:EXIT) { @socket.close } + + master_loop + wait_for_workers + shutdown_writer + end + + ####### + private + + # Spawns workers as needed to fulfill @concurrency. + # Uses select() to monitor IPC and signal pipes. + def master_loop + pipe_sig_reader = @pipe_sig.reader loop do spawn_up_to_concurrency - break if @pipes_in.empty? - #debug 'select() on %d read pipes' % @pipes_in.length - if selected = IO.select(@pipes_in.values + [@pipe_sig_read], nil, nil, SOCKET_TIMEOUT) - break if selected.first.include? @pipe_sig_read + pipes = @pipes_in.values + [pipe_sig_reader] + #debug 'select() on %d read pipes' % pipes.length + if selected = IO.select(pipes, nil, nil, SOCKET_TIMEOUT) + break if selected.first.include? pipe_sig_reader selected.first.each { |pipe| read_pipe pipe } end end + end - debug "waiting for worker processes.." + def wait_for_workers + debug "waiting for workers %s to exit.." % @pipes_in.keys.join(', ') until @pipes_in.empty? do pid = Process.wait - @pipes_in.delete pid - @pipes_out.delete pid + [@pipes_in, @pipes_out].each { |pipe| pipe.delete pid } end + end - debug 'writing any sequences to disk' + # Persists actual value for each known sequence. + # Instructs writer to stop its child process, waits for it to exit. + def shutdown_writer + debug 'writing sequences to disk, stopping writer' @sequences.each_value { |seq| seq.save_sequence } - - debug 'stopping writer' - @pipe_writer_write.putc 0 - Process.waitall - + @writer.stop end - ####### - private - def read_pipe(pipe) pid = @pipes_in.index(pipe) @@ -74,7 +96,7 @@ module Sequement when COMMAND[:next] length = pipe.getc - seq_name = pipe.read(length) + seq_name = pipe.read length #debug 'seq_name: %s' % seq_name @pipes_out[pid].puts sequence(seq_name).next @@ -84,6 +106,7 @@ module Sequement else raise "Unrecognized command from pipe: %d" % command + end end @@ -100,39 +123,22 @@ module Sequement # fork workers until configured concurrency def spawn_up_to_concurrency while @pipes_in.length < CONCURRENCY - fork_worker - end - end - - # Forks a single worker, opening a pair of IPC pipes - def fork_worker - pipe_worker_to_master = Pipe.new - pipe_master_to_worker = Pipe.new - if pid = fork - #debug 'forked worker: PID %d' % pid - @pipes_in[pid] = pipe_worker_to_master.reader! - @pipes_out[pid] = pipe_master_to_worker.writer! - else - Worker.new( - @socket, - pipe_worker_to_master, - pipe_master_to_worker, - @pipe_sig_read - ).run - exit - end - end - - def traps(*args, &cmd) - cmd = args.pop unless block_given? - args.each { |signal| trap signal, cmd } - end - - def signal_init - traps :INT, :TERM do - traps :INT, :TERM, 'DEFAULT' - puts "PID #$$ shutting down..." - @pipe_sig_write.putc 0 + worker_to_master = Pipe.new + master_to_worker = Pipe.new + if pid = fork + #debug 'forked worker: PID %d' % pid + @pipes_in[pid] = worker_to_master.reader! + @pipes_out[pid] = master_to_worker.writer! + else + $0 = 'sequement_worker' + Worker.new( + @socket, + worker_to_master.writer!, + master_to_worker.reader!, + @pipe_sig.reader! + ).start + exit + end end end sequement/master.rb @@ -8,17 +8,18 @@ module Sequement def initialize(acceptor, pipe_out, pipe_in, pipe_sig) - trap :INT, 'IGNORE' - trap :TERM, 'DEFAULT' - @acceptor = acceptor - @pipe_out = pipe_out.writer! - @pipe_in = pipe_in.reader! + @pipe_out = pipe_out + @pipe_in = pipe_in @pipe_sig = pipe_sig end - def run + def start + + trap :INT, 'IGNORE' + trap :TERM, 'DEFAULT' + loop do if selected = IO.select([@acceptor, @pipe_sig], nil, nil, TIMEOUT) sequement/worker.rb @@ -2,12 +2,15 @@ module Sequement class Writer - attr_reader :pid - - def initialize(pipe_sig) - @pipe_sig = pipe_sig - @pipe = Pipe.new - fork_writer + def start + @pipe, @pipe_sig = Pipe.new, Pipe.new + if @pid = fork + initialize_parent + else + initialize_child + exit + end + self end def write(path, data) @@ -15,23 +18,27 @@ module Sequement @pipe.puts data end + def stop + @pipe_sig.putc 0 + Process.waitpid @pid + end + ####### private - def fork_writer - if @pid = fork + def initialize_parent #debug 'forked writer: PID %d' % @pid @pipe.writer! - else + @pipe_sig.writer! + end + + def initialize_child + traps :INT, :TERM, 'IGNORE' $0 = 'sequement_writer' @pipe.reader! - signal_init + @pipe_sig.reader! select_loop - end - end - - def signal_init - trap :INT, 'IGNORE' + debug 'writer stopped' end def select_loop @@ -57,9 +64,6 @@ module Sequement end - debug 'writer stopped' - exit - end end sequement/writer.rb 2ddf4ef6c61b55f700d159e611983ffe9b9ecfea Paul Annesley paul@annesley.cc http://github.com/pda/sequement/commit/c138f891d2270ef0482b8c650daa93ee4081db0a c138f891d2270ef0482b8c650daa93ee4081db0a 2009-11-11T06:05:47-08:00 2009-11-11T06:05:47-08:00 General code tidying and documenting. e81a4aa269ce64891617f67b53aaefe4a4a58d95 Paul Annesley paul@annesley.cc