123456789_123456789_123456789_123456789_123456789_

Class: Puma::Reactor

Relationships & Source Files
Inherits: Object
Defined in: lib/puma/reactor.rb

Overview

Internal Docs, Not a public interface.

The Reactor object is responsible for ensuring that a request has been completely received before it starts to be processed. This may be known as read buffering. If read buffering is not done, and no other read buffering is performed (such as by an application server such as nginx) then the application would be subject to a slow client attack.

Each Puma “worker” process has its own Reactor. For example if you start puma with $ puma -w 5 then it will have 5 workers and each worker will have it’s own reactor.

For a graphical representation of how the reactor works see architecture.md.

Reactor Flow

A connection comes into a Server instance, it is then passed to a Reactor instance, which stores it in an array and waits for any of the connections to be ready for reading.

The waiting/wake up is performed with nio4r, which will use the appropriate backend (libev, Java NIO or just plain IO#select). The call to NIO::Selector#select will “wake up” and return the references to any objects that caused it to “wake”. The reactor then loops through each of these request objects, and sees if they’re complete. If they have a full header and body then the reactor passes the request to a thread pool. Once in a thread pool, a “worker thread” can run the the application’s Ruby code against the request.

If the request is not complete, then it stays in the array, and the next time any data is written to that socket reference, then the loop is woken up and it is checked for completeness again.

A detailed example is given in the docs for #run_internal which is where the bulk of this logic lives.

Constant Summary

Class Method Summary

Instance Method Summary

Constructor Details

.new(server, app_pool) ⇒ Reactor

Creates an instance of Reactor

The server argument is an instance of Server that is used to write a response for “low level errors” when there is an exception inside of the reactor.

The app_pool is an instance of ThreadPool. Once a request is fully formed (header and body are received) it will be passed to the app_pool.

[ GitHub ] 

  


# File 'lib/puma/reactor.rb', line 50



def initialize(server, app_pool)
  @server = server
  @events = server.events
  @app_pool = app_pool

  @selector = NIO::Selector.new

  @mutex = Mutex.new

  # Read / Write pipes to wake up internal while loop
  @ready, @trigger = Puma::Util.pipe
  @input = []
  @sleep_for = DefaultSleepFor
  @timeouts = []

  mon = @selector.register(@ready, :r)
  mon.value = @ready

  @monitors = [mon]
end


  







Instance Method Details



  

    #add(c)  
  



  

    

This method adds a connection to the reactor



Typically called by Server the value passed in is usually a Client object that responds like an ::IO object.



The main body of the reactor loop is in #run_internal and it will sleep on NIO::Selector#select. When a new connection is added to the reactor it cannot be added directly to the sockets array, because the NIO::Selector#select will not be watching for it yet.



Instead what needs to happen is that NIO::Selector#select needs to be woken up, the contents of @input added to the sockets array, and then another call to NIO::Selector#select needs to happen. Since the Client object can be read immediately, it does not block, but instead returns right away.



This behavior is accomplished by writing to @trigger which wakes up the NIO::Selector#select and then there is logic to detect the value of *, pull the contents from @input and add them to the sockets array.



If the object passed in has a timeout value in timeout_at then it is added to a @timeouts array. This array is then re-arranged so that the first element to timeout will be at the front of the array. Then a value to sleep for is derived in the call to #calculate_sleep


  



  [ GitHub ]


  

  


# File 'lib/puma/reactor.rb', line 374



def add(c)
  @mutex.synchronize do
    @input << c
    @trigger << "*"
  end
end


  








  

    #calculate_sleep  
  



  

    

The calculate_sleep sets the value that the NIO::Selector#select will sleep for in the main reactor loop when no sockets are being written to.



The values kept in @timeouts are sorted so that the first timeout comes first in the array. When there are no timeouts the default timeout is used.



Otherwise a sleep value is set that is the same as the amount of time it would take for the first element to time out.



If that value is in the past, then a sleep value of zero is used.


  



  [ GitHub ]


  

  


# File 'lib/puma/reactor.rb', line 335



def calculate_sleep
  if @timeouts.empty?
    @sleep_for = DefaultSleepFor
  else
    diff = @timeouts.first.value.timeout_at.to_f - Time.now.to_f

    if diff < 0.0
      @sleep_for = 0
    else
      @sleep_for = diff
    end
  end
end


  








  

    #clear!  
  



  

    

Close all watched sockets and clear them from being watched


  



  [ GitHub ]


  

  


# File 'lib/puma/reactor.rb', line 382



def clear!
  begin
    @trigger << "c"
  rescue IOError
    Thread.current.purge_interrupt_queue if Thread.current.respond_to? :purge_interrupt_queue
  end
end


  








  

    #clear_monitor(mon)   (private)
  

  [ GitHub ]


  

  


# File 'lib/puma/reactor.rb', line 295



def clear_monitor(mon)
  @selector.deregister mon.value
  @monitors.delete mon
end


  








  

    #run  
  

  [ GitHub ]


  

  


# File 'lib/puma/reactor.rb', line 302



def run
  run_internal
ensure
  @trigger.close
  @ready.close
end


  








  

    #run_in_thread  
  

  [ GitHub ]


  

  


# File 'lib/puma/reactor.rb', line 309



def run_in_thread
  @thread = Thread.new do
    Puma.set_thread_name "reactor"
    begin
      run_internal
    rescue StandardError => e
      STDERR.puts "Error in reactor loop escaped: #{e.message} (#{e.class})"
      STDERR.puts e.backtrace
      retry
    ensure
      @trigger.close
      @ready.close
    end
  end
end


  








  

    #run_internal   (private)
  



  

    

Until a request is added via the #add method this method will internally loop, waiting on the sockets array objects. The only object in this array at first is the @ready ::IO object, which is the read end of a pipe connected to @trigger object. When @trigger is written to, then the loop will break on NIO::Selector#select and return an array.



When a request is added:



When the #add method is called, an instance of Client is added to the @input array. Next the @ready pipe is “woken” by writing a string of "*" to @trigger.



When that happens, the internal loop stops blocking at NIO::Selector#select and returns a reference to whatever “woke” it up. On the very first loop, the only thing in sockets is @ready. When @trigger is written-to, the loop “wakes” and the ready variable returns an array of arrays that looks like [[#<IO:fd 10>], [], []] where the first ::IO object is the @ready object. This first array [#<IO:fd 10>] is saved as a reads variable.



The reads variable is iterated through. In the case that the object is the same as the @ready input pipe, then we know that there was a trigger event.



If there was a trigger event, then one byte of @ready is read into memory. In the case of the first request, the reactor sees that it’s a "*" value and the reactor adds the contents of @input into the sockets array. The while then loop continues to iterate again, but now the sockets array contains a Client instance in addition to the @ready ::IO object. For example: [#<IO:fd 10>, #<Puma::Client:0x3fdc1103bee8 @ready=false>].



Since the Client in this example has data that has not been read yet, the NIO::Selector#select is immediately able to “wake” and read from the Client. At this point the ready output looks like this: [[#<Puma::Client:0x3fdc1103bee8 @ready=false>], [], []].



Each element in the first entry is iterated over. The Client object is not the @ready pipe, so the reactor checks to see if it has the full header and body with the Client#try_to_finish method. If the full request has been sent, then the request is passed off to the @app_pool thread pool so that a “worker thread” can pick up the request and begin to execute application logic. This is done via @app_pool << c. The Client is then removed from the sockets array.



If the request body is not present then nothing will happen, and the loop will iterate again. When the client sends more data to the socket the Client object will wake up the NIO::Selector#select and it can again be checked to see if it’s ready to be passed to the thread pool.



Time Out Case



In addition to being woken via a write to one of the sockets the NIO::Selector#select will periodically “time out” of the sleep. One of the functions of this is to check for any requests that have “timed out”. At the end of the loop it’s checked to see if the first element in the @timeout array has exceed its allowed time. If so, the client object is removed from the timeout array, a 408 response is written. Then its connection is closed, and the object is removed from the sockets array that watches for new data.



This behavior loops until all the objects that have timed out have been removed.



Once all the timeouts have been processed, the next duration of the NIO::Selector#select sleep will be set to be equal to the amount of time it will take for the next timeout to occur. This calculation happens in #calculate_sleep.


  



  [ GitHub ]


  

  


# File 'lib/puma/reactor.rb', line 130



def run_internal
  monitors = @monitors
  selector = @selector

  while true
    begin
      ready = selector.select @sleep_for
    rescue IOError => e
      Thread.current.purge_interrupt_queue if Thread.current.respond_to? :purge_interrupt_queue
      if monitors.any? { |mon| mon.value.closed? }
        STDERR.puts "Error in select: #{e.message} (#{e.class})"
        STDERR.puts e.backtrace

        monitors.reject! do |mon|
          if mon.value.closed?
            selector.deregister mon.value
            true
          end
        end

        retry
      else
        raise
      end
    end

    if ready
      ready.each do |mon|
        if mon.value == @ready
          @mutex.synchronize do
            case @ready.read(1)
            when "*"
              @input.each do |c|
                mon = nil
                begin
                  begin
                    mon = selector.register(c, :r)
                  rescue ArgumentError
                    # There is a bug where we seem to be registering an already registered
                    # client. This code deals with this situation but I wish we didn't have to.
                    monitors.delete_if { |submon| submon.value.to_io == c.to_io }
                    selector.deregister(c)
                    mon = selector.register(c, :r)
                  end
                rescue IOError
                  # Means that the io is closed, so we should ignore this request
                  # entirely
                else
                  mon.value = c
                  @timeouts << mon if c.timeout_at
                  monitors << mon
                end
              end
              @input.clear

              @timeouts.sort! { |a,b| a.value.timeout_at <=> b.value.timeout_at }
              calculate_sleep
            when "c"
              monitors.reject! do |submon|
                if submon.value == @ready
                  false
                else
                  submon.value.close
                  begin
                    selector.deregister submon.value
                  rescue IOError
                    # nio4r on jruby seems to throw an IOError here if the IO is closed, so
                    # we need to swallow it.
                  end
                  true
                end
              end
            when "!"
              return
            end
          end
        else
          c = mon.value

          # 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 mon
            end
          end

          begin
            if c.try_to_finish
              @app_pool << c
              clear_monitor mon
            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_error(500)
            c.close

            clear_monitor mon

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

            ssl_socket = c.io
            begin
              addr = ssl_socket.peeraddr.last
            # EINVAL can happen when browser closes socket w/security exception
            rescue IOError, Errno::EINVAL
              addr = "<unknown>"
            end

            cert = ssl_socket.peercert

            c.close
            clear_monitor mon

            @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_error(400)
            c.close

            clear_monitor mon

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

            c.write_error(500)
            c.close

            clear_monitor mon
          end
        end
      end
    end

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

        while @timeouts.first.value.timeout_at < now
          mon = @timeouts.shift
          c = mon.value
          c.write_error(408) if c.in_data_phase
          c.close

          clear_monitor mon

          break if @timeouts.empty?
        end

        calculate_sleep
      end
    end
  end
end


  








  

    #shutdown  
  

  [ GitHub ]


  

  


# File 'lib/puma/reactor.rb', line 390



def shutdown
  begin
    @trigger << "!"
  rescue IOError
    Thread.current.purge_interrupt_queue if Thread.current.respond_to? :purge_interrupt_queue
  end

  @thread.join
end