Fix a race condition on recv_bytes boundary when request is invalid

A remote client may send a request that is exactly recv_bytes long,
followed by a secondary request using HTTP pipelining.

When request lookahead is disabled (default) we won't read any more
requests, and when the first request fails due to a parsing error, we
simply close the connection.

However when request lookahead is enabled, it is possible to process and
receive the first request, start sending the error message back to the
client while we read the next request and queue it. This will allow the
secondar request to be serviced by the worker thread while the
connection should be closed.

The fix here checks if we should not have read the data in the first
place (because the conection is going to be torn down) while we hold the
`requests_lock` which means the service thread can't be in the middle of
flipping the `close_when_flushed` flag.

src/waitress/channel.py

@@ -140,7 +140,7 @@ def readable(self):
         # 1. We're not already about to close the connection.
         # 2. We're not waiting to flush remaining data before closing the
         #    connection
-        # 3. There are not too many tasks already queued
+        # 3. There are not too many tasks already queued (if lookahead is enabled)
         # 4. There's no data in the output buffer that needs to be sent
         #    before we potentially create a new task.
 
@@ -196,6 +196,15 @@ def received(self, data):
             return False
 
         with self.requests_lock:
+            # Don't bother processing anymore data if this connection is about
+            # to close. This may happen if readable() returned True, on the
+            # main thread before the service thread set the close_when_flushed
+            # flag, and we read data but our service thread is attempting to
+            # shut down the connection due to an error. We want to make sure we
+            # do this while holding the request_lock so that we can't race
+            if self.will_close or self.close_when_flushed:
+                return False
+
             while data:
                 if self.request is None:
                     self.request = self.parser_class(self.adj)