diff --git a/docs/topics/throttling.rst b/docs/topics/throttling.rst index bfa126fc8..b231e8586 100644 --- a/docs/topics/throttling.rst +++ b/docs/topics/throttling.rst @@ -617,8 +617,8 @@ By default, throttling is enforced at the engine, where a request waiting on its :ref:`throttling scopes ` holds a concurrency slot. In a crawl that mixes heavily-throttled scopes with unthrottled ones, this can let throttled requests starve unthrottled ones that could be sent right away -(**head-of-line blocking**; Scrapy logs a warning, see -:setting:`DELAYED_REQUESTS_WARN_THRESHOLD`). +(**head-of-line blocking**; Scrapy logs a warning the first time throttled +requests start consuming the global concurrency budget while they wait). :class:`~scrapy.core.scheduler.ThrottlingAwareScheduler` avoids this. To enable it: @@ -992,23 +992,6 @@ Additional settings by one :setting:`BACKOFF_DELAY_FACTOR` step toward the configured value. A new trigger resets the countdown. -- .. setting:: DELAYED_REQUESTS_WARN_THRESHOLD - - :setting:`DELAYED_REQUESTS_WARN_THRESHOLD` (default: ``500``) - - Number of requests held back by :ref:`throttling ` at which - Scrapy logs a warning, to help detect throttling configurations that hold - back more requests than expected. - - While throttled, requests in the :ref:`scheduler ` - remain in the scheduler. However, requests sent with :meth:`engine.download() - ` bypass the scheduler, - including requests sent by some built-in :ref:`components - ` and :ref:`inline requests `. When - such requests are throttled, they are paused and kept in memory, along - with any run time context from the code that is sending them. If they - accumulate, they can become a memory issue. - - .. setting:: RANDOMIZE_DOWNLOAD_DELAY :setting:`RANDOMIZE_DOWNLOAD_DELAY` (default: ``True``) diff --git a/scrapy/core/downloader/handlers/_base_http.py b/scrapy/core/downloader/handlers/_base_http.py index 83d130462..2ca8ff353 100644 --- a/scrapy/core/downloader/handlers/_base_http.py +++ b/scrapy/core/downloader/handlers/_base_http.py @@ -7,11 +7,36 @@ from .base import BaseDownloadHandler if TYPE_CHECKING: from scrapy.crawler import Crawler + from scrapy.settings import BaseSettings class BaseHttpDownloadHandler(BaseDownloadHandler, ABC): """Base class for built-in HTTP download handlers.""" + @staticmethod + def _max_per_host_concurrency(settings: BaseSettings) -> int: + """Highest per-host concurrency the throttler may admit: the per-domain + limit, the default ``other``-scope limit, and any explicit + :setting:`THROTTLING_SCOPES` concurrency. + + A scope with :ref:`rampup ` enabled has no configured + concurrency ceiling; it grows toward :setting:`CONCURRENT_REQUESTS`, so + it counts as that. And since :setting:`CONCURRENT_REQUESTS` caps the + total number of requests in flight, no host can ever exceed it, so it is + also the upper bound of the result. + """ + global_concurrency = settings.getint("CONCURRENT_REQUESTS") + candidates = [ + settings.getint("CONCURRENT_REQUESTS_PER_DOMAIN"), + settings.getint("THROTTLING_SCOPE_CONCURRENCY"), + ] + for scope in settings.getdict("THROTTLING_SCOPES").values(): + if scope.get("rampup"): + candidates.append(global_concurrency) + elif "concurrency" in scope: + candidates.append(int(scope["concurrency"])) + return min(max(candidates), global_concurrency) + def __init__(self, crawler: Crawler): super().__init__(crawler) self._default_maxsize: int = crawler.settings.getint("DOWNLOAD_MAXSIZE") diff --git a/scrapy/core/downloader/handlers/_base_streaming.py b/scrapy/core/downloader/handlers/_base_streaming.py index 6742781e6..d74a7d120 100644 --- a/scrapy/core/downloader/handlers/_base_streaming.py +++ b/scrapy/core/downloader/handlers/_base_streaming.py @@ -85,6 +85,7 @@ class BaseStreamingDownloadHandler(BaseHttpDownloadHandler, ABC, Generic[_Respon ) self._proxy_auth_encoding: str = crawler.settings.get("HTTPPROXY_AUTH_ENCODING") self._pool_size_total: int = crawler.settings.getint("CONCURRENT_REQUESTS") + self._pool_size_per_host: int = self._max_per_host_concurrency(crawler.settings) @staticmethod @abstractmethod diff --git a/scrapy/core/downloader/handlers/http11.py b/scrapy/core/downloader/handlers/http11.py index 49422b0bf..cb46f2103 100644 --- a/scrapy/core/downloader/handlers/http11.py +++ b/scrapy/core/downloader/handlers/http11.py @@ -92,22 +92,8 @@ class HTTP11DownloadHandler(BaseHttpDownloadHandler): from twisted.internet import reactor self._pool: HTTPConnectionPool = HTTPConnectionPool(reactor, persistent=True) - # Keep enough persistent connections per host to match the highest - # per-host concurrency the throttler may admit: the per-domain limit, - # the default "other"-scope limit, and any explicit THROTTLING_SCOPES - # concurrency. Per-scope concurrency can grow beyond this at runtime - # (rampup); the excess simply uses non-persistent connections. - scope_concurrencies = [ - scope["concurrency"] - for scope in crawler.settings.getdict("THROTTLING_SCOPES").values() - if "concurrency" in scope - ] - self._pool.maxPersistentPerHost = max( - [ - crawler.settings.getint("CONCURRENT_REQUESTS_PER_DOMAIN"), - crawler.settings.getint("THROTTLING_SCOPE_CONCURRENCY"), - *scope_concurrencies, - ] + self._pool.maxPersistentPerHost = self._max_per_host_concurrency( + crawler.settings ) self._pool._factory.noisy = False diff --git a/scrapy/core/engine.py b/scrapy/core/engine.py index 2469866b9..217d52826 100644 --- a/scrapy/core/engine.py +++ b/scrapy/core/engine.py @@ -138,14 +138,18 @@ class ExecutionEngine: # ``_enqueue_request_async``), so the spider is not considered idle # while a request is still on its way into the scheduler. self._scheduling: int = 0 - # A coalesced wakeup timer, armed when a throttling-aware scheduler - # reports that all pending requests are time-blocked (see - # ``get_next_request_delay``). - self._throttling_wakeup: CallLaterResult | None = None - self._delayed_requests_warn_threshold: int = self.settings.getint( - "DELAYED_REQUESTS_WARN_THRESHOLD" - ) - self._delayed_requests_warned: bool = False + # A coalesced wakeup timer, armed when the scheduler reports (through + # ``get_next_request_delay``) that every pending request is time-blocked. + self._delay_wakeup: CallLaterResult | None = None + # The scheduler's ``get_next_request_delay`` bound method, cached once + # per crawl in ``open_spider_async`` (``None`` if the scheduler does not + # expose one). Used both to arm the wakeup timer and to tell whether the + # scheduler holds time-blocked requests itself. + self._get_next_request_delay: Callable[[], float | None] | None = None + # Whether the scheduler exposes enqueue_request_async, cached once per + # crawl in ``open_spider_async``; checked for every scheduled request. + self._scheduler_enqueues_async: bool = False + self._throttling_backout_warned: bool = False downloader_cls: type[Downloader] = load_object(self.settings["DOWNLOADER"]) try: self.scheduler_cls: type[BaseScheduler] = self._get_scheduler_class( @@ -279,14 +283,14 @@ class ExecutionEngine: def pause(self) -> None: self.paused = True - self._cancel_throttling_wakeup() + self._cancel_delay_wakeup() def unpause(self) -> None: self.paused = False - # pause() cancels the coalesced throttling wakeup and the loop stops + # pause() cancels the coalesced delay wakeup and the loop stops # re-running itself while paused, so re-run it here: this re-arms the - # wakeup and keeps a crawl held only by a time-based throttling gate - # (nothing in flight to re-run the loop from _download) from stalling. + # wakeup and keeps a crawl held only by a time-based gate (nothing in + # flight to re-run the loop from _download) from stalling. if self._slot is not None: self._slot.nextcall.schedule() @@ -355,25 +359,25 @@ class ExecutionEngine: if self._slot is None or self._slot.closing is not None or self.paused: return - self._cancel_throttling_wakeup() + self._cancel_delay_wakeup() while not self.needs_backout(): if not self._start_scheduled_request(): break - self._maybe_arm_throttling_wakeup() + self._maybe_arm_delay_wakeup() if self.spider_is_idle() and self._slot.close_if_idle: self._spider_idle() - def _cancel_throttling_wakeup(self) -> None: - if self._throttling_wakeup is not None: - self._throttling_wakeup.cancel() - self._throttling_wakeup = None + def _cancel_delay_wakeup(self) -> None: + if self._delay_wakeup is not None: + self._delay_wakeup.cancel() + self._delay_wakeup = None - def _maybe_arm_throttling_wakeup(self) -> None: - """Arm a single coalesced wakeup when the scheduler is throttling-aware - and reports that every pending request is time-blocked. + def _maybe_arm_delay_wakeup(self) -> None: + """Arm a single coalesced wakeup when the scheduler reports that every + pending request is time-blocked. Concurrency-blocked states do not need this: a freed slot already re-runs the loop from :meth:`_download`'s ``finally``. Only time-based @@ -381,21 +385,19 @@ class ExecutionEngine: would re-run the loop while they are closed. """ assert self._slot is not None - scheduler = self._slot.scheduler - delay_fn = getattr(scheduler, "get_next_request_delay", None) - if delay_fn is None or not scheduler.has_pending_requests(): + if ( + self._get_next_request_delay is None + or not self._slot.scheduler.has_pending_requests() + ): return - delay = delay_fn() - # ``delay`` is ``None`` when no pending request is time-blocked, and - # ``0`` when some request is ready right now but could not be sent (e.g. - # the downloader is at capacity). Neither case needs a timer: a freed - # slot already re-runs the loop from :meth:`_download`'s ``finally``, - # and arming a ``0``-second timer here would busy-loop the engine while - # the downloader stays full. Only a positive delay, i.e. a time-based - # gate that nothing else would wake us for, needs one. + # A positive delay means a time-based gate nothing else would wake us + # for. ``None`` (nothing time-blocked) and ``0`` (something is ready but + # could not be sent, e.g. the downloader is full) are handled elsewhere: + # a freed slot re-runs the loop, and a ``0``-second timer would busy-loop. + delay = self._get_next_request_delay() if delay is None or delay <= 0: return - self._throttling_wakeup = call_later(delay, self._slot.nextcall.schedule) + self._delay_wakeup = call_later(delay, self._slot.nextcall.schedule) def needs_backout(self) -> bool: """Returns ``True`` if no more requests can be sent at the moment, or @@ -423,29 +425,28 @@ class ExecutionEngine: """ if not self._throttling_waiting: return False - return ( + if ( len(self._throttling_waiting) + len(self.downloader.active) - >= self.downloader.total_concurrency - ) + < self.downloader.total_concurrency + ): + return False + self._maybe_warn_throttling_backout() + return True - def _maybe_warn_delayed_requests(self) -> None: - if self._delayed_requests_warned: + def _maybe_warn_throttling_backout(self) -> None: + if self._throttling_backout_warned: return - if len(self._throttling_waiting) < self._delayed_requests_warn_threshold: + # A throttling-aware scheduler holds time-blocked requests itself instead + # of letting them pile up in _throttling_waiting and back-pressure the + # engine here, so the warning does not apply when one is in use. + if self._get_next_request_delay is not None: return - self._delayed_requests_warned = True - recommendation = "" - # A throttling-aware scheduler holds throttled requests in the - # scheduler instead of in _throttling_waiting, so it does not hit this - # path; recommend it only when it is not already in use. - scheduler = self._slot.scheduler if self._slot is not None else None - if scheduler is None or not hasattr(scheduler, "get_next_request_delay"): - recommendation = ( - " Consider switching to scrapy.core.scheduler.ThrottlingAwareScheduler." - ) + self._throttling_backout_warned = True logger.warning( - f"There are {len(self._throttling_waiting)} requests held back by " - f"throttling. See DELAYED_REQUESTS_WARN_THRESHOLD.{recommendation}", + "Throttling is holding requests back and they are now consuming the " + "global concurrency budget while they wait for a free slot. Consider " + "switching to scrapy.core.scheduler.ThrottlingAwareScheduler, which " + "holds throttled requests without consuming concurrency slots.", extra={"spider": self.spider}, ) @@ -549,7 +550,7 @@ class ExecutionEngine: if isinstance(result, Failure) and isinstance(result.value, IgnoreRequest): return scheduler = self._slot.scheduler - if hasattr(scheduler, "enqueue_request_async"): + if self._scheduler_enqueues_async: self._scheduling += 1 _schedule_coro(self._enqueue_request_async(request)) return @@ -559,10 +560,9 @@ class ExecutionEngine: ) async def _enqueue_request_async(self, request: Request) -> None: - # The counter is incremented in _schedule_request before this coroutine - # is scheduled, so it must be decremented here even on an early exit, - # otherwise spider_is_idle() never reports idle. The slot can be torn - # down (spider stopping) between the increment and this running. + # _scheduling is incremented in _schedule_request before this coroutine + # is scheduled, so it must be decremented on every path (hence finally), + # otherwise spider_is_idle() never reports idle. try: if self._slot is None: return @@ -618,7 +618,6 @@ class ExecutionEngine: """Wait at the throttling gate before *request* is sent, tracking it as held meanwhile.""" self._throttling_waiting.add(request) - self._maybe_warn_delayed_requests() throttler = self.crawler.throttler assert throttler is not None try: @@ -689,6 +688,10 @@ class ExecutionEngine: self.spider = self.crawler.spider nextcall = CallLaterOnce(self._start_scheduled_requests) scheduler = build_from_crawler(self.scheduler_cls, self.crawler) + self._get_next_request_delay = getattr( + scheduler, "get_next_request_delay", None + ) + self._scheduler_enqueues_async = hasattr(scheduler, "enqueue_request_async") self._slot = _Slot(close_if_idle, nextcall, scheduler) self._start = await self.scraper.spidermw.process_start() if hasattr(scheduler, "open") and (d := scheduler.open(self.crawler.spider)): @@ -765,7 +768,7 @@ class ExecutionEngine: "Closing spider (%(reason)s)", {"reason": reason}, extra={"spider": spider} ) - self._cancel_throttling_wakeup() + self._cancel_delay_wakeup() try: await self._slot.close() diff --git a/scrapy/settings/default_settings.py b/scrapy/settings/default_settings.py index b9b9d19f3..cc145d241 100644 --- a/scrapy/settings/default_settings.py +++ b/scrapy/settings/default_settings.py @@ -58,7 +58,6 @@ __all__ = [ "DEFAULT_DROPITEM_LOG_LEVEL", "DEFAULT_ITEM_CLASS", "DEFAULT_REQUEST_HEADERS", - "DELAYED_REQUESTS_WARN_THRESHOLD", "DEPTH_LIMIT", "DEPTH_PRIORITY", "DEPTH_STATS_VERBOSE", @@ -300,8 +299,6 @@ DEFAULT_REQUEST_HEADERS = { "Accept-Language": "en", } -DELAYED_REQUESTS_WARN_THRESHOLD = 500 - DEPTH_LIMIT = 0 DEPTH_PRIORITY = 0 DEPTH_STATS_VERBOSE = False diff --git a/scrapy/throttling.py b/scrapy/throttling.py index 5c77dad6c..646506f47 100644 --- a/scrapy/throttling.py +++ b/scrapy/throttling.py @@ -207,6 +207,36 @@ def _warn_on_deprecated_concurrency(settings: BaseSettings) -> None: ) +def _warn_on_unachievable_concurrency(settings: BaseSettings) -> None: + """Warn about configured concurrency limits that exceed + :setting:`CONCURRENT_REQUESTS`. Call once per crawl (see + :meth:`ThrottlingManager.__init__`). + + :setting:`CONCURRENT_REQUESTS` caps the total number of requests in flight, + so a per-scope (or per-domain) concurrency limit above it can never be + reached. Rampup is not flagged: it has no configured ceiling and simply + grows toward :setting:`CONCURRENT_REQUESTS`. + """ + global_concurrency = settings.getint("CONCURRENT_REQUESTS") + offenders: list[str] = [ + f"{name}={settings.getint(name)}" + for name in ("CONCURRENT_REQUESTS_PER_DOMAIN", "THROTTLING_SCOPE_CONCURRENCY") + if settings.getint(name) > global_concurrency + ] + offenders += [ + f"THROTTLING_SCOPES[{scope_id!r}]['concurrency']={config['concurrency']}" + for scope_id, config in settings.getdict("THROTTLING_SCOPES").items() + if config.get("concurrency") is not None + and int(config["concurrency"]) > global_concurrency + ] + if offenders: + logger.warning( + f"The following concurrency settings exceed CONCURRENT_REQUESTS " + f"({global_concurrency}), which caps the total number of requests in " + f"flight, so they cannot be reached: {', '.join(offenders)}." + ) + + def _to_scope_dict(scopes: RequestScopes) -> dict[ScopeID, float | None]: """Normalize *scopes* (``None``, a scope id, an iterable of scope ids or a ``{scope_id: quota}`` dict) into a ``{scope_id: quota}`` dict, using ``None`` @@ -505,6 +535,7 @@ class ThrottlingManager: def __init__(self, crawler: Crawler) -> None: self.crawler = crawler _warn_on_deprecated_concurrency(crawler.settings) + _warn_on_unachievable_concurrency(crawler.settings) self._debug = crawler.settings.getbool("THROTTLING_DEBUG") self._max_idle = crawler.settings.getfloat("THROTTLING_SCOPE_MAX_IDLE") self._robotstxt_obey = crawler.settings.getbool( diff --git a/tests/test_engine.py b/tests/test_engine.py index a7f81e19d..4bca525c5 100644 --- a/tests/test_engine.py +++ b/tests/test_engine.py @@ -663,74 +663,69 @@ class TestEngineThrottling: yield engine engine.downloader.close() - def test_pause_cancels_throttling_wakeup(self, engine): + def test_pause_cancels_delay_wakeup(self, engine): wakeup = Mock() - engine._throttling_wakeup = wakeup + engine._delay_wakeup = wakeup engine.pause() assert engine.paused is True wakeup.cancel.assert_called_once_with() - assert engine._throttling_wakeup is None + assert engine._delay_wakeup is None engine.unpause() assert engine.paused is False @pytest.mark.requires_reactor # call_later() needs a reactor or asyncio loop - def test_maybe_arm_throttling_wakeup_arms_timer(self, engine): - scheduler = Mock() - scheduler.has_pending_requests.return_value = True - scheduler.get_next_request_delay.return_value = 5.0 + def test_maybe_arm_delay_wakeup_arms_timer(self, engine): + engine._get_next_request_delay = lambda: 5.0 engine._slot = Mock() - engine._slot.scheduler = scheduler - engine._maybe_arm_throttling_wakeup() - assert engine._throttling_wakeup is not None + engine._slot.scheduler.has_pending_requests.return_value = True + engine._maybe_arm_delay_wakeup() + assert engine._delay_wakeup is not None # Cancel the scheduled reactor call so it does not leak into other tests. - engine._cancel_throttling_wakeup() + engine._cancel_delay_wakeup() - def test_maybe_arm_throttling_wakeup_no_delay(self, engine): - scheduler = Mock() - scheduler.has_pending_requests.return_value = True - scheduler.get_next_request_delay.return_value = None + def test_maybe_arm_delay_wakeup_no_delay(self, engine): + engine._get_next_request_delay = lambda: None engine._slot = Mock() - engine._slot.scheduler = scheduler - engine._maybe_arm_throttling_wakeup() - assert engine._throttling_wakeup is None + engine._slot.scheduler.has_pending_requests.return_value = True + engine._maybe_arm_delay_wakeup() + assert engine._delay_wakeup is None - def test_maybe_arm_throttling_wakeup_zero_delay(self, engine): + def test_maybe_arm_delay_wakeup_zero_delay(self, engine): # A 0 delay means a request is ready but could not be sent (e.g. the # downloader is at capacity); arming a 0-second timer would busy-loop # the engine, so no timer must be armed. - scheduler = Mock() - scheduler.has_pending_requests.return_value = True - scheduler.get_next_request_delay.return_value = 0.0 + engine._get_next_request_delay = lambda: 0.0 engine._slot = Mock() - engine._slot.scheduler = scheduler - engine._maybe_arm_throttling_wakeup() - assert engine._throttling_wakeup is None + engine._slot.scheduler.has_pending_requests.return_value = True + engine._maybe_arm_delay_wakeup() + assert engine._delay_wakeup is None - def test_warn_delayed_requests(self, engine): - engine._delayed_requests_warn_threshold = 1 - engine._throttling_waiting = {Request("http://a.example")} + def test_maybe_arm_delay_wakeup_not_supported(self, engine): + # A scheduler without get_next_request_delay never arms a timer. + engine._get_next_request_delay = None engine._slot = Mock() - # A scheduler without get_next_request_delay is not throttling-aware, so the - # warning recommends switching to one. - engine._slot.scheduler = Mock(spec=BaseScheduler) + engine._slot.scheduler.has_pending_requests.return_value = True + engine._maybe_arm_delay_wakeup() + assert engine._delay_wakeup is None + + def test_maybe_warn_throttling_backout(self, engine): + # A scheduler without get_next_request_delay is not throttling-aware, so + # the warning recommends switching to one. + engine._get_next_request_delay = None with LogCapture() as log: - engine._maybe_warn_delayed_requests() + engine._maybe_warn_throttling_backout() # A second call is a no-op (the warning is emitted only once). - engine._maybe_warn_delayed_requests() - assert engine._delayed_requests_warned is True - log_text = str(log) - assert "requests held back by throttling" in log_text - assert log_text.count("ThrottlingAwareScheduler") == 1 + engine._maybe_warn_throttling_backout() + assert engine._throttling_backout_warned is True + assert str(log).count("ThrottlingAwareScheduler") == 1 - def test_warn_delayed_requests_throttling_aware(self, engine): - engine._delayed_requests_warn_threshold = 1 - engine._throttling_waiting = {Request("http://a.example")} - engine._slot = Mock() + def test_maybe_warn_throttling_backout_throttling_aware(self, engine): # A throttling-aware scheduler (one with get_next_request_delay) holds - # throttled requests itself, so no switch is recommended. - engine._slot.scheduler = Mock() + # throttled requests itself, so no warning is emitted. + engine._get_next_request_delay = lambda: None with LogCapture() as log: - engine._maybe_warn_delayed_requests() + engine._maybe_warn_throttling_backout() + assert engine._throttling_backout_warned is False assert "ThrottlingAwareScheduler" not in str(log) def test_spider_is_idle_false_while_scheduling(self, engine):