class Puma::Reactor

of this logic lives.
A detailed example is given in the docs for ‘run_internal` which is where the bulk
data is written to that socket reference, then the loop is woken up and it is checked for completeness again.
If the request is not complete, then it stays in the array, and the next time any
Once in a thread pool, a “worker thread” can run the the application’s Ruby code against the request.
have a full header and body then the reactor passes the request to a thread pool.
then loops through each of these request objects, and sees if they’re complete. If they
return the references to any objects that caused it to “wake”. The reactor
just plain IO#select). The call to ‘NIO::Selector#select` will “wake up” and
The waiting/wake up is performed with nio4r, which will use the appropriate backend (libev, Java NIO or
which stores it in an array and waits for any of the connections to be ready for reading.
A connection comes into a `Puma::Server` instance, it is then passed to a `Puma::Reactor` instance,
## Reactor Flow
For a graphical representation of how the reactor works see [architecture.md](github.com/puma/puma/blob/master/docs/architecture.md#connection-pipeline).
it will have 5 workers and each worker will have it’s own reactor.
Each Puma “worker” process has its own Reactor. For example if you start puma with ‘$ puma -w 5` then
such as nginx) then the application would be subject to a slow client attack.
If read buffering is not done, and no other read buffering is performed (such as by an application server
completely received before it starts to be processed. This may be known as read buffering.
The Reactor object is responsible for ensuring that a request has been
Internal Docs, Not a public interface.

def add(c)

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

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

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

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

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

This method adds a connection to the reactor

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

def calculate_sleep

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

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

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

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

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

def clear!

Close all watched sockets and clear them from being watched

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

def clear_monitor(mon)

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

def initialize(server, app_pool)

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

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

Creates an instance of Puma::Reactor

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

def run

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

def run_in_thread

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

def run_internal

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

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

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

## Time Out Case

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

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

`ready` output looks like this: `[[#], [], []]`.
the `NIO::Selector#select` is immediately able to "wake" and read from the `Puma::Client`. At this point the
Since the `Puma::Client` in this example has data that has not been read yet,

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

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

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

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

## When a request is added:

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

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

def shutdown

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

Modules

Classes