From 62ce842afc8ef829ffd6f164712a8a9413eb9e1d Mon Sep 17 00:00:00 2001 From: Aditya Date: Wed, 15 Jul 2020 07:50:53 +0530 Subject: [PATCH] fix: multiple h2 connections to same uri - When multiple requests are sent to H2ConnectionPool to the same uri while the connection is in connecting state -- multiple connections were establised. - Fixed the bug using a deque of all the request deferred's which fire with the H2ClientProtocol (connection) instance when connection is established --- scrapy/core/http2/agent.py | 55 ++++++++++++++++++++++++++++++++------ 1 file changed, 47 insertions(+), 8 deletions(-) diff --git a/scrapy/core/http2/agent.py b/scrapy/core/http2/agent.py index 566c074c2..7a8847c38 100644 --- a/scrapy/core/http2/agent.py +++ b/scrapy/core/http2/agent.py @@ -1,11 +1,11 @@ -from typing import Dict, Tuple, Optional +from collections import deque +from typing import Deque, Dict, List, Tuple, Optional from twisted.internet import defer from twisted.internet._sslverify import _setAcceptableProtocols, ClientTLSOptions from twisted.internet.base import ReactorBase from twisted.internet.defer import Deferred -from twisted.internet.endpoints import SSL4ClientEndpoint -from twisted.python.failure import Failure +from twisted.internet.endpoints import HostnameEndpoint from twisted.web.client import URI, BrowserLikePolicyForHTTPS, _StandardEndpointFactory from twisted.web.iweb import IPolicyForHTTPS from zope.interface import implementer @@ -20,31 +20,70 @@ class H2ConnectionPool: def __init__(self, reactor: ReactorBase, settings: Settings) -> None: self._reactor = reactor self.settings = settings + + # Store a dictionary which is used to get the respective + # H2ClientProtocolInstance using the key as Tuple(scheme, hostname, port) self._connections: Dict[Tuple, H2ClientProtocol] = {} - def get_connection(self, uri: URI, endpoint: SSL4ClientEndpoint) -> Deferred: + # Save all requests that arrive before the connection is established + self._pending_requests: Dict[Tuple, Deque[Deferred]] = {} + + def get_connection(self, uri: URI, endpoint: HostnameEndpoint) -> Deferred: key = (uri.scheme, uri.host, uri.port) + if key in self._pending_requests: + # Received a request while connecting to remote + # Create a deferred which will fire with the H2ClientProtocol + # instance + d = Deferred() + self._pending_requests[key].append(d) + return d + + # Check if we already have a connection to the remote conn = self._connections.get(key, None) if conn: + # Return this connection instance wrapped inside a deferred return defer.succeed(conn) + + # No connection is established for the given URI return self._new_connection(key, uri, endpoint) - def _new_connection(self, key: Tuple, uri: URI, endpoint: SSL4ClientEndpoint) -> Deferred: + def _new_connection(self, key: Tuple, uri: URI, endpoint: HostnameEndpoint) -> Deferred: + self._pending_requests[key] = deque() + conn_lost_deferred = Deferred() conn_lost_deferred.addCallback(self._remove_connection, key) factory = H2ClientFactory(uri, self.settings, conn_lost_deferred) - d = endpoint.connect(factory) - d.addCallback(self.put_connection, key) + conn_d = endpoint.connect(factory) + conn_d.addCallback(self.put_connection, key) + + d = Deferred() + self._pending_requests[key].append(d) return d def put_connection(self, conn: H2ClientProtocol, key: Tuple) -> H2ClientProtocol: self._connections[key] = conn + + # Now as we have established a proper HTTP/2 connection + # we fire all the deferred's with the connection instance + pending_requests = self._pending_requests.pop(key) + while pending_requests: + d = pending_requests.popleft() + d.callback(conn) + + del pending_requests + return conn - def _remove_connection(self, reason: Failure, key: Tuple) -> None: + def _remove_connection(self, errors: List[BaseException], key: Tuple) -> None: self._connections.pop(key) + # Call the errback of all the pending requests for this connection + pending_requests = self._pending_requests.pop(key, None) + while pending_requests: + d = pending_requests.popleft() + d.errback(errors) + @implementer(IPolicyForHTTPS) class H2WrappedContextFactory: