RomanNum3ral
/
scrapy
mirror of https://github.com/scrapy/scrapy.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Add CONCURRENT_CONNECTIONS_PER_HANDLER and CONNECTION_KEEPALIVE_TIMEOUT

This commit is contained in:
Adrian Chaves 2026-08-07 19:06:39 +02:00
parent 1bd839b57d
commit 1b937b32a2
17 changed files with 464 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
docs/faq.rst
View File

@ -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 <topics-download-handlers>` 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: <filename>``
--------------------------------------------------------------------------

4
docs/topics/broad-crawls.rst
View File

@ -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
============================================

53
docs/topics/settings.rst
View File

@ -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
<topics-download-handlers>` 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 <https://github.com/pydantic/httpx2/issues/818>`_.
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

51
scrapy/core/downloader/handlers/_base_http.py
View File

@ -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"
)

5
scrapy/core/downloader/handlers/_base_streaming.py
View File

@ -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

3
scrapy/core/downloader/handlers/_httpx.py
View File

@ -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()

104
scrapy/core/downloader/handlers/http11.py
View File

@ -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.
#

6
scrapy/core/downloader/handlers/http2.py
View File

@ -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")

75
scrapy/core/http2/agent.py
View File

@ -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."""

17
scrapy/core/http2/protocol.py
View File

@ -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"
)
]
)

4
scrapy/settings/default_settings.py
View File

@ -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

2
tests/mockserver/http.py
View File

@ -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",

33
tests/mockserver/http_resources.py
View File

@ -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))

16
tests/test_downloader_handler_httpx.py
View File

@ -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()

54
tests/test_downloader_handler_twisted_http11.py
View File

@ -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

4
tests/test_http2_client_protocol.py
View File

@ -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

55
tests/utils/bases/download_handlers_http.py
View File

@ -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")