Prevent connections from entering Reactor after shutdown begins (#2377)

When a `Puma::Server` instance starts to shutdown, it sets the instance
variable `@queue_requests` to `false`, then starts to shut down the
Reactor. The intent here is that after the Reactor shuts down, threads
won't have the option of sending their connections to the Reactor (the
connection must either be closed, assuming we've written a response to
the socket, or the thread must wait until the client has finished
writing its request, then write a response, then finally close the
connection). This works most of the time just fine.

The problem is that there are races in the implementation of the
`ThreadPool` where it's possible for a thread to see that
`@queue_requests` is `true` before the `Server` shutdown has started,
then later try to add the request to the Reactor, not knowing that it's
already shutting down.

Depending on the precise order of operations, one possible outcome is
that the `ThreadPool` executes `@reactor.add` after the `@reactor` has
closed its `@trigger` pipe, resulting in the error `Error reached top of
thread-pool: closed stream (IOError)`. Clients experience connection
reset errors as a result.

The fix introduced in this commit is to add a `Mutex` that makes it
impossible for a thread in the `ThreadPool` to add a client connection
to the `@reactor` after `@queue_requests` has been set to `false`.

Co-Authored-By: Leo Belyi <leonid.belyi@mycase.com>

Co-authored-by: Leo Belyi <leonid.belyi@mycase.com>

History.md

@@ -4,7 +4,7 @@
   * Your feature goes here (#Github Number)
 
 * Bugfixes
-  * Your bugfix goes here (#Github Number)
+  * Prevent connections from entering Reactor after shutdown begins (#2377)
 
 * Refactor
   * Change Events#ssl_error signature from (error, peeraddr, peercert) to (error, ssl_socket) (#2375)

lib/puma/server.rb

@@ -95,6 +95,8 @@ def initialize(app, events=Events.stdio, options={})
       @precheck_closing = true
 
       @requests_count = 0
+
+      @shutdown_mutex = Mutex.new
     end
 
     def inherit_binder(bind)
@@ -247,12 +249,13 @@ def run(background=true)
 
           @events.connection_error e, client
         else
-          if process_now
-            process_client client, buffer
-          else
-            client.set_timeout @first_data_timeout
-            @reactor.add client
-          end
+          process_now ||= @shutdown_mutex.synchronize do
+                            next true unless @queue_requests
+                            client.set_timeout @first_data_timeout
+                            @reactor.add client
+                            false
+                          end
+          process_client client, buffer if process_now
         end
 
         process_now
@@ -338,7 +341,9 @@ def handle_servers
         @events.fire :state, @status
 
         if queue_requests
-          @queue_requests = false
+          @shutdown_mutex.synchronize do
+            @queue_requests = false
+          end
           @reactor.clear!
           @reactor.shutdown
         end
@@ -418,11 +423,13 @@ def process_client(client, buffer)
             end
 
             unless client.reset(check_for_more_data)
-              return unless @queue_requests
-              close_socket = false
-              client.set_timeout @persistent_timeout
-              @reactor.add client
-              return
+              @shutdown_mutex.synchronize do
+                return unless @queue_requests
+                close_socket = false
+                client.set_timeout @persistent_timeout
+                @reactor.add client
+                return
+              end
             end
           end
         end