WIP

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 <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 <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 <topics-scheduler>`
-    remain in the scheduler. However, requests sent with :meth:`engine.download()
-    <scrapy.core.engine.ExecutionEngine.download>` bypass the scheduler,
-    including requests sent by some built-in :ref:`components
-    <topics-components>` and :ref:`inline requests <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``)

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

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

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
 

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

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

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(

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