lib/unicorn.rb

   1 require 'logger'
   2
   3 require 'unicorn/socket'
   4 require 'unicorn/const'
   5 require 'unicorn/http_request'
   6 require 'unicorn/http_response'
   7 require 'unicorn/configurator'
   8 require 'unicorn/util'
   9
  10 # Unicorn module containing all of the classes (include C extensions) for running
  11 # a Unicorn web server.  It contains a minimalist HTTP server with just enough
  12 # functionality to service web application requests fast as possible.
  13 module Unicorn
  14   class << self
  15     def run(app, options = {})
  16       HttpServer.new(app, options).start.join
  17     end
  18   end
  19
  20   # This is the process manager of Unicorn. This manages worker
  21   # processes which in turn handle the I/O and application process.
  22   # Listener sockets are started in the master process and shared with
  23   # forked worker children.
  24   class HttpServer
  25     attr_reader :logger
  26     include ::Unicorn::SocketHelper
  27
  28     DEFAULT_START_CTX = {
  29       :argv => ARGV.map { |arg| arg.dup },
  30       # don't rely on Dir.pwd here since it's not symlink-aware, and
  31       # symlink dirs are the default with Capistrano...
  32       :cwd => `/bin/sh -c pwd`.chomp("\n"),
  33       :zero => $0.dup,
  34       :environ => {}.merge!(ENV),
  35       :umask => File.umask,
  36     }.freeze
  37
  38     Worker = Struct.new(:nr, :tempfile) unless defined?(Worker)
  39     class Worker
  40       # worker objects may be compared to just plain numbers
  41       def ==(other_nr)
  42         self.nr == other_nr
  43       end
  44     end
  45
  46     # Creates a working server on host:port (strange things happen if
  47     # port isn't a Number).  Use HttpServer::run to start the server and
  48     # HttpServer.workers.join to join the thread that's processing
  49     # incoming requests on the socket.
  50     def initialize(app, options = {})
  51       start_ctx = options.delete(:start_ctx)
  52       @start_ctx = DEFAULT_START_CTX.dup
  53       @start_ctx.merge!(start_ctx) if start_ctx
  54       @app = app
  55       @sig_queue = []
  56       @master_pid = $$
  57       @workers = Hash.new
  58       @io_purgatory = [] # prevents IO objects in here from being GC-ed
  59       @request = @rd_sig = @wr_sig = nil
  60       @reexec_pid = 0
  61       @config = Configurator.new(options.merge(:use_defaults => true))
  62       @config.commit!(self, :skip => [:listeners, :pid])
  63       @listeners = []
  64     end
  65
  66     # Runs the thing.  Returns self so you can run join on it
  67     def start
  68       BasicSocket.do_not_reverse_lookup = true
  69
  70       # inherit sockets from parents, they need to be plain Socket objects
  71       # before they become UNIXServer or TCPServer
  72       inherited = ENV['UNICORN_FD'].to_s.split(/,/).map do |fd|
  73         io = Socket.for_fd(fd.to_i)
  74         set_server_sockopt(io)
  75         @io_purgatory << io
  76         logger.info "inherited: #{io} fd=#{fd} addr=#{sock_name(io)}"
  77         server_cast(io)
  78       end
  79
  80       config_listeners = @config[:listeners].dup
  81       @listeners.replace(inherited)
  82
  83       # we start out with generic Socket objects that get cast to either
  84       # TCPServer or UNIXServer objects; but since the Socket objects
  85       # share the same OS-level file descriptor as the higher-level *Server
  86       # objects; we need to prevent Socket objects from being garbage-collected
  87       config_listeners -= listener_names
  88       config_listeners.each { |addr| listen(addr) }
  89       raise ArgumentError, "no listeners" if @listeners.empty?
  90       self.pid = @config[:pid]
  91       build_app! if @preload_app
  92       $stderr.reopen(File.open(@stderr_path, "a")) if @stderr_path
  93       $stdout.reopen(File.open(@stdout_path, "a")) if @stdout_path
  94       $stderr.sync = $stdout.sync = true
  95       spawn_missing_workers
  96       self
  97     end
  98
  99     # replaces current listener set with +listeners+.  This will
 100     # close the socket if it will not exist in the new listener set
 101     def listeners=(listeners)
 102       cur_names = listener_names
 103       set_names = listener_names(listeners)
 104       dead_names = cur_names - set_names
 105
 106       @listeners.delete_if do |io|
 107         if dead_names.include?(sock_name(io))
 108           @io_purgatory.delete_if { |pio| pio.fileno == io.fileno }
 109           true
 110         else
 111           false
 112         end
 113       end
 114
 115       (set_names - cur_names).each { |addr| listen(addr) }
 116     end
 117
 118     # sets the path for the PID file of the master process
 119     def pid=(path)
 120       if path
 121         if x = valid_pid?(path)
 122           return path if @pid && path == @pid && x == $$
 123           raise ArgumentError, "Already running on PID:#{x} " \
 124                                "(or pid=#{path} is stale)"
 125         end
 126         File.open(path, 'wb') { |fp| fp.syswrite("#{$$}\n") }
 127       end
 128       unlink_pid_safe(@pid) if @pid && @pid != path
 129       @pid = path
 130     end
 131
 132     # add a given address to the +listeners+ set, idempotently
 133     # Allows workers to add a private, per-process listener via the
 134     # @after_fork hook.  Very useful for debugging and testing.
 135     def listen(address)
 136       return if String === address && listener_names.include?(address)
 137
 138       if io = bind_listen(address, { :backlog => @backlog })
 139         if Socket == io.class
 140           @io_purgatory << io
 141           io = server_cast(io)
 142         end
 143         logger.info "#{io} listening on PID:#{$$} " \
 144                     "fd=#{io.fileno} addr=#{sock_name(io)}"
 145         @listeners << io
 146       else
 147         logger.error "adding listener failed addr=#{address} (in use)"
 148         raise Errno::EADDRINUSE, address
 149       end
 150     end
 151
 152     # monitors children and receives signals forever
 153     # (or until a termination signal is sent).  This handles signals
 154     # one-at-a-time time and we'll happily drop signals in case somebody
 155     # is signalling us too often.
 156     def join
 157       # this pipe is used to wake us up from select(2) in #join when signals
 158       # are trapped.  See trap_deferred
 159       @rd_sig, @wr_sig = IO.pipe unless (@rd_sig && @wr_sig)
 160       @rd_sig.nonblock = @wr_sig.nonblock = true
 161       mode = nil
 162       respawn = true
 163
 164       QUEUE_SIGS.each { |sig| trap_deferred(sig) }
 165       trap('CHLD') { |sig_nr| awaken_master }
 166       $0 = "unicorn master"
 167       logger.info "master process ready" # test_exec.rb relies on this message
 168       begin
 169         loop do
 170           reap_all_workers
 171           case (mode = @sig_queue.shift)
 172           when nil
 173             murder_lazy_workers
 174             spawn_missing_workers if respawn
 175             master_sleep
 176           when 'QUIT' # graceful shutdown
 177             break
 178           when 'TERM', 'INT' # immediate shutdown
 179             stop(false)
 180             break
 181           when 'USR1' # rotate logs
 182             logger.info "master rotating logs..."
 183             Unicorn::Util.reopen_logs
 184             logger.info "master done rotating logs"
 185             kill_each_worker('USR1')
 186           when 'USR2' # exec binary, stay alive in case something went wrong
 187             reexec
 188           when 'WINCH'
 189             if Process.ppid == 1 || Process.getpgrp != $$
 190               respawn = false
 191               logger.info "gracefully stopping all workers"
 192               kill_each_worker('QUIT')
 193             else
 194               logger.info "SIGWINCH ignored because we're not daemonized"
 195             end
 196           when 'HUP'
 197             respawn = true
 198             if @config.config_file
 199               load_config!
 200               redo # immediate reaping since we may have QUIT workers
 201             else # exec binary and exit if there's no config file
 202               logger.info "config_file not present, reexecuting binary"
 203               reexec
 204               break
 205             end
 206           else
 207             logger.error "master process in unknown mode: #{mode}"
 208           end
 209         end
 210       rescue Errno::EINTR
 211         retry
 212       rescue Object => e
 213         logger.error "Unhandled master loop exception #{e.inspect}."
 214         logger.error e.backtrace.join("\n")
 215         retry
 216       end
 217       stop # gracefully shutdown all workers on our way out
 218       logger.info "master PID:#{$$} join complete"
 219       unlink_pid_safe(@pid) if @pid
 220     end
 221
 222     # Terminates all workers, but does not exit master process
 223     def stop(graceful = true)
 224       kill_each_worker(graceful ? 'QUIT' : 'TERM')
 225       timeleft = @timeout
 226       step = 0.2
 227       reap_all_workers
 228       until @workers.empty?
 229         sleep(step)
 230         reap_all_workers
 231         (timeleft -= step) > 0 and next
 232         kill_each_worker('KILL')
 233       end
 234     ensure
 235       self.listeners = []
 236     end
 237
 238     private
 239
 240     # list of signals we care about and trap in master.
 241     QUEUE_SIGS =
 242       %w(WINCH QUIT INT TERM USR1 USR2 HUP).map { |x| x.freeze }.freeze
 243
 244     # defer a signal for later processing in #join (master process)
 245     def trap_deferred(signal)
 246       trap(signal) do |sig_nr|
 247         if @sig_queue.size < 5
 248           @sig_queue << signal
 249           awaken_master
 250         else
 251           logger.error "ignoring SIG#{signal}, queue=#{@sig_queue.inspect}"
 252         end
 253       end
 254     end
 255
 256     # wait for a signal hander to wake us up and then consume the pipe
 257     # Wake up every second anyways to run murder_lazy_workers
 258     def master_sleep
 259       begin
 260         ready = IO.select([@rd_sig], nil, nil, 1)
 261         ready && ready[0] && ready[0][0] or return
 262         loop { @rd_sig.sysread(Const::CHUNK_SIZE) }
 263       rescue Errno::EAGAIN, Errno::EINTR
 264       end
 265     end
 266
 267     def awaken_master
 268       begin
 269         @wr_sig.syswrite('.') # wakeup master process from IO.select
 270       rescue Errno::EAGAIN # pipe is full, master should wake up anyways
 271       rescue Errno::EINTR
 272         retry
 273       end
 274     end
 275
 276     # reaps all unreaped workers
 277     def reap_all_workers
 278       begin
 279         loop do
 280           pid, status = Process.waitpid2(-1, Process::WNOHANG)
 281           pid or break
 282           if @reexec_pid == pid
 283             logger.error "reaped #{status.inspect} exec()-ed"
 284             @reexec_pid = 0
 285             self.pid = @pid.chomp('.oldbin') if @pid
 286             $0 = "unicorn master"
 287           else
 288             worker = @workers.delete(pid)
 289             worker.tempfile.close rescue nil
 290             logger.info "reaped #{status.inspect} " \
 291                         "worker=#{worker.nr rescue 'unknown'}"
 292           end
 293         end
 294       rescue Errno::ECHILD
 295       end
 296     end
 297
 298     # reexecutes the @start_ctx with a new binary
 299     def reexec
 300       if @reexec_pid > 0
 301         begin
 302           Process.kill(0, @reexec_pid)
 303           logger.error "reexec-ed child already running PID:#{@reexec_pid}"
 304           return
 305         rescue Errno::ESRCH
 306           @reexec_pid = 0
 307         end
 308       end
 309
 310       if @pid
 311         old_pid = "#{@pid}.oldbin"
 312         prev_pid = @pid.dup
 313         begin
 314           self.pid = old_pid  # clear the path for a new pid file
 315         rescue ArgumentError
 316           logger.error "old PID:#{valid_pid?(old_pid)} running with " \
 317                        "existing pid=#{old_pid}, refusing rexec"
 318           return
 319         rescue Object => e
 320           logger.error "error writing pid=#{old_pid} #{e.class} #{e.message}"
 321           return
 322         end
 323       end
 324
 325       @reexec_pid = fork do
 326         @rd_sig.close if @rd_sig
 327         @wr_sig.close if @wr_sig
 328         @workers.values.each { |other| other.tempfile.close rescue nil }
 329
 330         ENV.replace(@start_ctx[:environ])
 331         ENV['UNICORN_FD'] = @listeners.map { |sock| sock.fileno }.join(',')
 332         File.umask(@start_ctx[:umask])
 333         Dir.chdir(@start_ctx[:cwd])
 334         cmd = [ @start_ctx[:zero] ] + @start_ctx[:argv]
 335         logger.info "executing #{cmd.inspect} (in #{Dir.pwd})"
 336         @before_exec.call(self) if @before_exec
 337         exec(*cmd)
 338       end
 339       $0 = "unicorn master (old)"
 340     end
 341
 342     # forcibly terminate all workers that haven't checked in in @timeout
 343     # seconds.  The timeout is implemented using an unlinked tempfile
 344     # shared between the parent process and each worker.  The worker
 345     # runs File#chmod to modify the ctime of the tempfile.  If the ctime
 346     # is stale for >@timeout seconds, then we'll kill the corresponding
 347     # worker.
 348     def murder_lazy_workers
 349       now = Time.now
 350       @workers.each_pair do |pid, worker|
 351         (now - worker.tempfile.ctime) <= @timeout and next
 352         logger.error "worker=#{worker.nr} PID:#{pid} is too old, killing"
 353         kill_worker('KILL', pid) # take no prisoners for @timeout violations
 354         worker.tempfile.close rescue nil
 355       end
 356     end
 357
 358     def spawn_missing_workers
 359       return if @workers.size == @worker_processes
 360       (0...@worker_processes).each do |worker_nr|
 361         @workers.values.include?(worker_nr) and next
 362         begin
 363           Dir.chdir(@start_ctx[:cwd])
 364         rescue Errno::ENOENT => err
 365           logger.fatal "#{err.inspect} (#{@start_ctx[:cwd]})"
 366           @sig_queue << 'QUIT' # forcibly emulate SIGQUIT
 367           return
 368         end
 369         tempfile = Tempfile.new('') # as short as possible to save dir space
 370         tempfile.unlink # don't allow other processes to find or see it
 371         tempfile.sync = true
 372         worker = Worker.new(worker_nr, tempfile)
 373         @before_fork.call(self, worker.nr)
 374         pid = fork { worker_loop(worker) }
 375         @workers[pid] = worker
 376       end
 377     end
 378
 379     # once a client is accepted, it is processed in its entirety here
 380     # in 3 easy steps: read request, call app, write app response
 381     def process_client(client)
 382       env = @request.read(client) or return
 383       app_response = @app.call(env)
 384       HttpResponse.write(client, app_response)
 385     rescue EOFError,Errno::ECONNRESET,Errno::EPIPE,Errno::EINVAL,Errno::EBADF
 386       client.closed? or client.close rescue nil
 387     rescue Object => e
 388       logger.error "Read error: #{e.inspect}"
 389       logger.error e.backtrace.join("\n")
 390     ensure
 391       begin
 392         client.closed? or client.close
 393       rescue Object => e
 394         logger.error "Client error: #{e.inspect}"
 395         logger.error e.backtrace.join("\n")
 396       end
 397       @request.reset
 398     end
 399
 400     # gets rid of stuff the worker has no business keeping track of
 401     # to free some resources and drops all sig handlers.
 402     # traps for USR1, USR2, and HUP may be set in the @after_fork Proc
 403     # by the user.
 404     def init_worker_process(worker)
 405       build_app! unless @preload_app
 406       @sig_queue.clear
 407       QUEUE_SIGS.each { |sig| trap(sig, 'IGNORE') }
 408       trap('CHLD', 'DEFAULT')
 409
 410       $0 = "unicorn worker[#{worker.nr}]"
 411       @rd_sig.close if @rd_sig
 412       @wr_sig.close if @wr_sig
 413       @workers.values.each { |other| other.tempfile.close rescue nil }
 414       @workers.clear
 415       @start_ctx.clear
 416       @start_ctx = @workers = @rd_sig = @wr_sig = nil
 417       @listeners.each { |sock| set_cloexec(sock) }
 418       ENV.delete('UNICORN_FD')
 419       @after_fork.call(self, worker.nr) if @after_fork
 420       @request = HttpRequest.new(logger)
 421     end
 422
 423     # runs inside each forked worker, this sits around and waits
 424     # for connections and doesn't die until the parent dies (or is
 425     # given a INT, QUIT, or TERM signal)
 426     def worker_loop(worker)
 427       init_worker_process(worker)
 428       nr = 0
 429       tempfile = worker.tempfile
 430       alive = true
 431       ready = @listeners
 432       client = nil
 433       %w(TERM INT).each { |sig| trap(sig) { exit(0) } } # instant shutdown
 434       trap('QUIT') do
 435         alive = false
 436         @listeners.each { |sock| sock.close rescue nil } # break IO.select
 437       end
 438       reopen_logs, (rd, wr) = false, IO.pipe
 439       trap(:USR1) { reopen_logs = true; rd.close rescue nil }
 440       @logger.info "worker=#{worker.nr} ready"
 441
 442       while alive && @master_pid == Process.ppid
 443         if reopen_logs
 444           reopen_logs = false
 445           @logger.info "worker=#{worker.nr} rotating logs..."
 446           Unicorn::Util.reopen_logs
 447           @logger.info "worker=#{worker.nr} done rotating logs"
 448           wr.close rescue nil
 449           rd, wr = IO.pipe
 450         end
 451         # we're a goner in @timeout seconds anyways if tempfile.chmod
 452         # breaks, so don't trap the exception.  Using fchmod() since
 453         # futimes() is not available in base Ruby and I very strongly
 454         # prefer temporary files to be unlinked for security,
 455         # performance and reliability reasons, so utime is out.  No-op
 456         # changes with chmod doesn't update ctime on all filesystems; so
 457         # we increment our counter each and every time.
 458         tempfile.chmod(nr += 1)
 459
 460         begin
 461           accepted = false
 462           ready.each do |sock|
 463             begin
 464               client = begin
 465                 sock.accept_nonblock
 466               rescue Errno::EAGAIN
 467                 next
 468               end
 469               accepted = client.sync = true
 470               client.nonblock = false
 471               set_client_sockopt(client) if TCPSocket === client
 472               process_client(client)
 473             rescue Errno::ECONNABORTED
 474               # client closed the socket even before accept
 475               if client && !client.closed?
 476                 client.close rescue nil
 477               end
 478             end
 479             tempfile.chmod(nr += 1)
 480             break if reopen_logs
 481           end
 482           client = nil
 483
 484           # make the following bet: if we accepted clients this round,
 485           # we're probably reasonably busy, so avoid calling select(2)
 486           # and try to do a blind non-blocking accept(2) on everything
 487           # before we sleep again in select
 488           if accepted || reopen_logs
 489             ready = @listeners
 490           else
 491             begin
 492               tempfile.chmod(nr += 1)
 493               # timeout used so we can detect parent death:
 494               ret = IO.select(@listeners, nil, [rd], @timeout/2.0) or next
 495               ready = ret[0]
 496             rescue Errno::EINTR
 497               ready = @listeners
 498             rescue Errno::EBADF => e
 499               reopen_logs or exit(alive ? 1 : 0)
 500             end
 501           end
 502         rescue SystemExit => e
 503           exit(e.status)
 504         rescue Object => e
 505           if alive
 506             logger.error "Unhandled listen loop exception #{e.inspect}."
 507             logger.error e.backtrace.join("\n")
 508           end
 509         end
 510       end
 511     end
 512
 513     # delivers a signal to a worker and fails gracefully if the worker
 514     # is no longer running.
 515     def kill_worker(signal, pid)
 516       begin
 517         Process.kill(signal, pid)
 518       rescue Errno::ESRCH
 519         worker = @workers.delete(pid) and worker.tempfile.close rescue nil
 520       end
 521     end
 522
 523     # delivers a signal to each worker
 524     def kill_each_worker(signal)
 525       @workers.keys.each { |pid| kill_worker(signal, pid) }
 526     end
 527
 528     # unlinks a PID file at given +path+ if it contains the current PID
 529     # useful as an at_exit handler.
 530     def unlink_pid_safe(path)
 531       (File.read(path).to_i == $$ and File.unlink(path)) rescue nil
 532     end
 533
 534     # returns a PID if a given path contains a non-stale PID file,
 535     # nil otherwise.
 536     def valid_pid?(path)
 537       if File.exist?(path) && (pid = File.read(path).to_i) > 1
 538         begin
 539           Process.kill(0, pid)
 540           return pid
 541         rescue Errno::ESRCH
 542         end
 543       end
 544       nil
 545     end
 546
 547     def load_config!
 548       begin
 549         logger.info "reloading config_file=#{@config.config_file}"
 550         @config.reload
 551         @config.commit!(self)
 552         kill_each_worker('QUIT')
 553         logger.info "done reloading config_file=#{@config.config_file}"
 554       rescue Object => e
 555         logger.error "error reloading config_file=#{@config.config_file}: " \
 556                      "#{e.class} #{e.message}"
 557       end
 558     end
 559
 560     # returns an array of string names for the given listener array
 561     def listener_names(listeners = @listeners)
 562       listeners.map { |io| sock_name(io) }
 563     end
 564
 565     def build_app!
 566       @app = @app.call if @app.respond_to?(:arity) && @app.arity == 0
 567     end
 568
 569   end
 570 end