From 1b937b32a2577acffed13899c9701fee3e6e8327 Mon Sep 17 00:00:00 2001 From: Adrian Chaves Date: Fri, 7 Aug 2026 19:06:39 +0200 Subject: [PATCH] Add CONCURRENT_CONNECTIONS_PER_HANDLER and CONNECTION_KEEPALIVE_TIMEOUT --- docs/faq.rst | 17 +++ docs/topics/broad-crawls.rst | 4 + docs/topics/settings.rst | 53 +++++++++ scrapy/core/downloader/handlers/_base_http.py | 51 +++++++++ .../downloader/handlers/_base_streaming.py | 5 - scrapy/core/downloader/handlers/_httpx.py | 3 +- scrapy/core/downloader/handlers/http11.py | 104 +++++++++++++++++- scrapy/core/downloader/handlers/http2.py | 6 +- scrapy/core/http2/agent.py | 75 ++++++++++--- scrapy/core/http2/protocol.py | 17 ++- scrapy/settings/default_settings.py | 4 + tests/mockserver/http.py | 2 + tests/mockserver/http_resources.py | 33 ++++++ tests/test_downloader_handler_httpx.py | 16 ++- .../test_downloader_handler_twisted_http11.py | 54 +++++++++ tests/test_http2_client_protocol.py | 4 +- tests/utils/bases/download_handlers_http.py | 55 +++++++++ 17 files changed, 464 insertions(+), 39 deletions(-) diff --git a/docs/faq.rst b/docs/faq.rst index 80658a5bf..04de5a0e7 100644 --- a/docs/faq.rst +++ b/docs/faq.rst @@ -413,6 +413,23 @@ inline within web pages, similar to external resources. The "data:" scheme with content (",") essentially creates a request to a data URL without any specific content. +.. _faq-too-many-open-files: + +I get ``too many file descriptors in select()`` or ``Too many open files`` +-------------------------------------------------------------------------- + +The crawl is opening more sockets than the system allows. On Windows +:func:`select.select` handles at most 512 sockets; elsewhere the ceiling is the +maximum number of open files allowed for the process, which sockets, log files +and feed exports all count towards. + +Lower :setting:`CONCURRENT_CONNECTIONS_PER_HANDLER`, keeping in mind that every +:ref:`download handler ` applies it on its own, so a +crawl of both ``http://`` and ``https://`` URLs can open twice as many +connections. Raising the limit of the operating system works as well on +platforms other than Windows. + + Running ``runspider`` I get ``error: No spider found in file: `` -------------------------------------------------------------------------- diff --git a/docs/topics/broad-crawls.rst b/docs/topics/broad-crawls.rst index cace1f883..6379ed131 100644 --- a/docs/topics/broad-crawls.rst +++ b/docs/topics/broad-crawls.rst @@ -68,6 +68,10 @@ should pick a concurrency where CPU usage is at 80-90%. Increasing concurrency also increases memory usage. If memory usage is a concern, you might need to lower your global concurrency limit accordingly. +Concurrency also drives how many connections stay open, which is capped +separately by :setting:`CONCURRENT_CONNECTIONS_PER_HANDLER`. See +:ref:`faq-too-many-open-files` if a crawl runs out of file descriptors. + Increase Twisted IO thread pool maximum size ============================================ diff --git a/docs/topics/settings.rst b/docs/topics/settings.rst index 65ee77258..69c4e831a 100644 --- a/docs/topics/settings.rst +++ b/docs/topics/settings.rst @@ -557,6 +557,44 @@ project name). This name will be used for the logging too. It's automatically populated with your project name when you create your project with the :command:`startproject` command. +.. setting:: CONCURRENT_CONNECTIONS_PER_HANDLER + +CONCURRENT_CONNECTIONS_PER_HANDLER +---------------------------------- + +Default: ``None`` + +.. versionadded:: VERSION + +The maximum number of connections that a single :ref:`download handler +` may keep open, whether they are being used for a +request or kept around for reuse by a later request. Use ``0`` for no limit. + +``None`` lets Scrapy pick a limit that fits the system it runs on. How it is +picked may change in any future Scrapy version, without a deprecation period, +so set a number if you need it to stay the same. + +Scrapy currently picks half of the maximum number of open files allowed for the +process, or 480 on Windows, where :func:`select.select` handles at most 512 +sockets. + +Each download handler applies this limit on its own, so a crawl of both +``http://`` and ``https://`` URLs may open up to twice as many connections. +:class:`~scrapy.core.downloader.handlers._httpx.HttpxDownloadHandler` +additionally applies it once per proxy, since httpx `cannot use a different +proxy per request `_. + +HTTP/2 multiplexes requests over a single connection per remote, so +:class:`~scrapy.core.downloader.handlers.http2.H2DownloadHandler` reaches this +limit only after as many remotes, regardless of concurrency, and it keeps a +connection that is serving requests even if that leaves it over the limit. + +How the limit is reached also depends on the handler: some close the connection +that has been idle the longest to make room for a new one, others make requests +wait for a connection to free up. Either way, a value below +:setting:`CONCURRENT_REQUESTS` keeps the crawl from reaching the configured +concurrency, and logs a warning. + .. setting:: CONCURRENT_ITEMS CONCURRENT_ITEMS @@ -593,6 +631,21 @@ See also: :ref:`topics-autothrottle` and its It is possible to change this setting per domain by using :setting:`DOWNLOAD_SLOTS`. +.. setting:: CONNECTION_KEEPALIVE_TIMEOUT + +CONNECTION_KEEPALIVE_TIMEOUT +---------------------------- + +Default: ``240.0`` + +.. versionadded:: VERSION + +How long, in seconds, a connection is kept open for reuse after the response it +was serving is complete. Use ``0`` to close connections as soon as they become +idle. + +Servers usually close idle connections earlier than this on their own. + .. setting:: DEFAULT_DROPITEM_LOG_LEVEL DEFAULT_DROPITEM_LOG_LEVEL diff --git a/scrapy/core/downloader/handlers/_base_http.py b/scrapy/core/downloader/handlers/_base_http.py index 83d130462..930c22285 100644 --- a/scrapy/core/downloader/handlers/_base_http.py +++ b/scrapy/core/downloader/handlers/_base_http.py @@ -1,12 +1,55 @@ from __future__ import annotations +import logging +import sys from abc import ABC from typing import TYPE_CHECKING from .base import BaseDownloadHandler +try: + # stdlib's resource module is only available on unix platforms + import resource +except ImportError: + resource = None # type: ignore[assignment] + if TYPE_CHECKING: from scrapy.crawler import Crawler + from scrapy.settings import BaseSettings + + +logger = logging.getLogger(__name__) + +# The number of sockets that select() accepts in CPython's Windows build; see +# FD_SETSIZE in Modules/selectmodule.c. +_FD_SETSIZE = 512 + +# Room left for the sockets that a Windows crawl registers with the event loop +# outside of download handlers, e.g. the asyncio self-pipe and the telnet +# console. +_WINDOWS_RESERVED_SOCKETS = 32 + + +def _auto_connection_limit() -> int: + if sys.platform == "win32": + return _FD_SETSIZE - _WINDOWS_RESERVED_SOCKETS + soft_limit: int = resource.getrlimit(resource.RLIMIT_NOFILE)[0] + if soft_limit == resource.RLIM_INFINITY: + return 0 + return soft_limit // 2 + + +def _get_connection_limit(settings: BaseSettings) -> int: + limit = settings.get("CONCURRENT_CONNECTIONS_PER_HANDLER") + limit = _auto_connection_limit() if limit is None else int(limit) + concurrent_requests = settings.getint("CONCURRENT_REQUESTS") + if 0 < limit < concurrent_requests: + logger.warning( + f"CONCURRENT_CONNECTIONS_PER_HANDLER ({limit}) is lower than" + f" CONCURRENT_REQUESTS ({concurrent_requests}), which keeps the" + f" crawl from reaching the configured concurrency." + ) + return limit class BaseHttpDownloadHandler(BaseDownloadHandler, ABC): @@ -23,3 +66,11 @@ class BaseHttpDownloadHandler(BaseDownloadHandler, ABC): "DOWNLOADER_CLIENT_TLS_VERBOSE_LOGGING" ) self._fail_on_dataloss_warned: bool = False + # these are useful for many handlers but used in different ways by them + self._pool_size_total: int = _get_connection_limit(crawler.settings) + self._pool_size_per_host: int = crawler.settings.getint( + "CONCURRENT_REQUESTS_PER_DOMAIN" + ) + self._keepalive_timeout: float = crawler.settings.getfloat( + "CONNECTION_KEEPALIVE_TIMEOUT" + ) diff --git a/scrapy/core/downloader/handlers/_base_streaming.py b/scrapy/core/downloader/handlers/_base_streaming.py index 5a669c732..d580baf1d 100644 --- a/scrapy/core/downloader/handlers/_base_streaming.py +++ b/scrapy/core/downloader/handlers/_base_streaming.py @@ -84,11 +84,6 @@ class BaseStreamingDownloadHandler(BaseHttpDownloadHandler, ABC, Generic[_Respon crawler.settings.get("DOWNLOAD_BIND_ADDRESS") ) self._proxy_auth_encoding: str = crawler.settings.get("HTTPPROXY_AUTH_ENCODING") - # these are useful for many handlers but used in different ways by them - self._pool_size_total: int = crawler.settings.getint("CONCURRENT_REQUESTS") - self._pool_size_per_host: int = crawler.settings.getint( - "CONCURRENT_REQUESTS_PER_DOMAIN" - ) @staticmethod @abstractmethod diff --git a/scrapy/core/downloader/handlers/_httpx.py b/scrapy/core/downloader/handlers/_httpx.py index 8bbffb233..da39e1bbc 100644 --- a/scrapy/core/downloader/handlers/_httpx.py +++ b/scrapy/core/downloader/handlers/_httpx.py @@ -93,10 +93,11 @@ class HttpxDownloadHandler(_Base): self._bind_host: str | None = self._get_bind_address_host() self._limits: httpx.Limits = httpx.Limits( # hard limit on simultaneous connections (None for no limit, which - # is what a CONCURRENT_REQUESTS of 0 means) + # is what a CONCURRENT_CONNECTIONS_PER_HANDLER of 0 means) max_connections=self._pool_size_total or None, # total number of idle connections in the pool (extra ones are closed) max_keepalive_connections=self._pool_size_total or None, + keepalive_expiry=self._keepalive_timeout, ) self._default_client: httpx.AsyncClient = self._make_client() diff --git a/scrapy/core/downloader/handlers/http11.py b/scrapy/core/downloader/handlers/http11.py index 17dca5acb..83fa2b526 100644 --- a/scrapy/core/downloader/handlers/http11.py +++ b/scrapy/core/downloader/handlers/http11.py @@ -17,12 +17,14 @@ from twisted.internet.defer import Deferred, succeed from twisted.internet.endpoints import TCP4ClientEndpoint from twisted.internet.protocol import Factory, Protocol, connectionDone from twisted.python.failure import Failure +from twisted.web._newclient import HTTP11ClientProtocol from twisted.web.client import ( URI, Agent, HTTPConnectionPool, ResponseDone, ResponseFailed, + _HTTP11ClientFactory, ) from twisted.web.client import Response as TxResponse from twisted.web.http import PotentialDataLoss, _DataLoss @@ -59,8 +61,10 @@ from scrapy.utils.url import add_http_if_no_scheme from ._base_http import BaseHttpDownloadHandler if TYPE_CHECKING: + from collections.abc import Callable + from twisted.internet.base import ReactorBase - from twisted.internet.interfaces import IConsumer + from twisted.internet.interfaces import IAddress, IConsumer # typing.NotRequired requires Python 3.11 from typing_extensions import NotRequired @@ -82,6 +86,95 @@ class _ResultT(TypedDict): stop_download: NotRequired[StopDownload | None] +class _TrackedHTTP11ClientProtocol(HTTP11ClientProtocol): + """Connection that tells the pool that opened it when it closes, whether it + was cached at the time or in use.""" + + _pool: _BoundedHTTPConnectionPool + _pool_key: Any + + def connectionLost(self, reason: Failure = connectionDone) -> None: + self._pool._forget(self) + super().connectionLost(reason) # type: ignore[no-untyped-call] + + +class _BoundedHTTP11ClientFactory(_HTTP11ClientFactory): + noisy = False + + def __init__( + self, + quiescentCallback: Callable[[HTTP11ClientProtocol], None], + metadata: str, + pool: _BoundedHTTPConnectionPool, + key: Any, + ): + super().__init__(quiescentCallback, metadata) # type: ignore[no-untyped-call] + self._pool: _BoundedHTTPConnectionPool = pool + self._key: Any = key + + def buildProtocol(self, addr: IAddress | None) -> _TrackedHTTP11ClientProtocol: + protocol = _TrackedHTTP11ClientProtocol(self._quiescentCallback) # type: ignore[no-untyped-call] + protocol._pool = self._pool + protocol._pool_key = self._key + self._pool._open.add(protocol) + return protocol + + +class _BoundedHTTPConnectionPool(HTTPConnectionPool): + """Connection pool that keeps the number of open connections, cached and in + use alike, under *limit*, where 0 means no limit. + + Opening a connection once the limit is reached closes the least recently + used cached connection to make room for it. When there is no cached + connection left to close, the limit is exceeded rather than requests + delayed. + """ + + _factory = _BoundedHTTP11ClientFactory # type: ignore[assignment] + + def __init__(self, reactor: ReactorBase, limit: int, keepalive_timeout: float): + super().__init__(reactor, persistent=bool(keepalive_timeout)) # type: ignore[no-untyped-call] + self.cachedConnectionTimeout = keepalive_timeout # type: ignore[assignment] + self._limit: int = limit + self._open: set[_TrackedHTTP11ClientProtocol] = set() + + def _newConnection( + self, key: Any, endpoint: TCP4ClientEndpoint + ) -> Deferred[HTTP11ClientProtocol]: + while self._limit and len(self._open) >= self._limit and self._evict(): + pass + + def quiescentCallback(protocol: HTTP11ClientProtocol) -> None: + self._putConnection(key, protocol) # type: ignore[no-untyped-call] + + factory = self._factory(quiescentCallback, repr(endpoint), self, key) + return endpoint.connect(factory) + + def _forget(self, connection: _TrackedHTTP11ClientProtocol) -> None: + self._open.discard(connection) + + def _evict(self) -> bool: + """Close the least recently used cached connection, and report whether + there was one to close. + + ``_timeouts`` holds cached connections only, and ``getConnection()`` + drops the ones it hands out, so its insertion order goes from least to + most recently used. + """ + connection = next(iter(self._timeouts), None) + if connection is None: + return False + self._timeouts[connection].cancel() + # dropped early so that a caller looping on this method sees the room + # made by connections that have yet to finish closing + self._forget(connection) + key = connection._pool_key + self._removeConnection(key, connection) # type: ignore[no-untyped-call] + if not self._connections[key]: + del self._connections[key] + return True + + class HTTP11DownloadHandler(BaseHttpDownloadHandler): def __init__(self, crawler: Crawler): if not crawler.settings.getbool("TWISTED_REACTOR_ENABLED"): @@ -91,11 +184,10 @@ class HTTP11DownloadHandler(BaseHttpDownloadHandler): from twisted.internet import reactor - self._pool: HTTPConnectionPool = HTTPConnectionPool(reactor, persistent=True) - self._pool.maxPersistentPerHost = crawler.settings.getint( - "CONCURRENT_REQUESTS_PER_DOMAIN" + self._pool: _BoundedHTTPConnectionPool = _BoundedHTTPConnectionPool( + reactor, self._pool_size_total, self._keepalive_timeout ) - self._pool._factory.noisy = False + self._pool.maxPersistentPerHost = self._pool_size_per_host self._contextFactory: IPolicyForHTTPS = _load_context_factory_from_settings( crawler @@ -137,7 +229,7 @@ class HTTP11DownloadHandler(BaseHttpDownloadHandler): async def close(self) -> None: from twisted.internet import reactor - d: Deferred[None] = self._pool.closeCachedConnections() + d: Deferred[None] = self._pool.closeCachedConnections() # type: ignore[no-untyped-call] # closeCachedConnections will hang on network or server issues, so # we'll manually timeout the deferred. # diff --git a/scrapy/core/downloader/handlers/http2.py b/scrapy/core/downloader/handlers/http2.py index f60c58d1b..6b69093dc 100644 --- a/scrapy/core/downloader/handlers/http2.py +++ b/scrapy/core/downloader/handlers/http2.py @@ -5,7 +5,7 @@ from typing import TYPE_CHECKING from urllib.parse import urldefrag from scrapy.core.downloader.contextfactory import _load_context_factory_from_settings -from scrapy.core.downloader.handlers.base import BaseDownloadHandler +from scrapy.core.downloader.handlers._base_http import BaseHttpDownloadHandler from scrapy.core.http2.agent import H2Agent, H2ConnectionPool from scrapy.exceptions import ( DownloadTimeoutError, @@ -29,7 +29,7 @@ if TYPE_CHECKING: from scrapy.spiders import Spider -class H2DownloadHandler(BaseDownloadHandler): +class H2DownloadHandler(BaseHttpDownloadHandler): lazy = True def __init__(self, crawler: Crawler): @@ -40,7 +40,7 @@ class H2DownloadHandler(BaseDownloadHandler): from twisted.internet import reactor - self._pool = H2ConnectionPool(reactor, crawler.settings) + self._pool = H2ConnectionPool(reactor, crawler.settings, self._pool_size_total) self._context_factory = _load_context_factory_from_settings(crawler) self._bind_address = crawler.settings.get("DOWNLOAD_BIND_ADDRESS") diff --git a/scrapy/core/http2/agent.py b/scrapy/core/http2/agent.py index aa55e29a0..7ebe28814 100644 --- a/scrapy/core/http2/agent.py +++ b/scrapy/core/http2/agent.py @@ -30,9 +30,12 @@ ConnectionKeyT = tuple[bytes, bytes, int] class H2ConnectionPool: - def __init__(self, reactor: ReactorBase, settings: Settings) -> None: + def __init__( + self, reactor: ReactorBase, settings: Settings, limit: int = 0 + ) -> None: self._reactor = reactor self.settings = settings + self._limit = limit # Store a dictionary which is used to get the respective # H2ClientProtocolInstance using the key as Tuple(scheme, hostname, port) @@ -58,9 +61,13 @@ class H2ConnectionPool: 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: + # Check if we already have a usable connection to the remote, moving + # it to the end so that the pool stays ordered from least to most + # recently used + conn = self._connections.get(key) + if conn and not conn.closing: + del self._connections[key] + self._connections[key] = conn # Return this connection instance wrapped inside a deferred return defer.succeed(conn) @@ -70,10 +77,14 @@ class H2ConnectionPool: def _new_connection( self, key: ConnectionKeyT, uri: URI, endpoint: HostnameEndpoint ) -> Deferred[H2ClientProtocol]: - self._pending_requests[key] = deque() + self._enforce_limit() + pending_requests: deque[Deferred[H2ClientProtocol]] = deque() + self._pending_requests[key] = pending_requests conn_lost_deferred: Deferred[list[BaseException]] = Deferred() - conn_lost_deferred.addCallback(self._remove_connection, key) + conn_lost_deferred.addCallback( + self._fail_pending_requests, key, pending_requests + ) factory = H2ClientFactory( uri, @@ -82,16 +93,40 @@ class H2ConnectionPool: tls_verbose_logging=self._tls_verbose_logging, ) conn_d = endpoint.connect(factory) - conn_d.addCallback(self.put_connection, key) + conn_d.addCallback(self.put_connection, key, conn_lost_deferred) d: Deferred[H2ClientProtocol] = Deferred() - self._pending_requests[key].append(d) + pending_requests.append(d) return d + def _enforce_limit(self) -> None: + """Close the least recently used connections that are not serving any + stream, to make room for one more connection. + + HTTP/2 multiplexes requests over a single connection per remote, so a + connection with active streams is kept even if that leaves the pool + over the limit. + """ + if not self._limit: + return + surplus = len(self._connections) + len(self._pending_requests) + 1 - self._limit + for key, conn in list(self._connections.items()): + if surplus <= 0: + return + if conn.metadata["active_streams"]: + continue + del self._connections[key] + conn.close_idle() + surplus -= 1 + def put_connection( - self, conn: H2ClientProtocol, key: ConnectionKeyT + self, + conn: H2ClientProtocol, + key: ConnectionKeyT, + conn_lost_deferred: Deferred[list[BaseException]], ) -> H2ClientProtocol: self._connections[key] = conn + conn_lost_deferred.addCallback(self._remove_connection, key, conn) # Now as we have established a proper HTTP/2 connection # we fire all the deferred's with the connection instance @@ -103,15 +138,27 @@ class H2ConnectionPool: return conn def _remove_connection( - self, errors: list[BaseException], key: ConnectionKeyT - ) -> None: - self._connections.pop(key) + self, errors: list[BaseException], key: ConnectionKeyT, conn: H2ClientProtocol + ) -> list[BaseException]: + # a newer connection may have taken over the key already + if self._connections.get(key) is conn: + del self._connections[key] + return errors - # Call the errback of all the pending requests for this connection - pending_requests = self._pending_requests.pop(key, None) + def _fail_pending_requests( + self, + errors: list[BaseException], + key: ConnectionKeyT, + pending_requests: deque[Deferred[H2ClientProtocol]], + ) -> list[BaseException]: + """Call the errback of the requests that were waiting for this + connection, unless a newer connection is expected to serve them.""" + if self._pending_requests.get(key) is pending_requests: + del self._pending_requests[key] while pending_requests: d = pending_requests.popleft() d.errback(ResponseFailed(errors)) + return errors def close_connections(self) -> None: """Close all the HTTP/2 connections and remove them from pool.""" diff --git a/scrapy/core/http2/protocol.py b/scrapy/core/http2/protocol.py index 7136e829e..9d029312c 100644 --- a/scrapy/core/http2/protocol.py +++ b/scrapy/core/http2/protocol.py @@ -85,8 +85,6 @@ class MethodNotAllowed405(H2Error): @implementer(IHandshakeListener) class H2ClientProtocol(Protocol, TimeoutMixin): - IDLE_TIMEOUT = 240 - def __init__( self, uri: URI, @@ -107,6 +105,8 @@ class H2ClientProtocol(Protocol, TimeoutMixin): """ self._conn_lost_deferred: Deferred[list[BaseException]] = conn_lost_deferred self._tls_verbose_logging: bool = tls_verbose_logging + self._idle_timeout: float = settings.getfloat("CONNECTION_KEEPALIVE_TIMEOUT") + self.closing: bool = False config = H2Configuration(client_side=True, header_encoding="utf-8") self.conn = H2Connection(config=config) @@ -193,6 +193,8 @@ class H2ClientProtocol(Protocol, TimeoutMixin): stream = self.streams.pop(stream_id) self.metadata["active_streams"] -= 1 self._send_pending_requests() + if not self._idle_timeout and not self.metadata["active_streams"]: + self.close_idle() return stream def _new_stream(self, request: Request, spider: Spider) -> Stream: @@ -251,7 +253,7 @@ class H2ClientProtocol(Protocol, TimeoutMixin): sending some data now: we should open with the connection preamble. """ # Initialize the timeout - self.setTimeout(self.IDLE_TIMEOUT) # type: ignore[no-untyped-call] + self.setTimeout(self._idle_timeout or None) # type: ignore[no-untyped-call] assert self.transport is not None # typing destination = self.transport.getPeer() @@ -264,10 +266,17 @@ class H2ClientProtocol(Protocol, TimeoutMixin): def _lose_connection_with_error(self, errors: list[BaseException]) -> None: """Helper function to lose the connection with the error sent as a reason""" + self.closing = True self._conn_lost_errors.extend(errors) assert self.transport is not None # typing self.transport.loseConnection() + def close_idle(self) -> None: + """Close a connection that is not serving any stream.""" + self.conn.close_connection(error_code=ErrorCodes.NO_ERROR) + self._write_to_transport() + self._lose_connection_with_error([]) + def handshakeCompleted(self) -> None: """ Close the connection if it's not made via the expected protocol @@ -342,7 +351,7 @@ class H2ClientProtocol(Protocol, TimeoutMixin): self._lose_connection_with_error( [ DownloadTimeoutError( - f"Connection was IDLE for more than {self.IDLE_TIMEOUT}s" + f"Connection was IDLE for more than {self._idle_timeout}s" ) ] ) diff --git a/scrapy/settings/default_settings.py b/scrapy/settings/default_settings.py index a44b36c8a..bdb9d2968 100644 --- a/scrapy/settings/default_settings.py +++ b/scrapy/settings/default_settings.py @@ -249,11 +249,15 @@ COMMANDS_MODULE = "" COMPRESSION_ENABLED = True +CONCURRENT_CONNECTIONS_PER_HANDLER = None + CONCURRENT_ITEMS = 100 CONCURRENT_REQUESTS = 16 CONCURRENT_REQUESTS_PER_DOMAIN = 8 +CONNECTION_KEEPALIVE_TIMEOUT = 240.0 + COOKIES_ENABLED = True COOKIES_DEBUG = False diff --git a/tests/mockserver/http.py b/tests/mockserver/http.py index c4fd4464e..81e74dce1 100644 --- a/tests/mockserver/http.py +++ b/tests/mockserver/http.py @@ -17,6 +17,7 @@ from .http_resources import ( ChunkedResource, ClientIPResource, Compress, + ConnectionId, ContentLengthHeaderResource, Delay, Drop, @@ -58,6 +59,7 @@ class Root(BaseResource): put_child(self, b"static", File(str(Path(tests_datadir, "test_site/")))) put_child(self, b"redirect-to", RedirectTo()) put_child(self, b"text", Data(b"Works", "text/plain")) + put_child(self, b"connection-id", ConnectionId()) put_child( self, b"html", diff --git a/tests/mockserver/http_resources.py b/tests/mockserver/http_resources.py index cb028bc10..a1c3928d0 100644 --- a/tests/mockserver/http_resources.py +++ b/tests/mockserver/http_resources.py @@ -1,6 +1,7 @@ from __future__ import annotations import gzip +import itertools import json import random from typing import TYPE_CHECKING, Any, ParamSpec, TypeVar @@ -417,3 +418,35 @@ class SetCookie(BaseResource): cookie = (cookie_name.decode() + "=" + cookie_value.decode()).encode() request.setHeader(b"Set-Cookie", cookie) return b"" + + +_connection_ids = itertools.count() + + +class ConnectionId(LeafResource): + """Return the identifier of the connection serving the request, after the + number of seconds given in the ``delay`` argument. + + Two responses carrying the same identifier were served over the same + connection, which lets tests tell connection reuse from reconnection. + """ + + def render_GET(self, request: Request) -> int | bytes: + delay = getarg(request, b"delay", 0, type_=float) + if not delay: + return self._connection_id(request) + self.deferRequest(request, delay, self._delayedRender, request) + return NOT_DONE_YET + + def _delayedRender(self, request: Request) -> None: + request.write(self._connection_id(request)) + request.finish() + + @staticmethod + def _connection_id(request: Request) -> bytes: + # under HTTP/2 the channel of a request is its stream, so the + # connection has to be reached through it + channel: Any = getattr(request.channel, "_conn", request.channel) + if not hasattr(channel, "_mockserver_connection_id"): + channel._mockserver_connection_id = next(_connection_ids) + return to_bytes(str(channel._mockserver_connection_id)) diff --git a/tests/test_downloader_handler_httpx.py b/tests/test_downloader_handler_httpx.py index 976daacaf..25d63281d 100644 --- a/tests/test_downloader_handler_httpx.py +++ b/tests/test_downloader_handler_httpx.py @@ -165,13 +165,23 @@ class TestRealWebsite(HttpxDownloadHandlerMixin, TestRealWebsiteBase): pass -@pytest.mark.parametrize(("concurrency", "expected"), [(16, 16), (0, None)]) +@pytest.mark.parametrize(("limit", "expected"), [(16, 16), (0, None)]) @coroutine_test -async def test_pool_limits(concurrency: int, expected: int | None) -> None: - crawler = get_crawler(settings_dict={"CONCURRENT_REQUESTS": concurrency}) +async def test_pool_limits(limit: int, expected: int | None) -> None: + crawler = get_crawler(settings_dict={"CONCURRENT_CONNECTIONS_PER_HANDLER": limit}) handler = build_from_crawler(HttpxDownloadHandler, crawler) try: assert handler._limits.max_connections == expected assert handler._limits.max_keepalive_connections == expected finally: await handler.close() + + +@coroutine_test +async def test_keepalive_timeout() -> None: + crawler = get_crawler(settings_dict={"CONNECTION_KEEPALIVE_TIMEOUT": 5}) + handler = build_from_crawler(HttpxDownloadHandler, crawler) + try: + assert handler._limits.keepalive_expiry == 5 + finally: + await handler.close() diff --git a/tests/test_downloader_handler_twisted_http11.py b/tests/test_downloader_handler_twisted_http11.py index 32dfd7540..7f6072523 100644 --- a/tests/test_downloader_handler_twisted_http11.py +++ b/tests/test_downloader_handler_twisted_http11.py @@ -3,14 +3,19 @@ from __future__ import annotations import sys +from contextlib import asynccontextmanager from typing import TYPE_CHECKING, Any import pytest from scrapy import Spider +from scrapy.core.downloader.handlers._base_http import _auto_connection_limit from scrapy.core.downloader.handlers.http11 import HTTP11DownloadHandler from scrapy.crawler import Crawler from scrapy.exceptions import NotConfigured +from scrapy.utils.misc import build_from_crawler +from scrapy.utils.spider import DefaultSpider +from scrapy.utils.test import get_crawler from tests.utils.bases.download_handlers_http import ( TestHttpBase, TestHttpProxyBase, @@ -25,8 +30,11 @@ from tests.utils.bases.download_handlers_http import ( TestRealWebsiteBase, TestSimpleHttpsBase, ) +from tests.utils.decorators import coroutine_test if TYPE_CHECKING: + from collections.abc import AsyncGenerator + from scrapy.core.downloader.handlers import DownloadHandlerProtocol @@ -54,6 +62,52 @@ def test_not_configured_without_reactor() -> None: HTTP11DownloadHandler.from_crawler(crawler) +@asynccontextmanager +async def _get_dh( + settings_dict: dict[str, Any], +) -> AsyncGenerator[HTTP11DownloadHandler]: + crawler = get_crawler(DefaultSpider, settings_dict) + crawler.spider = crawler._create_spider() + dh = build_from_crawler(HTTP11DownloadHandler, crawler) + try: + yield dh + finally: + await dh.close() + + +@coroutine_test +async def test_connection_limit_auto() -> None: + async with _get_dh({}) as dh: + assert dh._pool._limit == _auto_connection_limit() + + +@coroutine_test +@pytest.mark.parametrize("limit", [0, 20]) +async def test_connection_limit_explicit(limit: int) -> None: + async with _get_dh({"CONCURRENT_CONNECTIONS_PER_HANDLER": limit}) as dh: + assert dh._pool._limit == limit + + +@coroutine_test +async def test_connection_limit_below_concurrent_requests( + caplog: pytest.LogCaptureFixture, +) -> None: + settings_dict = { + "CONCURRENT_CONNECTIONS_PER_HANDLER": 4, + "CONCURRENT_REQUESTS": 8, + } + with caplog.at_level("WARNING"): + async with _get_dh(settings_dict): + pass + assert "CONCURRENT_CONNECTIONS_PER_HANDLER (4)" in caplog.text + + +@coroutine_test +async def test_keepalive_timeout() -> None: + async with _get_dh({"CONNECTION_KEEPALIVE_TIMEOUT": 5}) as dh: + assert dh._pool.cachedConnectionTimeout == 5 + + class TestHttp(HTTP11DownloadHandlerMixin, TestHttpBase): pass diff --git a/tests/test_http2_client_protocol.py b/tests/test_http2_client_protocol.py index b8586d1ca..ee38ff6de 100644 --- a/tests/test_http2_client_protocol.py +++ b/tests/test_http2_client_protocol.py @@ -737,11 +737,9 @@ class TestHttps2ClientProtocol: yield make_request_dfd(client, request) for err in exc_info.value.reasons: - from scrapy.core.http2.protocol import H2ClientProtocol # noqa: PLC0415 - if isinstance(err, DownloadTimeoutError): assert ( - f"Connection was IDLE for more than {H2ClientProtocol.IDLE_TIMEOUT}s" + f"Connection was IDLE for more than {client._idle_timeout}s" in str(err) ) break diff --git a/tests/utils/bases/download_handlers_http.py b/tests/utils/bases/download_handlers_http.py index e44f9bcb8..bf24da4ed 100644 --- a/tests/utils/bases/download_handlers_http.py +++ b/tests/utils/bases/download_handlers_http.py @@ -16,6 +16,7 @@ from urllib.parse import urlparse import pytest from cryptography.x509 import load_der_x509_certificate +from twisted.internet.defer import DeferredList from twisted.internet.ssl import Certificate from twisted.python.failure import Failure @@ -97,6 +98,60 @@ class TestHttpBase(ABC): finally: await dh.close() + async def _connection_ids( + self, urls: list[str], settings_dict: dict[str, Any] + ) -> set[str]: + """Request *urls* one after another and return the identifiers of the + connections that the server used to serve them.""" + async with self.get_dh(settings_dict) as dh: + responses = [await dh.download_request(Request(url)) for url in urls] + return {response.text for response in responses} + + @staticmethod + def _other_host(url: str) -> str: + """Return *url* pointing at the same server through a different name, + so that its connection cannot be reused.""" + return url.replace("127.0.0.1", "localhost") + + @coroutine_test + async def test_keepalive(self, mockserver: MockServer) -> None: + url = mockserver.url("/connection-id", is_secure=self.is_secure) + assert len(await self._connection_ids([url, url], {})) == 1 + + @coroutine_test + async def test_no_keepalive(self, mockserver: MockServer) -> None: + url = mockserver.url("/connection-id", is_secure=self.is_secure) + settings_dict = {"CONNECTION_KEEPALIVE_TIMEOUT": 0} + assert len(await self._connection_ids([url, url], settings_dict)) == 2 + + @coroutine_test + async def test_connection_limit(self, mockserver: MockServer) -> None: + url = mockserver.url("/connection-id", is_secure=self.is_secure) + urls = [url, self._other_host(url)] * 2 + + assert len(await self._connection_ids(urls, {})) == 2 + + settings_dict = {"CONCURRENT_CONNECTIONS_PER_HANDLER": 1} + assert len(await self._connection_ids(urls, settings_dict)) == 4 + + @coroutine_test + async def test_connection_limit_spares_connections_in_use( + self, mockserver: MockServer + ) -> None: + url = mockserver.url("/connection-id?delay=0.5", is_secure=self.is_secure) + urls = [url, self._other_host(url)] + async with self.get_dh({"CONCURRENT_CONNECTIONS_PER_HANDLER": 1}) as dh: + results = await maybe_deferred_to_future( + DeferredList( + [ + deferred_from_coro(dh.download_request(Request(url))) + for url in urls + ], + fireOnOneErrback=True, + ) + ) + assert len({response.text for _, response in results}) == 2 + @coroutine_test async def test_unsupported_scheme(self) -> None: request = Request("unsupp://unsupported.scheme")