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 connection comes into a Puma::Server instance, it is then passed to a Puma::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 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 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 Puma::ThreadPool. Once a request is fully formed (header and body are received) it will be passed to the app_pool.



50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
 
# 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) ⇒ 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 ‘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 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 `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



374
375
376
377
378
379
 
# File 'lib/puma/reactor.rb', line 374

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

#calculate_sleep ⇒ Object

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.



335
336
337
338
339
340
341
342
343
344
345
346
347
 
# 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! ⇒ Object

Close all watched sockets and clear them from being watched



382
383
384
385
386
387
388
 
# 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

#run ⇒ Object 



302
303
304
305
306
307
 
# File 'lib/puma/reactor.rb', line 302

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

#run_in_thread ⇒ Object 



309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
 
# 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

#shutdown ⇒ Object 



390
391
392
393
394
395
396
397
398
 
# 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