Reactor refactor (#2279)

* Refactor Reactor and Client request buffering
Refactor Reactor into a more generic IO-with-timeout monitor,
using a Queue to simplify the implementation.
Move request-buffering logic into Server#reactor_wakeup.
Fixes bug in managing timeouts on clients.
Move, update and rewrite documentation to match updated class structure.

* Fix a few concurrency bugs
- In `Reactor#shutdown`, `@selector` can be closed before the call to `#wakeup`, so catch/ignore the `IOError` that may be thrown.
- `Reactor#wakeup!` can delete elements from the `@timeouts` array so calling it from an `#each` block can cause the array iteration to miss elements. Call @block directly instead.
- Change `Reactor#add` to return `false` if the reactor is already shut down instead of invoking the block immediately, so a client-request currently being processed can continue, rather than re-adding to the thread-pool (which may already be shutting down and unable to accept new work).

Co-authored-by: Nate Berkopec <nate.berkopec@gmail.com>

History.md

@@ -7,6 +7,8 @@
   * Cleanup daemonization in rc.d script (#2409)
 
 * Refactor
+  * Consolidate option handling in Server, Server small refactors, doc chang (#2389)
+  * Refactor Reactor and Client request buffering (#2279)
   * client.rb - remove JRuby specific 'finish' code (#2412)
   * Consolidate fast_write calls in Server, extract early_hints assembly (#2405)
   * Remove upstart from docs (#2408)

lib/puma/client.rb

@@ -103,7 +103,12 @@ def in_data_phase
     end
 
     def set_timeout(val)
-      @timeout_at = Time.now + val
+      @timeout_at = Process.clock_gettime(Process::CLOCK_MONOTONIC) + val
+    end
+
+    # Number of seconds until the timeout elapses.
+    def timeout
+      [@timeout_at - Process.clock_gettime(Process::CLOCK_MONOTONIC), 0].max
     end
 
     def reset(fast_check=true)
@@ -195,19 +200,13 @@ def eagerly_finish
     end
 
     def finish(timeout)
-      return true if @ready
-      until try_to_finish
-        can_read = begin
-          IO.select([@to_io], nil, nil, timeout)
-        rescue ThreadPool::ForceShutdown
-          nil
-        end
-        unless can_read
-          write_error(408) if in_data_phase
-          raise ConnectionError
-        end
-      end
-      true
+      return if @ready
+      IO.select([@to_io], nil, nil, timeout) || timeout! until try_to_finish
+    end
+
+    def timeout!
+      write_error(408) if in_data_phase
+      raise ConnectionError
     end
 
     def write_error(status_code)

lib/puma/queue_close.rb

@@ -0,0 +1,24 @@
+# Queue#close was added in Ruby 2.3.
+# Add a simple implementation for earlier Ruby versions.
+unless Queue.instance_methods.include?(:close)
+  class ClosedQueueError < StandardError; end
+  module Puma
+    module QueueClose
+      def initialize
+        @closed = false
+        super
+      end
+      def close
+        @closed = true
+      end
+      def closed?
+        @closed
+      end
+      def push(object)
+        raise ClosedQueueError if @closed
+        super
+      end
+    end
+    Queue.prepend QueueClose
+  end
+end