Ruby Puma 原理解析

sockets = @sockets

while true
  begin
    ready = IO.select sockets, nil, nil, @sleep_for
  rescue IOError => e
    Thread.current.purge_interrupt_queue if Thread.current.respond_to? :purge_interrupt_queue
    if sockets.any? { |socket| socket.closed? }
      STDERR.puts "Error in select: #{e.message} (#{e.class})"
      STDERR.puts e.backtrace
      sockets = sockets.reject { |socket| socket.closed? }
      retry
    else
      raise
    end
  end

  if ready and reads = ready[0]
    reads.each do |c|
      if c == @ready
        @mutex.synchronize do
          case @ready.read(1)
          when "*"
            sockets += @input
            @input.clear
          when "c"
            sockets.delete_if do |s|
              if s == @ready
                false
              else
                s.close
                true
              end
            end
          when "!"
            return
          end
        end

      else
        # We have to be sure to remove it from the timeout
        # list or we'll accidentally close the socket when
        # it's in use!
        if c.timeout_at
          @mutex.synchronize do
            @timeouts.delete c
          end
        end

        begin
          if c.try_to_finish
            @app_pool << c
            sockets.delete c
          end

        # Don't report these to the lowlevel_error handler, otherwise
        # will be flooding them with errors when persistent connections
        # are closed.
        rescue ConnectionError
          c.write_500
          c.close

          sockets.delete c

        # SSL handshake failure
        rescue MiniSSL::SSLError => e
          @server.lowlevel_error(e, c.env)

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

          c.close
          sockets.delete c

          @events.ssl_error @server, addr, cert, e

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

          c.write_400
          c.close

          sockets.delete c

          @events.parse_error @server, c.env, e
        rescue StandardError => e
          @server.lowlevel_error(e, c.env)

          c.write_500
          c.close

          sockets.delete c
        end
      end
    end
  end

  unless @timeouts.empty?
    @mutex.synchronize do
      now = Time.now

      while @timeouts.first.timeout_at < now
        c = @timeouts.shift
        c.write_408 if c.in_data_phase
        c.close
        sockets.delete c

        break if @timeouts.empty?
      end

      calculate_sleep
    end
  end
end

read, @wakeup = Puma::Util.pipe (1)
@check_pipe, @suicide_pipe = Puma::Util.pipe (2)
@master_read, @worker_write = read, @wakeup (3)

def app
  @app ||= @launcher.config.app
end

def spawn_thread
  @spawned += 1

  th = Thread.new(@spawned) do |spawned|
    # Thread name is new in Ruby 2.3
    Thread.current.name = 'puma %03i' % spawned if Thread.current.respond_to?(:name=)
    todo  = @todo
    block = @block
    mutex = @mutex
    not_empty = @not_empty
    not_full = @not_full

    extra = @extra.map { |i| i.new }

    while true
      work = nil

      continue = true

      mutex.synchronize do
        while todo.empty?
          if @trim_requested > 0
            @trim_requested -= 1
            continue = false
            not_full.signal
            break
          end

          if @shutdown
            continue = false
            break
          end

          @waiting += 1
          not_full.signal
          not_empty.wait mutex
          @waiting -= 1
        end

        work = todo.shift if continue
      end

      break unless continue

      if @clean_thread_locals
        ThreadPool.clean_thread_locals
      end

      begin
        block.call(work, *extra)
      rescue Exception => e
        STDERR.puts "Error reached top of thread-pool: #{e.message} (#{e.class})"
      end
    end

    mutex.synchronize do
      @spawned -= 1
      @workers.delete th
    end
  end

  @workers << th

  th
end

while todo.empty?
  if @trim_requested > 0
    @trim_requested -= 1
    continue = false
    not_full.signal
    break
  end

  if @shutdown
    continue = false
    break
  end

  @waiting += 1
  not_full.signal
  not_empty.wait mutex
  @waiting -= 1
end

  if queue_requests
    process_now = client.eagerly_finish
  else
    client.finish
    process_now = true
  end

  def eagerly_finish
    return true if @ready
    return false unless IO.select([@to_io], nil, nil, 0)
    try_to_finish
  end

  def finish
    return true if @ready
    until try_to_finish
      IO.select([@to_io], nil, nil)
    end
    true
  end

def try_to_finish
  return read_body unless @read_header

  begin
    data = @io.read_nonblock(CHUNK_SIZE)
  rescue Errno::EAGAIN
    return false
  rescue SystemCallError, IOError
    raise ConnectionError, "Connection error detected during read"
  end

  # No data means a closed socket
  unless data
    @buffer = nil
    @requests_served += 1
    @ready = true
    raise EOFError
  end

  if @buffer
    @buffer << data
  else
    @buffer = data
  end

  @parsed_bytes = @parser.execute(@env, @buffer, @parsed_bytes)

  if @parser.finished?
    return setup_body
  elsif @parsed_bytes >= MAX_HEADER
    raise HttpParserError,
      "HEADER is longer than allowed, aborting client early."
  end

  false
end