feat: FIFO policy to handle large no. of requests

- add required test -- test by sending 1000 requests
- increase test timeout to 180 seconds to account for tests taking long
time

scrapy/core/http2/protocol.py

@@ -38,13 +38,16 @@ class H2ClientProtocol(Protocol):
         # Streams are stored in a dictionary keyed off their stream IDs
         self.streams: Dict[int, Stream] = {}
 
-        # Boolean to keep track the connection is made
-        # If requests are received before connection is made
-        # we keep all requests in a pool and send them as the connection
-        # is made
-        self.is_connection_made = False
+        # If requests are received before connection is made we keep
+        # all requests in a pool and send them as the connection is made
         self._pending_request_stream_pool = deque()
 
+        # Counter to keep track of opened stream. This counter
+        # is used to make sure that not more than MAX_CONCURRENT_STREAMS
+        # streams are opened which leads to ProtocolError
+        # We use simple FIFO policy to handle pending requests
+        self._active_streams = 0
+
         # Save an instance of ProtocolError raised by hyper-h2
         # We pass this instance to the streams ResponseFailed() failure
         self._protocol_error: Union[None, ProtocolError] = None
@@ -54,10 +57,46 @@ class H2ClientProtocol(Protocol):
             'ip_address': None
         }
 
+    @property
+    def is_connected(self):
+        """Boolean to keep track of the connection status.
+        This is used while initiating pending streams to make sure
+        that we initiate stream only during active HTTP/2 Connection
+        """
+        return bool(self.transport.connected)
+
+    @property
+    def allowed_max_concurrent_streams(self) -> int:
+        """We keep total two streams for client (sending data) and
+        server side (receiving data) for a single request. To be safe
+        we choose the minimum. Since this value can change in event
+        RemoteSettingsChanged we make variable a property.
+        """
+        return min(
+            self.conn.local_settings.max_concurrent_streams,
+            self.conn.remote_settings.max_concurrent_streams
+        )
+
+    def _send_pending_requests(self):
+        """Initiate all pending requests from the deque following FIFO
+        We make sure that at any time {allowed_max_concurrent_streams}
+        streams are active.
+        """
+        while (
+            self._pending_request_stream_pool
+            and self._active_streams < self.allowed_max_concurrent_streams
+            and self.is_connected
+        ):
+            self._active_streams += 1
+            stream = self._pending_request_stream_pool.popleft()
+            stream.initiate_request()
+
     def _stream_close_cb(self, stream_id: int):
         """Called when stream is closed completely
         """
-        self.streams.pop(stream_id, None)
+        self.streams.pop(stream_id)
+        self._active_streams -= 1
+        self._send_pending_requests()
 
     def _new_stream(self, request: Request):
         """Instantiates a new Stream object
@@ -75,13 +114,6 @@ class H2ClientProtocol(Protocol):
         self.streams[stream.stream_id] = stream
         return stream
 
-    def _send_pending_requests(self):
-        # TODO: handle MAX_CONCURRENT_STREAMS
-        # Initiate all pending requests
-        while self._pending_request_stream_pool:
-            stream = self._pending_request_stream_pool.popleft()
-            stream.initiate_request()
-
     def _write_to_transport(self):
         """ Write data to the underlying transport connection
         from the HTTP2 connection instance if any
@@ -89,19 +121,12 @@ class H2ClientProtocol(Protocol):
         data = self.conn.data_to_send()
         self.transport.write(data)
 
-        logger.debug("Sent {} bytes to {} via transport".format(len(data), self._metadata['ip_address']))
-
     def request(self, request: Request):
         stream = self._new_stream(request)
         d = stream.get_response()
 
-        # If connection is not yet established then add the
-        # stream to pool or initiate request
-        if self.is_connection_made:
-            stream.initiate_request()
-        else:
-            self._pending_request_stream_pool.append(stream)
-
+        # Add the stream to the request pool
+        self._pending_request_stream_pool.append(stream)
         return d
 
     def connectionMade(self):
@@ -116,7 +141,6 @@ class H2ClientProtocol(Protocol):
 
         self.conn.initiate_connection()
         self._write_to_transport()
-        self.is_connection_made = True
 
     def dataReceived(self, data):
         try:
@@ -144,7 +168,10 @@ class H2ClientProtocol(Protocol):
         # which raises `RuntimeError: dictionary changed size during iteration`
         # Hence, we copy the streams into a list.
         for stream in list(self.streams.values()):
-            stream.close(StreamCloseReason.CONNECTION_LOST, self._protocol_error)
+            if stream.request_sent:
+                stream.close(StreamCloseReason.CONNECTION_LOST, self._protocol_error)
+            else:
+                stream.close(StreamCloseReason.INACTIVE)
 
         self.conn.close_connection()
 
@@ -160,7 +187,6 @@ class H2ClientProtocol(Protocol):
                 triggered by sending data
         """
         for event in events:
-            logger.debug(event)
             if isinstance(event, DataReceived):
                 self.data_received(event)
             elif isinstance(event, ResponseReceived):
@@ -174,7 +200,7 @@ class H2ClientProtocol(Protocol):
             elif isinstance(event, SettingsAcknowledged):
                 self.settings_acknowledged(event)
             else:
-                logger.info("Received unhandled event {}".format(event))
+                logger.debug("Received unhandled event {}".format(event))
 
     # Event handler functions starts here
     def data_received(self, event: DataReceived):
@@ -184,8 +210,8 @@ class H2ClientProtocol(Protocol):
         self.streams[event.stream_id].receive_headers(event.headers)
 
     def settings_acknowledged(self, event: SettingsAcknowledged):
-        # Send off all the pending requests
-        # as now we have established a proper HTTP/2 connection
+        # Send off all the pending requests as now we have
+        # established a proper HTTP/2 connection
         self._send_pending_requests()
 
     def stream_ended(self, event: StreamEnded):
@@ -195,9 +221,8 @@ class H2ClientProtocol(Protocol):
         self.streams[event.stream_id].close(StreamCloseReason.RESET)
 
     def window_updated(self, event: WindowUpdated):
-        stream_id = event.stream_id
-        if stream_id != 0:
-            self.streams[stream_id].receive_window_update()
+        if event.stream_id != 0:
+            self.streams[event.stream_id].receive_window_update()
         else:
             # Send leftover data for all the streams
             for stream in self.streams.values():

scrapy/core/http2/stream.py

@@ -8,6 +8,7 @@ from h2.connection import H2Connection
 from h2.errors import ErrorCodes
 from h2.exceptions import StreamClosedError
 from twisted.internet.defer import Deferred, CancelledError
+from twisted.internet.error import ConnectionClosed
 from twisted.python.failure import Failure
 from twisted.web.client import ResponseFailed
 
@@ -19,6 +20,15 @@ from scrapy.responsetypes import responsetypes
 logger = logging.getLogger(__name__)
 
 
+class InactiveStreamClosed(ConnectionClosed):
+    """Connection was closed without sending request headers
+    of the stream. This happens when a stream is waiting for other
+    streams to close and connection is lost."""
+
+    def __init__(self, request: Request):
+        self.request = request
+
+
 class StreamCloseReason(Enum):
     # Received a StreamEnded event
     ENDED = 1
@@ -32,10 +42,13 @@ class StreamCloseReason(Enum):
     # Expected response body size is more than allowed limit
     MAXSIZE_EXCEEDED = 4
 
-    # When the response deferred is cancelled by the client
+    # Response deferred is cancelled by the client
     # (happens when client called response_deferred.cancel())
     CANCELLED = 5
 
+    # Connection lost and the stream was not initiated
+    INACTIVE = 6
+
 
 class Stream:
     """Represents a single HTTP/2 Stream.
@@ -108,8 +121,12 @@ class Stream:
         }
 
         def _cancel(_):
-            # Close this stream as gracefully as possible :)
-            self.reset_stream(StreamCloseReason.CANCELLED)
+            # Close this stream as gracefully as possible
+            # Check if the stream has started
+            if self.request_sent:
+                self.reset_stream(StreamCloseReason.CANCELLED)
+            else:
+                self.close(StreamCloseReason.CANCELLED)
 
         self._deferred_response = Deferred(_cancel)
 
@@ -355,6 +372,9 @@ class Stream:
                 error if error else Failure()
             ]))
 
+        elif reason is StreamCloseReason.INACTIVE:
+            self._deferred_response.errback(InactiveStreamClosed(self._request))
+
     def _fire_response_deferred(self, flags: List[str] = None):
         """Builds response from the self._response dict
         and fires the response deferred callback with the

tests/test_http2_client_protocol.py

@@ -18,6 +18,7 @@ from twisted.web.server import Site, NOT_DONE_YET
 from twisted.web.static import File
 
 from scrapy.core.http2.protocol import H2ClientProtocol
+from scrapy.core.http2.stream import InactiveStreamClosed
 from scrapy.http import Request, Response, JsonRequest
 from scrapy.utils.python import to_bytes, to_unicode
 from tests.mockserver import ssl_context_factory, LeafResource
@@ -174,10 +175,14 @@ class Https2ClientProtocolTestCase(TestCase):
         client_endpoint = SSL4ClientEndpoint(reactor, self.hostname, self.port_number, client_options)
         self.client = yield client_endpoint.connect(h2_client_factory)
 
+        # Increase the total time taken for each tests
+        self.timeout = 180 # default is 120 seconds
+
     @inlineCallbacks
     def tearDown(self):
-        yield self.client.transport.loseConnection()
-        yield self.client.transport.abortConnection()
+        if self.client.is_connected:
+            yield self.client.transport.loseConnection()
+            yield self.client.transport.abortConnection()
         yield self.server.stopListening()
         shutil.rmtree(self.temp_directory)
 
@@ -430,3 +435,45 @@ class Https2ClientProtocolTestCase(TestCase):
         )
 
         yield self._check_log_warnsize(request, warn_pattern, Data.NO_CONTENT_LENGTH)
+
+    def test_max_concurrent_streams(self):
+        """Send 1000 requests to check if we can handle
+        very large number of request
+        """
+
+        def get_deferred():
+            return self._check_GET(
+                Request(self.get_url('/get-data-html-small')),
+                Data.HTML_SMALL,
+                200
+            )
+
+        return self._check_repeat(get_deferred, 1000)
+
+    def test_inactive_stream(self):
+        """Here we send 110 requests considering the MAX_CONCURRENT_STREAMS
+        by default is 100. After sending the first 100 requests we close the
+        connection."""
+        d_list = []
+
+        def assert_inactive_stream(failure):
+            self.assertIsNotNone(failure.check(InactiveStreamClosed))
+
+        # Send 100 request (we do not check the result)
+        for _ in range(100):
+            d = self.client.request(Request(self.get_url('/get-data-html-small')))
+            d.addBoth(lambda _: None)
+            d_list.append(d)
+
+        # Now send 10 extra request and save the response deferred in a list
+        for _ in range(10):
+            d = self.client.request(Request(self.get_url('/get-data-html-small')))
+            d.addCallback(self.fail)
+            d.addErrback(assert_inactive_stream)
+            d_list.append(d)
+
+        # Close the connection now to fire all the extra 10 requests errback
+        # with InactiveStreamClosed
+        self.client.transport.abortConnection()
+
+        return DeferredList(d_list, consumeErrors=True, fireOnOneErrback=True)