diff --git a/docs/topics/throttling.rst b/docs/topics/throttling.rst index 436d1e336..cc8519990 100644 --- a/docs/topics/throttling.rst +++ b/docs/topics/throttling.rst @@ -459,8 +459,7 @@ path as a string): THROTTLING_MANAGER = "myproject.throttling.MyThrottlingManager" - -.. _multi-throttling-scopes: +.. _multiple-throttling-scopes: Handling of multiple throttling scopes -------------------------------------- @@ -468,7 +467,6 @@ Handling of multiple throttling scopes When a request has multiple throttling scopes, it is not sent until all of its throttling scopes allow it. - .. _throttling-quotas: Throttling quotas @@ -505,7 +503,6 @@ Everything else being equal, :class:`~scrapy.pqueues.ScrapyPriorityQueue` will prioritize requests that consume a higher portion of the available throttling quota, to minimize the risk of those requests getting stuck. - .. _custom-throttling-scope-managers: Customizing throttling scope managers @@ -603,6 +600,28 @@ exponential backoff, similar to a fixed-window rate limiter: def is_idle(self, now, max_idle): return self.active == 0 +.. _throttling-aware-scheduler: + +Throttling-aware scheduling +=========================== + +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`). + +:class:`~scrapy.core.scheduler.ThrottlingAwareScheduler` avoids this. To enable +it: + +.. code-block:: python + :caption: ``settings.py`` + + SCHEDULER = "scrapy.core.scheduler.ThrottlingAwareScheduler" + SCHEDULER_PRIORITY_QUEUE = "scrapy.pqueues.ThrottlingAwarePriorityQueue" + +.. autoclass:: scrapy.core.scheduler.ThrottlingAwareScheduler .. _throttling-examples: @@ -822,8 +841,8 @@ window (:setting:`BACKOFF_WINDOW`). You can use :ref:`throttling quotas - Implement a :ref:`throttling manager ` that: - - Sets a ``cost`` throttling scope on each request to some estimation based - e.g. on request URL parameters: + - Sets a ``cost`` throttling scope on each request to some estimation + based e.g. on request URL parameters: .. code-block:: python @@ -871,6 +890,35 @@ window (:setting:`BACKOFF_WINDOW`). You can use :ref:`throttling quotas This will allow you to spend up to 100.0 units of cost per time window (default: 60 seconds) before throttling kicks in. +.. _throttling-per-ip: + +Per-IP concurrency limiting +--------------------------- + +A concurrency limit keyed by IP is just a throttling scope whose id is the +request's IP, with a ``concurrency`` limit. A request then carries two scopes, +its domain and its IP, and is only sent when **both** allow it (see +:ref:`multiple-throttling-scopes`). + +- Implement a :ref:`throttling manager ` that adds + the request's IP as a second scope: + + .. code-block:: python + + import socket + + from scrapy.throttling import ThrottlingManager, add_scope, scope_cache + from scrapy.utils.asyncio import run_in_thread + from scrapy.utils.httpobj import urlparse_cached + + + class IPThrottlingManager(ThrottlingManager): + @scope_cache + async def get_scopes(self, request): + scopes = await super().get_scopes(request) + host = urlparse_cached(request).hostname + address = await run_in_thread(socket.gethostbyname, host) + return add_scope(scopes, address) .. _throttling-settings: @@ -1024,6 +1072,8 @@ API .. autoclass:: scrapy.throttling.ThrottlingScopeManager +.. autoclass:: scrapy.pqueues.ThrottlingAwarePriorityQueue + .. autoclass:: scrapy.throttling.ThrottlingScopeConfig .. autoclass:: scrapy.throttling.BackoffConfig diff --git a/scrapy/core/engine.py b/scrapy/core/engine.py index 9096cff5a..7b761d80b 100644 --- a/scrapy/core/engine.py +++ b/scrapy/core/engine.py @@ -31,6 +31,7 @@ from scrapy.exceptions import ( from scrapy.http import Request, Response from scrapy.utils.asyncio import ( AsyncioLoopingCall, + call_later, create_looping_call, is_asyncio_available, ) @@ -57,6 +58,7 @@ if TYPE_CHECKING: from scrapy.settings import BaseSettings, Settings from scrapy.signalmanager import SignalManager from scrapy.spiders import Spider + from scrapy.utils.asyncio import CallLaterResult logger = logging.getLogger(__name__) @@ -132,6 +134,14 @@ class ExecutionEngine: # Requests currently held by the throttling manager, waiting for their # scopes to allow them through to the downloader. self._throttling_waiting: set[Request] = set() + # Number of in-flight asynchronous enqueue operations (see + # ``_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 + # ``next_request_delay``). + self._throttling_wakeup: CallLaterResult | None = None self._delayed_requests_warn_threshold: int = self.settings.getint( "DELAYED_REQUESTS_WARN_THRESHOLD" ) @@ -269,6 +279,7 @@ class ExecutionEngine: def pause(self) -> None: self.paused = True + self._cancel_throttling_wakeup() def unpause(self) -> None: self.paused = False @@ -338,13 +349,43 @@ class ExecutionEngine: if self._slot is None or self._slot.closing is not None or self.paused: return + self._cancel_throttling_wakeup() + while not self.needs_backout(): if not self._start_scheduled_request(): break + self._maybe_arm_throttling_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 _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. + + Concurrency-blocked states do not need this: a freed slot already + re-runs the loop from :meth:`_download`'s ``finally``. Only time-based + gates (delay, backoff, quota windows) need a timer, since nothing else + would re-run the loop while they are closed. + """ + assert self._slot is not None + scheduler = self._slot.scheduler + delay_fn = getattr(scheduler, "next_request_delay", None) + if delay_fn is None or not scheduler.has_pending_requests(): + return + delay = delay_fn() + if delay is None: + return + self._throttling_wakeup = call_later( + max(0.0, delay), self._slot.nextcall.schedule + ) + def needs_backout(self) -> bool: """Returns ``True`` if no more requests can be sent at the moment, or ``False`` otherwise. @@ -382,10 +423,18 @@ class ExecutionEngine: if len(self._throttling_waiting) < self._delayed_requests_warn_threshold: 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, "next_request_delay"): + recommendation = ( + " Consider switching to scrapy.core.scheduler.ThrottlingAwareScheduler." + ) logger.warning( - f"There are {len(self._throttling_waiting)} requests held back by throttling. This may " - "indicate a throttling configuration that is too aggressive or a " - "site that is rate-limiting heavily. See DELAYED_REQUESTS_WARN_THRESHOLD.", + f"There are {len(self._throttling_waiting)} requests held back by " + f"throttling. See DELAYED_REQUESTS_WARN_THRESHOLD.{recommendation}", extra={"spider": self.spider}, ) @@ -464,6 +513,8 @@ class ExecutionEngine: return False if self._throttling_waiting: # requests held by the throttling manager return False + if self._scheduling: # requests still on their way into the scheduler + return False if self._start is not None: # not all start requests are handled return False return not self._slot.scheduler.has_pending_requests() @@ -476,6 +527,7 @@ class ExecutionEngine: self._slot.nextcall.schedule() # type: ignore[union-attr] def _schedule_request(self, request: Request) -> None: + assert self._slot is not None # typing request_scheduled_result = self.signals.send_catch_log( signals.request_scheduled, request=request, @@ -485,11 +537,35 @@ class ExecutionEngine: for _, result in request_scheduled_result: if isinstance(result, Failure) and isinstance(result.value, IgnoreRequest): return - if not self._slot.scheduler.enqueue_request(request): # type: ignore[union-attr] + scheduler = self._slot.scheduler + if hasattr(scheduler, "enqueue_request_async"): + self._scheduling += 1 + _schedule_coro(self._enqueue_request_async(request)) + return + if not scheduler.enqueue_request(request): self.signals.send_catch_log( signals.request_dropped, request=request, spider=self.spider ) + async def _enqueue_request_async(self, request: Request) -> None: + assert self._slot is not None + try: + stored = await self._slot.scheduler.enqueue_request_async(request) # type: ignore[attr-defined] + if not stored: + self.signals.send_catch_log( + signals.request_dropped, request=request, spider=self.spider + ) + except Exception: + logger.error( + "Error while enqueuing request", + exc_info=True, + extra={"spider": self.spider}, + ) + finally: + self._scheduling -= 1 + if self._slot is not None: + self._slot.nextcall.schedule() + def download(self, request: Request) -> Deferred[Response]: """Return a Deferred which fires with a Response as result, only downloader middlewares are applied""" warnings.warn( @@ -675,6 +751,8 @@ class ExecutionEngine: "Closing spider (%(reason)s)", {"reason": reason}, extra={"spider": spider} ) + self._cancel_throttling_wakeup() + try: await self._slot.close() except Exception: diff --git a/scrapy/core/scheduler.py b/scrapy/core/scheduler.py index 7217da942..add1891db 100644 --- a/scrapy/core/scheduler.py +++ b/scrapy/core/scheduler.py @@ -10,6 +10,7 @@ from typing import TYPE_CHECKING, Any, cast from twisted.internet.defer import Deferred # noqa: TC002 from scrapy.spiders import Spider # noqa: TC001 +from scrapy.throttling import iter_scopes from scrapy.utils.job import job_dir from scrapy.utils.misc import build_from_crawler, load_object @@ -25,6 +26,7 @@ if TYPE_CHECKING: from scrapy.http.request import Request from scrapy.pqueues import ScrapyPriorityQueue from scrapy.statscollectors import StatsCollector + from scrapy.throttling import ThrottlingManagerProtocol logger = logging.getLogger(__name__) @@ -496,3 +498,109 @@ class Scheduler(BaseScheduler): def _write_dqs_state(self, dqdir: str, state: Any) -> None: with Path(dqdir, "active.json").open("w", encoding="utf-8") as f: json.dump(state, f) + + +class ThrottlingAwareScheduler(Scheduler): + """A :class:`Scheduler` that only ever hands the engine requests whose + :ref:`throttling scopes ` allow them to be sent **right + now**. + + This avoids the head-of-line blocking that the default scheduler can suffer + when a crawl mixes heavily-throttled scopes with unthrottled ones: with the + default scheduler, requests taken from the scheduler in order wait at the + throttling gate + (:meth:`~scrapy.throttling.ThrottlingManagerProtocol.acquire`) while + holding a concurrency slot, so enough throttled requests waiting on a clock + can fill :setting:`CONCURRENT_REQUESTS` and starve unthrottled requests + that could be sent right away. Requests held back by this scheduler instead + occupy neither concurrency slots nor memory (beyond what is needed to track + each distinct scope set), so they cannot starve other scopes. + + When several requests could be sent at the same time, the one with the + highest request :attr:`~scrapy.Request.priority` is sent first; ties are + broken by preferring the least-busy scopes. + + It requires :setting:`SCHEDULER_PRIORITY_QUEUE` to be set to + :class:`~scrapy.pqueues.ThrottlingAwarePriorityQueue` (or a compatible + subclass), and the configured :setting:`SCHEDULER_DISK_QUEUE` / + :setting:`SCHEDULER_MEMORY_QUEUE` to support ``peek``. + """ + + def open(self, spider: Spider) -> Deferred[None] | None: + result = super().open(spider) + if not hasattr(self.mqs, "next_request_delay"): + raise ValueError( + f"{type(self).__name__} requires SCHEDULER_PRIORITY_QUEUE to be " + f"set to a throttling-aware priority queue such as " + f"scrapy.pqueues.ThrottlingAwarePriorityQueue, but the " + f"configured one ({type(self.mqs).__name__}) is not." + ) + assert self.crawler is not None + assert self.crawler.throttler is not None + self._throttler: ThrottlingManagerProtocol = self.crawler.throttler + return result + + def enqueue_request(self, request: Request) -> bool: + raise RuntimeError( + "ThrottlingAwareScheduler requires the asynchronous enqueue path; " + "enqueue_request_async() is used by the engine instead of " + "enqueue_request()." + ) + + async def enqueue_request_async(self, request: Request) -> bool: + """Resolve the :ref:`throttling scope set ` of + *request* and store it under the queue for that scope set. + + This is the asynchronous counterpart of + :meth:`Scheduler.enqueue_request`; scope resolution + (:meth:`~scrapy.throttling.ThrottlingManagerProtocol.get_scopes`) is + asynchronous, which is why enqueuing has to be too. + """ + if not request.dont_filter and self.df.request_seen(request): + self.df.log(request, self.spider) + return False + scope_set = frozenset(iter_scopes(await self._throttler.get_scopes(request))) + assert self.stats is not None + if self._dqpush_throttling(request, scope_set): + self.stats.inc_value("scheduler/enqueued/disk") + else: + self.mqs.push(request, scope_set) # type: ignore[call-arg] + self.stats.inc_value("scheduler/enqueued/memory") + self.stats.inc_value("scheduler/enqueued") + return True + + def _dqpush_throttling(self, request: Request, scope_set: frozenset[str]) -> bool: + if self.dqs is None: + return False + try: + self.dqs.push(request, scope_set) # type: ignore[call-arg] + except ValueError as e: # non serializable request + if self.logunser: + logger.warning( + "Unable to serialize request: %(request)s - reason:" + " %(reason)s - no more unserializable requests will be" + " logged (stats being collected)", + {"request": request, "reason": e}, + exc_info=True, + extra={"spider": self.spider}, + ) + self.logunser = False + assert self.stats is not None + self.stats.inc_value("scheduler/unserializable") + return False + return True + + def next_request_delay(self) -> float | None: + """Return the minimum number of seconds until some pending request + becomes sendable because a time-based throttling gate opens, or + ``None`` if no pending request is time-blocked. + + The engine uses this to arm a single wakeup timer when + :meth:`next_request` returns ``None`` while requests are still + pending.""" + delays = [ + delay + for pq in (self.mqs, self.dqs) + if pq is not None and (delay := pq.next_request_delay()) is not None # type: ignore[attr-defined] + ] + return min(delays) if delays else None diff --git a/scrapy/pqueues.py b/scrapy/pqueues.py index 0ad0b5d78..60675fe7f 100644 --- a/scrapy/pqueues.py +++ b/scrapy/pqueues.py @@ -1,6 +1,7 @@ from __future__ import annotations import hashlib +import json import logging from typing import TYPE_CHECKING, Protocol, cast @@ -15,6 +16,7 @@ if TYPE_CHECKING: from scrapy import Request from scrapy.core.downloader import Downloader from scrapy.crawler import Crawler + from scrapy.throttling import ScopeID, ThrottlingManagerProtocol logger = logging.getLogger(__name__) @@ -437,3 +439,179 @@ class DownloaderAwarePriorityQueue: def __contains__(self, slot: str) -> bool: return slot in self.pqueues + + +def _scope_set_key(scope_set: frozenset[ScopeID]) -> str: + """Return a reversible, JSON-safe string key for *scope_set*. + + Used both as the in-memory dict key encoding for the on-disk state and to + derive the per-scope-set subdirectory name. The encoding is + order-independent (the scope ids are sorted).""" + return json.dumps(sorted(scope_set)) + + +def _scope_set_from_key(key: str) -> frozenset[ScopeID]: + return frozenset(json.loads(key)) + + +class ThrottlingAwarePriorityQueue: + """Priority queue that partitions requests by their full :ref:`throttling + scope set ` and only ever pops a request that can be + sent right now. + + The downstream queue class must support ``peek``. + + Disk persistence + ================ + + .. warning:: The files that this class generates on disk are an + implementation detail, and may change without a warning in a future + version of Scrapy. Do not rely on the following information for + anything other than debugging purposes. + + When instantiated with a non-empty *key* argument, *key* is used as a + persistence directory, and inside it this class creates a subdirectory per + scope set, named from a path-safe, order-independent encoding of its scope + ids. + """ + + @classmethod + def from_crawler( + cls, + crawler: Crawler, + downstream_queue_cls: type[QueueProtocol], + key: str, + startprios: dict[str, Iterable[int]] | None = None, + *, + start_queue_cls: type[QueueProtocol] | None = None, + ) -> Self: + return cls( + crawler, + downstream_queue_cls, + key, + startprios, + start_queue_cls=start_queue_cls, + ) + + def __init__( + self, + crawler: Crawler, + downstream_queue_cls: type[QueueProtocol], + key: str, + slot_startprios: dict[str, Iterable[int]] | None = None, + *, + start_queue_cls: type[QueueProtocol] | None = None, + ): + if slot_startprios and not isinstance(slot_startprios, dict): + raise ValueError( + "ThrottlingAwarePriorityQueue accepts ``slot_startprios`` as a " + f"dict; {slot_startprios.__class__!r} instance is passed. Most " + "likely, it means the state is created by an incompatible " + "priority queue. Only a crawl started with the same priority " + "queue class can be resumed." + ) + + assert crawler.throttler is not None + self._throttler: ThrottlingManagerProtocol = crawler.throttler + self.downstream_queue_cls: type[QueueProtocol] = downstream_queue_cls + self._start_queue_cls: type[QueueProtocol] | None = start_queue_cls + self.key: str = key + self.crawler: Crawler = crawler + + # scope set -> priority queue + self.pqueues: dict[frozenset[ScopeID], ScrapyPriorityQueue] = {} + if slot_startprios: + for set_key, startprios in slot_startprios.items(): + scope_set = _scope_set_from_key(set_key) + self.pqueues[scope_set] = self.pqfactory(scope_set, startprios) + + def pqfactory( + self, scope_set: frozenset[ScopeID], startprios: Iterable[int] = () + ) -> ScrapyPriorityQueue: + return ScrapyPriorityQueue( + self.crawler, + self.downstream_queue_cls, + self.key + "/" + _path_safe(_scope_set_key(scope_set)), + startprios, + start_queue_cls=self._start_queue_cls, + ) + + def push(self, request: Request, scope_set: frozenset[ScopeID]) -> None: + if scope_set not in self.pqueues: + self.pqueues[scope_set] = self.pqfactory(scope_set) + self.pqueues[scope_set].push(request) + + def _select( + self, + ) -> tuple[frozenset[ScopeID], ScrapyPriorityQueue] | None: + """Return the sendable ``(scope_set, queue)`` pair to pop from, or + ``None`` if no queue can be popped from right now. + + Among the sendable queues (those whose scope set can be sent right now), + the one whose head has the highest request priority is chosen; ties are + broken by ascending load (the maximum + :meth:`~scrapy.throttling.ThrottlingManagerProtocol.scope_load` over the + scopes of the queue), i.e. by preferring the least-busy scopes. + """ + best_sort_key: tuple[int, float] | None = None + best: tuple[frozenset[ScopeID], ScrapyPriorityQueue] | None = None + for scope_set, queue in self.pqueues.items(): + head = queue.peek() + if head is None or not self._throttler.is_ready(head): + continue + load = max( + (self._throttler.scope_load(scope_id) for scope_id in scope_set), + default=0.0, + ) + sort_key = (queue.priority(head), load) + if best_sort_key is None or sort_key < best_sort_key: + best_sort_key = sort_key + best = (scope_set, queue) + return best + + def pop(self) -> Request | None: + selected = self._select() + if selected is None: + return None + scope_set, queue = selected + request = queue.pop() + if request is not None: + self._throttler.reserve(request) + if len(queue) == 0: + del self.pqueues[scope_set] + return request + + def peek(self) -> Request | None: + selected = self._select() + if selected is None: + return None + return selected[1].peek() + + def next_request_delay(self) -> float | None: + delay: float | None = None + for queue in self.pqueues.values(): + head = queue.peek() + if head is None: + continue + if self._throttler.is_ready(head): + return 0.0 + head_delay = self._throttler.time_until_ready(head) + if head_delay is None: + continue + if delay is None or head_delay < delay: + delay = head_delay + return delay + + def close(self) -> dict[str, list[int]]: + active = { + _scope_set_key(scope_set): queue.close() + for scope_set, queue in self.pqueues.items() + } + self.pqueues.clear() + return active + + def __len__(self) -> int: + return sum(len(x) for x in self.pqueues.values()) if self.pqueues else 0 + + def __contains__(self, scope_set: frozenset[ScopeID]) -> bool: + return scope_set in self.pqueues diff --git a/scrapy/throttling.py b/scrapy/throttling.py index 9b70ec23f..de6589796 100644 --- a/scrapy/throttling.py +++ b/scrapy/throttling.py @@ -317,6 +317,48 @@ class ThrottlingManagerProtocol(Protocol): enforce a concurrency limit can let other requests through. """ + def is_ready(self, request: Request) -> bool: + """Return whether every scope of *request* allows it to be sent right + now, i.e. all time-based gates (delay, backoff, quota window) are open + *and* a concurrency slot is free in every scope. + + This is the synchronous, non-blocking counterpart of :meth:`acquire`, + used by a :ref:`throttling-aware scheduler ` + to decide whether a request can be dequeued now. It assumes the scopes + of *request* have already been resolved (e.g. by an earlier + :meth:`get_scopes` call at enqueue time). + """ + + def reserve(self, request: Request) -> None: + """Claim a send for *request*: record the send on every one of its + scopes and mark *request* as reserved, so that a later :meth:`acquire` + for it returns immediately without reserving again. + + A :ref:`throttling-aware scheduler ` calls + this when it decides to dequeue *request* (after :meth:`is_ready` + returned ``True``). The reservation is released by :meth:`release`. + """ + + def time_until_ready(self, request: Request) -> float | None: + """Return the number of seconds until every time-based gate of + *request* would be open, or ``None`` if no time-based gate is currently + blocking it (only a concurrency slot could be). + + Used by a :ref:`throttling-aware scheduler + ` to schedule a wakeup when all pending + requests are time-blocked. + """ + + def scope_load(self, scope_id: str) -> float: + """Return the current load of the scope identified by *scope_id*: its + active sends divided by its concurrency limit (or by the global + :setting:`CONCURRENT_REQUESTS` when the scope has no explicit limit). + + Used by a :ref:`throttling-aware scheduler + ` to balance dequeuing across scopes, + preferring the least-loaded ones. + """ + async def process_response(self, response: Response) -> None: """Update the throttling state based on *response*.""" @@ -352,10 +394,14 @@ def scope_cache(f: _GetScopesMethod) -> _GetScopesMethod: async def get_scopes(self, request): return urlparse_cached(request).netloc """ - cache: WeakKeyDictionary[Request, RequestScopes] = WeakKeyDictionary() @wraps(f) async def wrapper(self: Any, request: Request) -> RequestScopes: + cache: WeakKeyDictionary[Request, RequestScopes] | None = self.__dict__.get( + "_scope_cache" + ) + if cache is None: + cache = self.__dict__["_scope_cache"] = WeakKeyDictionary() if request in cache: return cache[request] scopes = await f(self, request) @@ -405,6 +451,7 @@ class ThrottlingManager: "THROTTLING_SCOPES" ) self._scope_managers: dict[ScopeID, ThrottlingScopeManagerProtocol] = {} + self._global_concurrency: int = crawler.settings.getint("CONCURRENT_REQUESTS") self._last_eviction: float | None = None # Concurrency slots reserved by acquire(), to be released once the # request finishes downloading. @@ -414,11 +461,37 @@ class ThrottlingManager: @scope_cache async def get_scopes(self, request: Request) -> RequestScopes: + return self._resolve_scopes_sync(request) + + def _resolve_scopes_sync(self, request: Request) -> RequestScopes: + """Best-effort synchronous scope resolution. + + It mirrors :meth:`get_scopes`, and is used by the synchronous + readiness methods (:meth:`is_ready`, :meth:`reserve`, + :meth:`time_until_ready`) when the cached result of an earlier + :meth:`get_scopes` call is not available (e.g. for a request restored + from disk). Subclasses whose :meth:`get_scopes` cannot be resolved + synchronously should rely on the enqueue-time cache instead. + """ scopes = request.meta.get("throttling_scopes") if scopes is not None: return cast("RequestScopes", scopes) return urlparse_cached(request).netloc + def _cached_scope_values( + self, request: Request + ) -> list[tuple[ScopeID, float | None]]: + """Return the ``(scope_id, quota_amount)`` pairs of *request*, reading + the cache populated by :meth:`get_scopes` and falling back to + :meth:`_resolve_scopes_sync`.""" + cache: WeakKeyDictionary[Request, RequestScopes] | None = self.__dict__.get( + "_scope_cache" + ) + scopes = cache.get(request) if cache is not None else None + if scopes is None: + scopes = self._resolve_scopes_sync(request) + return list(iter_scope_values(scopes)) + async def get_initial_backoff(self) -> BackoffData: return None @@ -472,6 +545,10 @@ class ThrottlingManager: return manager async def acquire(self, request: Request) -> None: + # A throttling-aware scheduler reserves the request before handing it + # to the engine, so there is nothing left to wait for or record here. + if request in self._reserved: + return now = time.monotonic() self._maybe_evict(now) await self._apply_request_delay(request) @@ -518,6 +595,44 @@ class ThrottlingManager: for manager, _ in managers: manager.record_done() + # -- Synchronous readiness API (used by a throttling-aware scheduler) ------ + + def is_ready(self, request: Request) -> bool: + now = time.monotonic() + for scope_id, value in self._cached_scope_values(request): + manager = self._get_scope_manager(scope_id) + if manager.can_send(now=now, amount=value) > 0: + return False + if manager.concurrency_blocked(): + return False + return True + + def reserve(self, request: Request) -> None: + managers = [ + (self._get_scope_manager(scope_id), value) + for scope_id, value in self._cached_scope_values(request) + ] + for manager, value in managers: + manager.record_sent(amount=value) + self._reserved[request] = managers + + def time_until_ready(self, request: Request) -> float | None: + now = time.monotonic() + wait = 0.0 + for scope_id, value in self._cached_scope_values(request): + manager = self._get_scope_manager(scope_id) + wait = max(wait, manager.can_send(now=now, amount=value)) + return wait if wait > 0 else None + + def scope_load(self, scope_id: ScopeID) -> float: + manager = self._get_scope_manager(scope_id) + active: int = getattr(manager, "_active", 0) + concurrency: int | None = getattr(manager, "_concurrency", None) + limit = concurrency if concurrency is not None else self._global_concurrency + if not limit: + return 0.0 + return active / limit + async def _wait_for_slot(self, managers: list[Any]) -> None: """Block until any of *managers* frees a concurrency slot. @@ -749,13 +864,13 @@ class ThrottlingScopeManagerProtocol(Protocol): Return ``False`` when no concurrency limit is enforced. """ - def slot_event(self) -> Deferred: + def slot_event(self) -> Deferred[None]: """Return a :class:`~twisted.internet.defer.Deferred` that fires when a concurrency slot next becomes available (e.g. when :meth:`record_done` is called or the limit is raised via :meth:`set_concurrency`).""" - def discard_slot_event(self, event: Deferred) -> None: + def discard_slot_event(self, event: Deferred[None]) -> None: """Cancel a pending slot event returned by :meth:`slot_event`. Called by :class:`ThrottlingManager` when the wait ends without the diff --git a/tests/test_pqueues.py b/tests/test_pqueues.py index 6c6a6584a..9843ca4d5 100644 --- a/tests/test_pqueues.py +++ b/tests/test_pqueues.py @@ -6,12 +6,18 @@ import queuelib from scrapy.core.downloader import Downloader from scrapy.http.request import Request -from scrapy.pqueues import DownloaderAwarePriorityQueue, ScrapyPriorityQueue +from scrapy.pqueues import ( + DownloaderAwarePriorityQueue, + ScrapyPriorityQueue, + ThrottlingAwarePriorityQueue, +) from scrapy.spiders import Spider from scrapy.squeues import FifoMemoryQueue, PickleFifoDiskQueue +from scrapy.throttling import iter_scopes from scrapy.utils.misc import build_from_crawler, load_object from scrapy.utils.test import get_crawler from tests.test_scheduler import MockDownloader +from tests.utils.decorators import coroutine_test class TestPriorityQueue: @@ -309,3 +315,107 @@ def test_pop_order(input_, output): actual_output_urls.append(request.url) assert actual_output_urls == expected_output_urls + + +class TestThrottlingAwarePriorityQueue: + def _queue(self, crawler, key=""): + return build_from_crawler( + ThrottlingAwarePriorityQueue, + crawler, + downstream_queue_cls=FifoMemoryQueue, + key=key, + ) + + async def _push(self, queue, crawler, request): + scope_set = frozenset(iter_scopes(await crawler.throttler.get_scopes(request))) + queue.push(request, scope_set) + + @coroutine_test + async def test_partitions_by_scope_set(self): + crawler = get_crawler(Spider) + queue = self._queue(crawler) + await self._push(queue, crawler, Request("http://a.com/1")) + await self._push(queue, crawler, Request("http://a.com/2")) + await self._push(queue, crawler, Request("http://b.com/1")) + # Two distinct scope sets -> two internal queues, three requests. + assert len(queue.pqueues) == 2 + assert len(queue) == 3 + + @coroutine_test + async def test_pop_skips_blocked_scope(self): + crawler = get_crawler( + Spider, + settings_dict={ + "THROTTLING_SCOPES": {"slow.com": {"delay": 1000.0}}, + "RANDOMIZE_DOWNLOAD_DELAY": False, + }, + ) + queue = self._queue(crawler) + await self._push(queue, crawler, Request("http://slow.com/1")) + await self._push(queue, crawler, Request("http://slow.com/2")) + await self._push(queue, crawler, Request("http://fast.com/1")) + # The first slow request is sendable (no delay accrued yet); after it is + # popped (and reserved), the second slow request is blocked, but the + # fast one is still served. + popped = [queue.pop(), queue.pop(), queue.pop()] + urls = [r.url if r else None for r in popped] + assert "http://fast.com/1" in urls + assert "http://slow.com/1" in urls + # The blocked second slow request stays in the queue. + assert None in urls + assert len(queue) == 1 + delay = queue.next_request_delay() + assert delay is not None + assert delay == pytest.approx(1000.0, abs=1.0) + + @coroutine_test + async def test_least_loaded_first(self): + crawler = get_crawler( + Spider, + settings_dict={ + "THROTTLING_SCOPES": { + "a.com": {"concurrency": 4}, + "b.com": {"concurrency": 4}, + } + }, + ) + queue = self._queue(crawler) + # Make a.com busier than b.com. + busy = Request("http://a.com/0") + await crawler.throttler.get_scopes(busy) + crawler.throttler.reserve(busy) + await self._push(queue, crawler, Request("http://a.com/1")) + await self._push(queue, crawler, Request("http://b.com/1")) + # Equal priority, so the lower-load scope (b.com) is served first. + assert queue.pop().url == "http://b.com/1" + + @coroutine_test + async def test_priority_beats_load(self): + crawler = get_crawler( + Spider, + settings_dict={ + "THROTTLING_SCOPES": { + "a.com": {"concurrency": 4}, + "b.com": {"concurrency": 4}, + } + }, + ) + queue = self._queue(crawler) + # Make a.com busier than b.com. + busy = Request("http://a.com/0") + await crawler.throttler.get_scopes(busy) + crawler.throttler.reserve(busy) + # The a.com request has higher priority, and a.com still has room, so it + # is served first despite a.com being the busier scope. + await self._push(queue, crawler, Request("http://a.com/1", priority=10)) + await self._push(queue, crawler, Request("http://b.com/1")) + assert queue.pop().url == "http://a.com/1" + + @coroutine_test + async def test_empty_and_close(self): + crawler = get_crawler(Spider) + queue = self._queue(crawler) + assert queue.pop() is None + assert queue.next_request_delay() is None + await self._push(queue, crawler, Request("http://a.com/1")) + assert queue.close() != {} diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py index 8637de41e..9919b05a0 100644 --- a/tests/test_scheduler.py +++ b/tests/test_scheduler.py @@ -10,7 +10,7 @@ from unittest.mock import Mock import pytest from scrapy.core.downloader import Downloader -from scrapy.core.scheduler import BaseScheduler, Scheduler +from scrapy.core.scheduler import BaseScheduler, Scheduler, ThrottlingAwareScheduler from scrapy.crawler import Crawler from scrapy.exceptions import ScrapyDeprecationWarning from scrapy.http import Request @@ -406,3 +406,175 @@ class TestIncompatibility: ValueError, match="does not support CONCURRENT_REQUESTS_PER_IP" ): self._incompatible() + + +_THROTTLING_AWARE_PQ = "scrapy.pqueues.ThrottlingAwarePriorityQueue" + + +class TestThrottlingAwareScheduler: + def _crawler(self, settings_dict: dict[str, Any] | None = None) -> Crawler: + settings = { + "SCHEDULER_PRIORITY_QUEUE": _THROTTLING_AWARE_PQ, + "DUPEFILTER_CLASS": "scrapy.dupefilters.BaseDupeFilter", + **(settings_dict or {}), + } + return get_crawler(Spider, settings) + + def _scheduler(self, crawler: Crawler) -> ThrottlingAwareScheduler: + spider = Spider(name="spider") + crawler.spider = spider + scheduler = ThrottlingAwareScheduler.from_crawler(crawler) + scheduler.open(spider) + return scheduler + + @coroutine_test + async def test_enqueue_async_and_dequeue(self) -> None: + scheduler = self._scheduler(self._crawler()) + assert await scheduler.enqueue_request_async(Request("http://a.com/1")) is True + assert scheduler.has_pending_requests() + assert len(scheduler) == 1 + request = scheduler.next_request() + assert request is not None + assert request.url == "http://a.com/1" + assert scheduler.next_request() is None + assert not scheduler.has_pending_requests() + scheduler.close("finished") + + def test_sync_enqueue_raises(self) -> None: + scheduler = self._scheduler(self._crawler()) + with pytest.raises(RuntimeError, match="asynchronous enqueue path"): + scheduler.enqueue_request(Request("http://a.com/1")) + scheduler.close("finished") + + def test_requires_throttling_aware_priority_queue(self) -> None: + crawler = self._crawler( + {"SCHEDULER_PRIORITY_QUEUE": "scrapy.pqueues.ScrapyPriorityQueue"} + ) + spider = Spider(name="spider") + crawler.spider = spider + scheduler = ThrottlingAwareScheduler.from_crawler(crawler) + with pytest.raises(ValueError, match="throttling-aware priority queue"): + scheduler.open(spider) + + @coroutine_test + async def test_delay_blocks_and_reports_delay(self) -> None: + crawler = self._crawler( + { + "THROTTLING_SCOPES": {"slow.com": {"delay": 1000.0}}, + "RANDOMIZE_DOWNLOAD_DELAY": False, + }, + ) + scheduler = self._scheduler(crawler) + await scheduler.enqueue_request_async(Request("http://slow.com/1")) + await scheduler.enqueue_request_async(Request("http://slow.com/2")) + # The first request is sendable; the second is blocked by the delay. + first = scheduler.next_request() + assert first is not None + assert scheduler.next_request() is None + assert scheduler.has_pending_requests() + assert scheduler.next_request_delay() == pytest.approx(1000.0, abs=1.0) + scheduler.close("finished") + + @coroutine_test + async def test_no_delay_when_only_concurrency_blocked(self) -> None: + crawler = self._crawler( + {"THROTTLING_SCOPES": {"slow.com": {"concurrency": 1}}}, + ) + scheduler = self._scheduler(crawler) + await scheduler.enqueue_request_async(Request("http://slow.com/1")) + await scheduler.enqueue_request_async(Request("http://slow.com/2")) + assert scheduler.next_request() is not None + assert scheduler.next_request() is None + # A purely concurrency-blocked state has no time-based wakeup. + assert scheduler.next_request_delay() is None + scheduler.close("finished") + + @coroutine_test + async def test_resume_from_disk(self, tmp_path: Path) -> None: + settings = {"JOBDIR": str(tmp_path)} + scheduler = self._scheduler(self._crawler(settings)) + await scheduler.enqueue_request_async(Request("http://a.com/1")) + await scheduler.enqueue_request_async(Request("http://b.com/1")) + scheduler.close("shutdown") + + scheduler2 = self._scheduler(self._crawler(settings)) + assert len(scheduler2) == 2 + urls = set() + while (request := scheduler2.next_request()) is not None: + urls.add(request.url) + assert urls == {"http://a.com/1", "http://b.com/1"} + scheduler2.close("finished") + + +class TestIntegrationWithThrottlingAwareScheduler: + @inline_callbacks_test + def test_integration(self): + crawler = get_crawler( + spidercls=StartUrlsSpider, + settings_dict={ + "SCHEDULER": "scrapy.core.scheduler.ThrottlingAwareScheduler", + "SCHEDULER_PRIORITY_QUEUE": "scrapy.pqueues.ThrottlingAwarePriorityQueue", + "DUPEFILTER_CLASS": "scrapy.dupefilters.BaseDupeFilter", + }, + ) + with MockServer() as mockserver: + url = mockserver.url("/status?n=200", is_secure=False) + start_urls = [url] * 6 + yield crawler.crawl(start_urls) + assert crawler.stats.get_value("downloader/response_count") == len( + start_urls + ) + + @inline_callbacks_test + def test_integration_follow_requests(self): + # Exercises the engine's asynchronous enqueue path for requests yielded + # from a callback (not just start requests), and the idle guard that + # keeps the spider open while an enqueue is in flight. + class FollowSpider(Spider): + name = "follow" + + def __init__(self, base_url, **kwargs): + self.base_url = base_url + super().__init__(**kwargs) + + async def start(self): + yield Request(self.base_url + "/status?n=200") + + def parse(self, response): + if b"follow" not in response.url.encode(): + yield Request(response.url + "&follow=1", callback=self.parse) + + with MockServer() as mockserver: + base_url = mockserver.url("", is_secure=False).rstrip("/") + crawler = get_crawler( + spidercls=FollowSpider, + settings_dict={ + "SCHEDULER": "scrapy.core.scheduler.ThrottlingAwareScheduler", + "SCHEDULER_PRIORITY_QUEUE": "scrapy.pqueues.ThrottlingAwarePriorityQueue", + "DUPEFILTER_CLASS": "scrapy.dupefilters.BaseDupeFilter", + }, + ) + yield crawler.crawl(base_url=base_url) + assert crawler.stats.get_value("downloader/response_count") == 2 + + @inline_callbacks_test + def test_integration_with_delay(self): + # A small per-scope delay forces the engine to arm the throttling wakeup + # timer between requests; the crawl must still complete. + crawler = get_crawler( + spidercls=StartUrlsSpider, + settings_dict={ + "SCHEDULER": "scrapy.core.scheduler.ThrottlingAwareScheduler", + "SCHEDULER_PRIORITY_QUEUE": "scrapy.pqueues.ThrottlingAwarePriorityQueue", + "DUPEFILTER_CLASS": "scrapy.dupefilters.BaseDupeFilter", + "RANDOMIZE_DOWNLOAD_DELAY": False, + "THROTTLING_SCOPES": {"127.0.0.1": {"delay": 0.05}}, + }, + ) + with MockServer() as mockserver: + url = mockserver.url("/status?n=200", is_secure=False) + start_urls = [url] * 4 + yield crawler.crawl(start_urls) + assert crawler.stats.get_value("downloader/response_count") == len( + start_urls + ) diff --git a/tests/test_throttling.py b/tests/test_throttling.py index 0e0269c95..bd71ac255 100644 --- a/tests/test_throttling.py +++ b/tests/test_throttling.py @@ -540,6 +540,91 @@ class TestThrottlingScopeManager: assert scope._concurrency == 1 +class TestThrottlingManagerReadiness: + """The synchronous readiness API used by a throttling-aware scheduler.""" + + @coroutine_test + async def test_is_ready_unconstrained_scope(self): + manager = _manager() + request = Request("http://example.com/a") + # A scope with no configured delay/concurrency/quota is always ready. + assert await manager.get_scopes(request) == "example.com" + assert manager.is_ready(request) is True + + @coroutine_test + async def test_is_ready_without_cached_scopes(self): + # is_ready falls back to synchronous resolution when get_scopes was not + # called first (e.g. for a request restored from disk). + manager = _manager() + request = Request("http://example.com/a") + assert manager.is_ready(request) is True + + @coroutine_test + async def test_reserve_blocks_delay_scope(self): + manager = _manager( + { + "THROTTLING_SCOPES": {"example.com": {"delay": 100.0}}, + "RANDOMIZE_DOWNLOAD_DELAY": False, + } + ) + first = Request("http://example.com/1") + second = Request("http://example.com/2") + await manager.get_scopes(first) + await manager.get_scopes(second) + assert manager.is_ready(first) is True + manager.reserve(first) + # The base delay now blocks any further request for the scope. + assert manager.is_ready(second) is False + assert manager.time_until_ready(second) == pytest.approx(100.0, abs=1.0) + + @coroutine_test + async def test_reserve_blocks_on_concurrency(self): + manager = _manager({"THROTTLING_SCOPES": {"example.com": {"concurrency": 1}}}) + first = Request("http://example.com/1") + second = Request("http://example.com/2") + await manager.get_scopes(first) + await manager.get_scopes(second) + manager.reserve(first) + assert manager.is_ready(second) is False + # Pure concurrency blocking is not time-gated. + assert manager.time_until_ready(second) is None + manager.release(first) + assert manager.is_ready(second) is True + + @coroutine_test + async def test_acquire_noop_when_reserved(self): + manager = _manager( + { + "THROTTLING_SCOPES": {"example.com": {"delay": 100.0}}, + "RANDOMIZE_DOWNLOAD_DELAY": False, + } + ) + request = Request("http://example.com/1") + await manager.get_scopes(request) + manager.reserve(request) + # A reserved request fast-paths through acquire() without re-recording + # the send or waiting for the delay. + await manager.acquire(request) + scope = manager._get_scope_manager("example.com") + assert scope._active == 1 # reserve recorded exactly one send + + @coroutine_test + async def test_scope_load(self): + manager = _manager({"THROTTLING_SCOPES": {"example.com": {"concurrency": 4}}}) + assert manager.scope_load("example.com") == 0.0 + request = Request("http://example.com/1") + await manager.get_scopes(request) + manager.reserve(request) + assert manager.scope_load("example.com") == pytest.approx(0.25) + + def test_scope_load_falls_back_to_global_concurrency(self): + manager = _manager({"CONCURRENT_REQUESTS": 8}) + # A scope with no explicit concurrency limit uses CONCURRENT_REQUESTS. + request = Request("http://example.com/1") + manager.reserve(request) + assert manager.scope_load("example.com") == pytest.approx(1 / 8) + + class TestThrottlingIntegration: @coroutine_test async def test_backoff_recorded_on_429(self, mockserver):