Class: Puma::Reactor

Inherits:
Object
  • Object
show all
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](github.com/puma/puma/blob/master/docs/architecture.md#connection-pipeline).

## Reactor Flow

A request comes into a ‘Puma::Server` instance, it is then passed to a `Puma::Reactor` instance. The reactor stores the request in an array and calls `IO.select` on the array in a loop.

When the request is written to by the client then the ‘IO.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 collapse

DefaultSleepFor = 
5

Instance Method Summary collapse

Constructor Details

#initialize(server, app_pool) ⇒ Reactor

Creates an instance of Puma::Reactor

The ‘server` argument is an instance of `Puma::Server` this 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 `Puma::ThreadPool`. Once a request is fully formed (header and body are received) it will be passed to the `app_pool`.



47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/puma/reactor.rb', line 47

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

  @mutex = Mutex.new

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

  @sockets = [@ready]
end

Instance Method Details

#add(c) ⇒ Object

This method adds a connection to the reactor

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

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

Instead what needs to happen is that ‘IO.select` needs to be woken up, the contents of `@input` added to the `sockets` array, and then another call to `IO.select` needs to happen. Since the `Puma::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 `IO.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`



314
315
316
317
318
319
320
321
322
323
324
325
326
 
# File 'lib/puma/reactor.rb', line 314

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

    if c.timeout_at
      @timeouts << c
      @timeouts.sort! { |a,b| a.timeout_at <=> b.timeout_at }

      calculate_sleep
    end
  end
end

#calculate_sleep ⇒ Object

The ‘calculate_sleep` sets the value that the `IO.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.



275
276
277
278
279
280
281
282
283
284
285
286
287
 
# File 'lib/puma/reactor.rb', line 275

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

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

#clear! ⇒ Object

Close all watched sockets and clear them from being watched



329
330
331
332
333
334
335
 
# File 'lib/puma/reactor.rb', line 329

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

#run ⇒ Object 



243
244
245
246
247
248
 
# File 'lib/puma/reactor.rb', line 243

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

#run_in_thread ⇒ Object 



250
251
252
253
254
255
256
257
258
259
260
261
262
263
 
# File 'lib/puma/reactor.rb', line 250

def run_in_thread
  @thread = Thread.new do
    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

#shutdown ⇒ Object 



337
338
339
340
341
342
343
344
345
 
# File 'lib/puma/reactor.rb', line 337

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

  @thread.join
end