@@ -50,14 +50,6 @@ OptionParser.new do |opts| options[:input] = input end - opts.on("-f", "--flush", "Shutdown the server on the socket specified by --socket") do |flush| - options[:flush] = flush - end - - opts.on("-w", "--wipe", "Shutdown the server and terminate the children immediately on the socket specified by --socket") do |wipe| - options[:wipe] = wipe - end - opts.separator "" opts.separator "Server options:" bin/jobby @@ -24,11 +24,7 @@ module Jobby def initialize(options = {}) @options = DEFAULT_OPTIONS.merge options - if @options[:wipe] - @options[:input] = "||JOBBY WIPE||" - elsif @options[:flush] - @options[:input] = "||JOBBY FLUSH||" - elsif @options[:input].nil? + if @options[:input].nil? message "--input not supplied, reading from STDIN (use ctrl-d to end input)" @options[:input] = $stdin.read end @@ -58,20 +54,12 @@ module Jobby def run change_process_ownership begin - if @options[:flush] or @options[:wipe] - if @options[:wipe] - message "Stopping Jobby server and terminating forked children..." - else - message "Stopping Jobby server..." - end - end run_client rescue Errno::EACCES => exception return error(exception.message) rescue Errno::ENOENT, Errno::ECONNREFUSED # Connect failed, fork and start the server process message "There doesn't seem to be a server listening on #{@options[:socket]} - starting one..." if @options[:verbose] - exit if @options[:flush] or @options[:wipe] fork do begin Jobby::Server.new(@options[:socket], @options[:max_child_processes], @options[:log], @options[:prerun_proc]).run(&get_proc_from_options) lib/runner.rb @@ -32,13 +32,14 @@ module Jobby # # ==Stopping the server # - # A client process can send one of two special strings to stop the server. + # There are two built-in ways of stopping the server: # - # "||JOBBY FLUSH||" will stop the server forking any more children and shut - # it down. + # SIGUSR1 will stop the server accepting any more connections, but it + # will continue to fork if there are any requests in the queue. + # It will then wait for the children to exit before terminating. # - # "||JOBBY WIPE||" will stop the server forking any more children, kill 9 - # any existing children and shut it down. + # SIGTERM will stop the server forking any more children, kill 9 any + # existing children and terminate it. # # ==Log rotation # @@ -55,6 +56,7 @@ module Jobby class Server # The log parameter can be either a filepath or an IO object. def initialize(socket_path, max_forked_processes, log, prerun = nil) + $0 = "jobbyd: #{socket_path}" # set the process name @log = log.path rescue log reopen_standard_streams close_fds @@ -62,6 +64,7 @@ module Jobby @socket_path = socket_path @max_forked_processes = max_forked_processes.to_i @queue = Queue.new + setup_signal_handling prerun.call(@logger) unless prerun.nil? end @@ -82,14 +85,7 @@ module Jobby input += bytes end client.close - if input == "||JOBBY FLUSH||" - terminate - elsif input == "||JOBBY WIPE||" - terminate_children - terminate - else - @queue << input - end + @queue << input end end @@ -119,6 +115,27 @@ module Jobby end end + # Traps SIGHUP, SIGTERM and SIGUSR1 for log rotation, immediate shutdown + # and very pleasant shutdown. + def setup_signal_handling + Signal.trap("HUP") do + @logger.info "HUP signal received" + rotate_log + end + Signal.trap("TERM") do + @logger.info "TERM signal received" + @socket.close unless @socket.closed? + @queue.clear + terminate_children + terminate + end + Signal.trap("USR1") do + @logger.info "USR1 signal received" + wait_for_children_to_return + terminate + end + end + # Runs a thread to manage the forked processes. It will block, waiting for a # child to finish if the maximum number of forked processes are already # running. It will then, read from the queue and fork off a new process. @@ -138,8 +155,13 @@ module Jobby # fork and run code that performs the actual work input = @queue.pop @pids << fork do - @socket.close # inherited from the Jobby::Server - $0 = "jobby: child" # set the process name + @socket.close unless @socket.closed? # inherited from the Jobby::Server + # re-trap TERM to simply exit, since it is inherited from the Jobby::Server + Signal.trap("TERM") do + @logger.info "Terminating child process #{Process.pid}" + exit 0 + end + $0 = "jobby: #{@socket_path}" # set the process name @logger.info "Child process started (#{Process.pid})" block.call(input, @logger) exit 0 @@ -180,9 +202,6 @@ module Jobby def start_logging @logger = Logger.new @log @logger.info "Server started at #{Time.now}" - Signal.trap("HUP") do - rotate_log - end end def reap_child @@ -190,16 +209,26 @@ module Jobby end def rotate_log + @logger.info "Rotating log file" @logger.close @logger = Logger.new @log - @logger.info "SIGHUP received, rotating log file" + end + + # Closes the socket and waits for any children to finish before + # terminating. New children that are already in the queue may be + # still be forked at this stage. + def wait_for_children_to_return + @socket.close + while @pids.length > 0 + sleep 1 + end end # Cleans up the server and exits the process with a return code 0. def terminate @queue.clear - @logger.info "Flush received - terminating server" - @socket.close + @logger.info "Terminating server #{Process.pid}" + @socket.close unless @socket.closed? FileUtils.rm(@socket_path, :force => true) exit! 0 end @@ -209,10 +238,10 @@ module Jobby # immediately, perhaps due to 'runaway' children. def terminate_children @queue.clear - @logger.info "Wipe received - terminating forked children" + @logger.info "Terminating forked children" @pids.each do |pid| begin - Process.kill 9, pid + Process.kill 15, pid rescue Errno::ESRCH end end lib/server.rb @@ -1,8 +1,6 @@ require "#{File.dirname(__FILE__)}/../lib/server" require "#{File.dirname(__FILE__)}/../lib/client" require 'fileutils' -$0 = "jobby spec" - # Due to the multi-process nature of these specs, there are a bunch of sleep calls # around. This is, of course, pretty brittle but I can't think of a better way of @@ -40,6 +38,7 @@ describe Jobby::Server do end before :each do + $0 = "jobby spec" run_server(@socket, @max_child_processes, @log_filepath) { File.open(@child_filepath, "a+") do |file| file << "#{Process.pid}" @@ -94,7 +93,7 @@ describe Jobby::Server do run_server(@socket, @max_child_processes, io) {} terminate_server sleep 0.5 - File.readlines(io_filepath).length.should eql(1) + File.readlines(io_filepath).length.should eql(4) FileUtils.rm io_filepath end @@ -102,7 +101,7 @@ describe Jobby::Server do FileUtils.rm @log_filepath Process.kill "HUP", @server_pid sleep 0.2 - File.read(@log_filepath).should match(/SIGHUP received, rotating log file/) + File.read(@log_filepath).should match(/# Logfile created on/) end it "should not run if a block is not given" do @@ -151,35 +150,23 @@ describe Jobby::Server do File.readlines(@child_filepath).length.should eql(@max_child_processes + 2) end - it "should receive a flush command from the client and terminate while the children continue processing" do + it "should receive a USR1 signal then stop accepting connections and terminate after reaping all child PIDs" do terminate_server sleep 1 run_server(@socket, 1, @log_filepath) do - sleep 2 + sleep 3 end 2.times do |i| Jobby::Client.new(@socket) { |c| c.send("hiya") } end - sleep 1 - Jobby::Client.new(@socket) { |c| c.send("||JOBBY FLUSH||") } + sleep 0.5 + Process.kill "USR1", @server_pid sleep 1.5 + lambda { Jobby::Client.new(@socket) { |c| c.send("hello?") } }.should raise_error(Errno::ECONNREFUSED) + sleep 5 lambda { Jobby::Client.new(@socket) { |c| c.send("hello?") } }.should raise_error(Errno::ENOENT) `pgrep -f 'jobby spec' | wc -l`.strip.should eql("2") - end - - it "should receive a wipe command from the client and terminate, taking the children with it" do - terminate_server - run_server(@socket, 1, @log_filepath) do - sleep 2 - end - 2.times do |i| - Jobby::Client.new(@socket) { |c| c.send("hiya") } - end - sleep 1 - Jobby::Client.new(@socket) { |c| c.send("||JOBBY WIPE||") } - sleep 2.5 - lambda { Jobby::Client.new(@socket) { |c| c.send("hello?") } }.should raise_error(Errno::ENOENT) - `pgrep -f 'jobby spec'`.strip.should eql("#{Process.pid}") + `pgrep -f 'jobbyd spec' | wc -l`.strip.should eql("1") end it "should be able to run a Ruby file before any forking" do spec/server_spec.rb cbdc8b2d2e28b57feed013ee1bdb13fc41896b94 Mark Somerville ms@localhost.localdomain http://github.com/Spakman/jobby/commit/afb33ae202cef4039d10ad98c596abcff0d0e964 afb33ae202cef4039d10ad98c596abcff0d0e964 2008-11-12T06:52:21-08:00 2008-11-12T06:50:24-08:00 Removed the special --flush and --wipe options and replaced them with standard UNIX signals: * HUP will rotate the logs * TERM will terminate the process (and the children too, if it is a server) * USR1 will very politely terminate by waiting for the children to complete The process name now includes the related socket and a server process is called jobbyd, not jobby. 698f4c0ac11f828d47195ad4c1ba6b0d3e0591a2 Mark Somerville ms@localhost.localdomain