puma--puma/lib/puma/server.rb

require 'stringio'

require 'puma/thread_pool'
require 'puma/const'
require 'puma/events'
require 'puma/null_io'
require 'puma/compat'
require 'puma/reactor'
require 'puma/client'
require 'puma/binder'
require 'puma/delegation'
require 'puma/accept_nonblock'
require 'puma/util'

require 'puma/puma_http11'

unless Puma.const_defined? "IOBuffer"
  require 'puma/io_buffer'
end

require 'socket'

module Puma

  # The HTTP Server itself. Serves out a single Rack app.
  #
  # This class is used by the `Puma::Single` and `Puma::Cluster` classes
  # to generate one or more `Puma::Server` instances capable of handling requests.
  # Each Puma process will contain one `Puma::Server` instacne.
  #
  # The `Puma::Server` instance pulls requests from the socket, adds them to a
  # `Puma::Reactor` where they get eventually passed to a `Puma::ThreadPool`.
  #
  # Each `Puma::Server` will have one reactor and one thread pool.
  class Server

    include Puma::Const
    extend  Puma::Delegation

    attr_reader :thread
    attr_reader :events
    attr_accessor :app

    attr_accessor :min_threads
    attr_accessor :max_threads
    attr_accessor :persistent_timeout
    attr_accessor :auto_trim_time
    attr_accessor :reaping_time
    attr_accessor :first_data_timeout

    # Create a server for the rack app +app+.
    #
    # +events+ is an object which will be called when certain error events occur
    # to be handled. See Puma::Events for the list of current methods to implement.
    #
    # Server#run returns a thread that you can join on to wait for the server
    # to do its work.
    #
    def initialize(app, events=Events.stdio, options={})
      @app = app
      @events = events

      @check, @notify = Puma::Util.pipe

      @status = :stop

      @min_threads = 0
      @max_threads = 16
      @auto_trim_time = 30
      @reaping_time = 1

      @thread = nil
      @thread_pool = nil
      @early_hints = nil

      @persistent_timeout = options.fetch(:persistent_timeout, PERSISTENT_TIMEOUT)
      @first_data_timeout = options.fetch(:first_data_timeout, FIRST_DATA_TIMEOUT)

      @binder = Binder.new(events)
      @own_binder = true

      @leak_stack_on_error = true

      @options = options
      @queue_requests = options[:queue_requests].nil? ? true : options[:queue_requests]

      ENV['RACK_ENV'] ||= "development"

      @mode = :http

      @precheck_closing = true
    end

    attr_accessor :binder, :leak_stack_on_error, :early_hints

    forward :add_tcp_listener,  :@binder
    forward :add_ssl_listener,  :@binder
    forward :add_unix_listener, :@binder
    forward :connected_port,    :@binder

    def inherit_binder(bind)
      @binder = bind
      @own_binder = false
    end

    def tcp_mode!
      @mode = :tcp
    end

    # On Linux, use TCP_CORK to better control how the TCP stack
    # packetizes our stream. This improves both latency and throughput.
    #
    if RUBY_PLATFORM =~ /linux/
      UNPACK_TCP_STATE_FROM_TCP_INFO = "C".freeze

      # 6 == Socket::IPPROTO_TCP
      # 3 == TCP_CORK
      # 1/0 == turn on/off
      def cork_socket(socket)
        begin
          socket.setsockopt(6, 3, 1) if socket.kind_of? TCPSocket
        rescue IOError, SystemCallError
          Thread.current.purge_interrupt_queue if Thread.current.respond_to? :purge_interrupt_queue
        end
      end

      def uncork_socket(socket)
        begin
          socket.setsockopt(6, 3, 0) if socket.kind_of? TCPSocket
        rescue IOError, SystemCallError
          Thread.current.purge_interrupt_queue if Thread.current.respond_to? :purge_interrupt_queue
        end
      end

      def closed_socket?(socket)
        return false unless socket.kind_of? TCPSocket
        return false unless @precheck_closing

        begin
          tcp_info = socket.getsockopt(Socket::SOL_TCP, Socket::TCP_INFO)
        rescue IOError, SystemCallError
          Thread.current.purge_interrupt_queue if Thread.current.respond_to? :purge_interrupt_queue
          @precheck_closing = false
          false
        else
          state = tcp_info.unpack(UNPACK_TCP_STATE_FROM_TCP_INFO)[0]
          # TIME_WAIT: 6, CLOSE: 7, CLOSE_WAIT: 8, LAST_ACK: 9, CLOSING: 11
          (state >= 6 && state <= 9) || state == 11
        end
      end
    else
      def cork_socket(socket)
      end

      def uncork_socket(socket)
      end

      def closed_socket?(socket)
        false
      end
    end

    def backlog
      @thread_pool and @thread_pool.backlog
    end

    def running
      @thread_pool and @thread_pool.spawned
    end

    # Lopez Mode == raw tcp apps

    def run_lopez_mode(background=true)
      @thread_pool = ThreadPool.new(@min_threads,
                                    @max_threads,
                                    Hash) do |client, tl|

        io = client.to_io
        addr = io.peeraddr.last

        if addr.empty?
          # Set unix socket addrs to localhost
          addr = "127.0.0.1:0"
        else
          addr = "#{addr}:#{io.peeraddr[1]}"
        end

        env = { 'thread' => tl, REMOTE_ADDR => addr }

        begin
          @app.call env, client.to_io
        rescue Object => e
          STDERR.puts "! Detected exception at toplevel: #{e.message} (#{e.class})"
          STDERR.puts e.backtrace
        end

        client.close unless env['detach']
      end

      @events.fire :state, :running

      if background
        @thread = Thread.new { handle_servers_lopez_mode }
        return @thread
      else
        handle_servers_lopez_mode
      end
    end

    def handle_servers_lopez_mode
      begin
        check = @check
        sockets = [check] + @binder.ios
        pool = @thread_pool

        while @status == :run
          begin
            ios = IO.select sockets
            ios.first.each do |sock|
              if sock == check
                break if handle_check
              else
                begin
                  if io = sock.accept_nonblock
                    client = Client.new io, nil
                    pool << client
                  end
                rescue SystemCallError
                  # nothing
                rescue Errno::ECONNABORTED
                  # client closed the socket even before accept
                  begin
                    io.close
                  rescue
                    Thread.current.purge_interrupt_queue if Thread.current.respond_to? :purge_interrupt_queue
                  end
                end
              end
            end
          rescue Object => e
            @events.unknown_error self, e, "Listen loop"
          end
        end

        @events.fire :state, @status

        graceful_shutdown if @status == :stop || @status == :restart

      rescue Exception => e
        STDERR.puts "Exception handling servers: #{e.message} (#{e.class})"
        STDERR.puts e.backtrace
      ensure
        begin
          @check.close
        rescue
          Thread.current.purge_interrupt_queue if Thread.current.respond_to? :purge_interrupt_queue
        end

        # Prevent can't modify frozen IOError (RuntimeError)
        @notify.close rescue nil

        if @status != :restart and @own_binder
          @binder.close
        end
      end

      @events.fire :state, :done
    end
    # Runs the server.
    #
    # If +background+ is true (the default) then a thread is spun
    # up in the background to handle requests. Otherwise requests
    # are handled synchronously.
    #
    def run(background=true)
      BasicSocket.do_not_reverse_lookup = true

      @events.fire :state, :booting

      @status = :run

      if @mode == :tcp
        return run_lopez_mode(background)
      end

      queue_requests = @queue_requests

      @thread_pool = ThreadPool.new(@min_threads,
                                    @max_threads,
                                    IOBuffer) do |client, buffer|

        # Advertise this server into the thread
        Thread.current[ThreadLocalKey] = self

        process_now = false

        begin
          if queue_requests
            process_now = client.eagerly_finish
          else
            client.finish
            process_now = true
          end
        rescue MiniSSL::SSLError => e
          ssl_socket = client.io
          addr = ssl_socket.peeraddr.last
          cert = ssl_socket.peercert

          client.close

          @events.ssl_error self, addr, cert, e
        rescue HttpParserError => e
          client.write_400
          client.close

          @events.parse_error self, client.env, e
        rescue ConnectionError, EOFError
          client.close
        else
          if process_now
            process_client client, buffer
          else
            client.set_timeout @first_data_timeout
            @reactor.add client
          end
        end
      end

      @thread_pool.clean_thread_locals = @options[:clean_thread_locals]

      if queue_requests
        @reactor = Reactor.new self, @thread_pool
        @reactor.run_in_thread
      end

      if @reaping_time
        @thread_pool.auto_reap!(@reaping_time)
      end

      if @auto_trim_time
        @thread_pool.auto_trim!(@auto_trim_time)
      end

      @events.fire :state, :running

      if background
        @thread = Thread.new { handle_servers }
        return @thread
      else
        handle_servers
      end
    end

    def handle_servers
      begin
        check = @check
        sockets = [check] + @binder.ios
        pool = @thread_pool
        queue_requests = @queue_requests

        remote_addr_value = nil
        remote_addr_header = nil

        case @options[:remote_address]
        when :value
          remote_addr_value = @options[:remote_address_value]
        when :header
          remote_addr_header = @options[:remote_address_header]
        end

        while @status == :run
          begin
            ios = IO.select sockets
            ios.first.each do |sock|
              if sock == check
                break if handle_check
              else
                begin
                  if io = sock.accept_nonblock
                    client = Client.new io, @binder.env(sock)
                    if remote_addr_value
                      client.peerip = remote_addr_value
                    elsif remote_addr_header
                      client.remote_addr_header = remote_addr_header
                    end

                    pool << client
                    pool.wait_until_not_full
                  end
                rescue SystemCallError
                  # nothing
                rescue Errno::ECONNABORTED
                  # client closed the socket even before accept
                  begin
                    io.close
                  rescue
                    Thread.current.purge_interrupt_queue if Thread.current.respond_to? :purge_interrupt_queue
                  end
                end
              end
            end
          rescue Object => e
            @events.unknown_error self, e, "Listen loop"
          end
        end

        @events.fire :state, @status

        graceful_shutdown if @status == :stop || @status == :restart
        if queue_requests
          @reactor.clear!
          @reactor.shutdown
        end
      rescue Exception => e
        STDERR.puts "Exception handling servers: #{e.message} (#{e.class})"
        STDERR.puts e.backtrace
      ensure
        @check.close
        @notify.close

        if @status != :restart and @own_binder
          @binder.close
        end
      end

      @events.fire :state, :done
    end

    # :nodoc:
    def handle_check
      cmd = @check.read(1)

      case cmd
      when STOP_COMMAND
        @status = :stop
        return true
      when HALT_COMMAND
        @status = :halt
        return true
      when RESTART_COMMAND
        @status = :restart
        return true
      end

      return false
    end

    # Given a connection on +client+, handle the incoming requests.
    #
    # This method support HTTP Keep-Alive so it may, depending on if the client
    # indicates that it supports keep alive, wait for another request before
    # returning.
    #
    def process_client(client, buffer)
      begin

        clean_thread_locals = @options[:clean_thread_locals]
        close_socket = true

        while true
          case handle_request(client, buffer)
          when false
            return
          when :async
            close_socket = false
            return
          when true
            return unless @queue_requests
            buffer.reset

            ThreadPool.clean_thread_locals if clean_thread_locals

            unless client.reset(@status == :run)
              close_socket = false
              client.set_timeout @persistent_timeout
              @reactor.add client
              return
            end
          end
        end

      # The client disconnected while we were reading data
      rescue ConnectionError
        # Swallow them. The ensure tries to close +client+ down

      # SSL handshake error
      rescue MiniSSL::SSLError => e
        lowlevel_error(e, client.env)

        ssl_socket = client.io
        addr = ssl_socket.peeraddr.last
        cert = ssl_socket.peercert

        close_socket = true

        @events.ssl_error self, addr, cert, e

      # The client doesn't know HTTP well
      rescue HttpParserError => e
        lowlevel_error(e, client.env)

        client.write_400

        @events.parse_error self, client.env, e

      # Server error
      rescue StandardError => e
        lowlevel_error(e, client.env)

        client.write_500

        @events.unknown_error self, e, "Read"

      ensure
        buffer.reset

        begin
          client.close if close_socket
        rescue IOError, SystemCallError
          Thread.current.purge_interrupt_queue if Thread.current.respond_to? :purge_interrupt_queue
          # Already closed
        rescue StandardError => e
          @events.unknown_error self, e, "Client"
        end
      end
    end

    # Given a Hash +env+ for the request read from +client+, add
    # and fixup keys to comply with Rack's env guidelines.
    #
    def normalize_env(env, client)
      if host = env[HTTP_HOST]
        if colon = host.index(":")
          env[SERVER_NAME] = host[0, colon]
          env[SERVER_PORT] = host[colon+1, host.bytesize]
        else
          env[SERVER_NAME] = host
          env[SERVER_PORT] = default_server_port(env)
        end
      else
        env[SERVER_NAME] = LOCALHOST
        env[SERVER_PORT] = default_server_port(env)
      end

      unless env[REQUEST_PATH]
        # it might be a dumbass full host request header
        uri = URI.parse(env[REQUEST_URI])
        env[REQUEST_PATH] = uri.path

        raise "No REQUEST PATH" unless env[REQUEST_PATH]

        # A nil env value will cause a LintError (and fatal errors elsewhere),
        # so only set the env value if there actually is a value.
        env[QUERY_STRING] = uri.query if uri.query
      end

      env[PATH_INFO] = env[REQUEST_PATH]

      # From http://www.ietf.org/rfc/rfc3875 :
      # "Script authors should be aware that the REMOTE_ADDR and
      # REMOTE_HOST meta-variables (see sections 4.1.8 and 4.1.9)
      # may not identify the ultimate source of the request.
      # They identify the client for the immediate request to the
      # server; that client may be a proxy, gateway, or other
      # intermediary acting on behalf of the actual source client."
      #

      unless env.key?(REMOTE_ADDR)
        begin
          addr = client.peerip
        rescue Errno::ENOTCONN
          # Client disconnects can result in an inability to get the
          # peeraddr from the socket; default to localhost.
          addr = LOCALHOST_IP
        end

        # Set unix socket addrs to localhost
        addr = LOCALHOST_IP if addr.empty?

        env[REMOTE_ADDR] = addr
      end
    end

    def default_server_port(env)
      return PORT_443 if env[HTTPS_KEY] == 'on' || env[HTTPS_KEY] == 'https'
      env['HTTP_X_FORWARDED_PROTO'] == 'https' ? PORT_443 : PORT_80
    end

    # Given the request +env+ from +client+ and a partial request body
    # in +body+, finish reading the body if there is one and invoke
    # the rack app. Then construct the response and write it back to
    # +client+
    #
    # +cl+ is the previously fetched Content-Length header if there
    # was one. This is an optimization to keep from having to look
    # it up again.
    #
    def handle_request(req, lines)
      env = req.env
      client = req.io

      return false if closed_socket?(client)

      normalize_env env, req

      env[PUMA_SOCKET] = client

      if env[HTTPS_KEY] && client.peercert
        env[PUMA_PEERCERT] = client.peercert
      end

      env[HIJACK_P] = true
      env[HIJACK] = req

      body = req.body

      head = env[REQUEST_METHOD] == HEAD

      env[RACK_INPUT] = body
      env[RACK_URL_SCHEME] =  env[HTTPS_KEY] ? HTTPS : HTTP

      if @early_hints
        env[EARLY_HINTS] = lambda { |headers|
          fast_write client, "HTTP/1.1 103 Early Hints\r\n".freeze

          headers.each_pair do |k, vs|
            if vs.respond_to?(:to_s) && !vs.to_s.empty?
              vs.to_s.split(NEWLINE).each do |v|
                fast_write client, "#{k}: #{v}\r\n"
              end
            else
              fast_write client, "#{k}: #{vs}\r\n"
            end
          end

          fast_write client, "\r\n".freeze
        }
      end

      # A rack extension. If the app writes #call'ables to this
      # array, we will invoke them when the request is done.
      #
      after_reply = env[RACK_AFTER_REPLY] = []

      begin
        begin
          status, headers, res_body = @app.call(env)

          return :async if req.hijacked

          status = status.to_i

          if status == -1
            unless headers.empty? and res_body == []
              raise "async response must have empty headers and body"
            end

            return :async
          end
        rescue ThreadPool::ForceShutdown => e
          @events.log "Detected force shutdown of a thread, returning 503"
          @events.unknown_error self, e, "Rack app"

          status = 503
          headers = {}
          res_body = ["Request was internally terminated early\n"]

        rescue Exception => e
          @events.unknown_error self, e, "Rack app", env

          status, headers, res_body = lowlevel_error(e, env)
        end

        content_length = nil
        no_body = head

        if res_body.kind_of? Array and res_body.size == 1
          content_length = res_body[0].bytesize
        end

        cork_socket client

        line_ending = LINE_END
        colon = COLON

        http_11 = if env[HTTP_VERSION] == HTTP_11
          allow_chunked = true
          keep_alive = env.fetch(HTTP_CONNECTION, "").downcase != CLOSE
          include_keepalive_header = false

          # An optimization. The most common response is 200, so we can
          # reply with the proper 200 status without having to compute
          # the response header.
          #
          if status == 200
            lines << HTTP_11_200
          else
            lines.append "HTTP/1.1 ", status.to_s, " ",
                         fetch_status_code(status), line_ending

            no_body ||= status < 200 || STATUS_WITH_NO_ENTITY_BODY[status]
          end
          true
        else
          allow_chunked = false
          keep_alive = env.fetch(HTTP_CONNECTION, "").downcase == KEEP_ALIVE
          include_keepalive_header = keep_alive

          # Same optimization as above for HTTP/1.1
          #
          if status == 200
            lines << HTTP_10_200
          else
            lines.append "HTTP/1.0 ", status.to_s, " ",
                         fetch_status_code(status), line_ending

            no_body ||= status < 200 || STATUS_WITH_NO_ENTITY_BODY[status]
          end
          false
        end

        response_hijack = nil

        headers.each do |k, vs|
          case k.downcase
          when CONTENT_LENGTH2
            content_length = vs
            next
          when TRANSFER_ENCODING
            allow_chunked = false
            content_length = nil
          when HIJACK
            response_hijack = vs
            next
          end

          if vs.respond_to?(:to_s) && !vs.to_s.empty?
            vs.to_s.split(NEWLINE).each do |v|
              lines.append k, colon, v, line_ending
            end
          else
            lines.append k, colon, line_ending
          end
        end

        if include_keepalive_header
          lines << CONNECTION_KEEP_ALIVE
        elsif http_11 && !keep_alive
          lines << CONNECTION_CLOSE
        end

        if no_body
          if content_length and status != 204
            lines.append CONTENT_LENGTH_S, content_length.to_s, line_ending
          end

          lines << line_ending
          fast_write client, lines.to_s
          return keep_alive
        end

        if content_length
          lines.append CONTENT_LENGTH_S, content_length.to_s, line_ending
          chunked = false
        elsif !response_hijack and allow_chunked
          lines << TRANSFER_ENCODING_CHUNKED
          chunked = true
        end

        lines << line_ending

        fast_write client, lines.to_s

        if response_hijack
          response_hijack.call client
          return :async
        end

        begin
          res_body.each do |part|
            next if part.bytesize.zero?
            if chunked
              fast_write client, part.bytesize.to_s(16)
              fast_write client, line_ending
              fast_write client, part
              fast_write client, line_ending
            else
              fast_write client, part
            end

            client.flush
          end

          if chunked
            fast_write client, CLOSE_CHUNKED
            client.flush
          end
        rescue SystemCallError, IOError
          raise ConnectionError, "Connection error detected during write"
        end

      ensure
        uncork_socket client

        body.close
        req.tempfile.unlink if req.tempfile
        res_body.close if res_body.respond_to? :close

        after_reply.each { |o| o.call }
      end

      return keep_alive
    end

    def fetch_status_code(status)
      HTTP_STATUS_CODES.fetch(status) { 'CUSTOM' }
    end
    private :fetch_status_code

    # Given the request +env+ from +client+ and the partial body +body+
    # plus a potential Content-Length value +cl+, finish reading
    # the body and return it.
    #
    # If the body is larger than MAX_BODY, a Tempfile object is used
    # for the body, otherwise a StringIO is used.
    #
    def read_body(env, client, body, cl)
      content_length = cl.to_i

      remain = content_length - body.bytesize

      return StringIO.new(body) if remain <= 0

      # Use a Tempfile if there is a lot of data left
      if remain > MAX_BODY
        stream = Tempfile.new(Const::PUMA_TMP_BASE)
        stream.binmode
      else
        # The body[0,0] trick is to get an empty string in the same
        # encoding as body.
        stream = StringIO.new body[0,0]
      end

      stream.write body

      # Read an odd sized chunk so we can read even sized ones
      # after this
      chunk = client.readpartial(remain % CHUNK_SIZE)

      # No chunk means a closed socket
      unless chunk
        stream.close
        return nil
      end

      remain -= stream.write(chunk)

      # Raed the rest of the chunks
      while remain > 0
        chunk = client.readpartial(CHUNK_SIZE)
        unless chunk
          stream.close
          return nil
        end

        remain -= stream.write(chunk)
      end

      stream.rewind

      return stream
    end

    # A fallback rack response if +@app+ raises as exception.
    #
    def lowlevel_error(e, env)
      if handler = @options[:lowlevel_error_handler]
        if handler.arity == 1
          return handler.call(e)
        else
          return handler.call(e, env)
        end
      end

      if @leak_stack_on_error
        [500, {}, ["Puma caught this error: #{e.message} (#{e.class})\n#{e.backtrace.join("\n")}"]]
      else
        [500, {}, ["An unhandled lowlevel error occurred. The application logs may have details.\n"]]
      end
    end

    # Wait for all outstanding requests to finish.
    #
    def graceful_shutdown
      if @options[:shutdown_debug]
        threads = Thread.list
        total = threads.size

        pid = Process.pid

        $stdout.syswrite "#{pid}: === Begin thread backtrace dump ===\n"

        threads.each_with_index do |t,i|
          $stdout.syswrite "#{pid}: Thread #{i+1}/#{total}: #{t.inspect}\n"
          $stdout.syswrite "#{pid}: #{t.backtrace.join("\n#{pid}: ")}\n\n"
        end
        $stdout.syswrite "#{pid}: === End thread backtrace dump ===\n"
      end

      if @options[:drain_on_shutdown]
        count = 0

        while true
          ios = IO.select @binder.ios, nil, nil, 0
          break unless ios

          ios.first.each do |sock|
            begin
              if io = sock.accept_nonblock
                count += 1
                client = Client.new io, @binder.env(sock)
                @thread_pool << client
              end
            rescue SystemCallError
            end
          end
        end

        @events.debug "Drained #{count} additional connections."
      end

      if @thread_pool
        if timeout = @options[:force_shutdown_after]
          @thread_pool.shutdown timeout.to_i
        else
          @thread_pool.shutdown
        end
      end
    end

    def notify_safely(message)
      begin
        @notify << message
      rescue IOError
         # The server, in another thread, is shutting down
        Thread.current.purge_interrupt_queue if Thread.current.respond_to? :purge_interrupt_queue
      rescue RuntimeError => e
        # Temporary workaround for https://bugs.ruby-lang.org/issues/13239
        if e.message.include?('IOError')
          Thread.current.purge_interrupt_queue if Thread.current.respond_to? :purge_interrupt_queue
        else
          raise e
        end
      end
    end
    private :notify_safely

    # Stops the acceptor thread and then causes the worker threads to finish
    # off the request queue before finally exiting.

    def stop(sync=false)
      notify_safely(STOP_COMMAND)
      @thread.join if @thread && sync
    end

    def halt(sync=false)
      notify_safely(HALT_COMMAND)
      @thread.join if @thread && sync
    end

    def begin_restart
      notify_safely(RESTART_COMMAND)
    end

    def fast_write(io, str)
      n = 0
      while true
        begin
          n = io.syswrite str
        rescue Errno::EAGAIN, Errno::EWOULDBLOCK
          if !IO.select(nil, [io], nil, WRITE_TIMEOUT)
            raise ConnectionError, "Socket timeout writing data"
          end

          retry
        rescue  Errno::EPIPE, SystemCallError, IOError
          raise ConnectionError, "Socket timeout writing data"
        end

        return if n == str.bytesize
        str = str.byteslice(n..-1)
      end
    end
    private :fast_write

    ThreadLocalKey = :puma_server

    def self.current
      Thread.current[ThreadLocalKey]
    end

    def shutting_down?
      @status == :stop || @status == :restart
    end
  end
end