diff --git a/docs/news.rst b/docs/news.rst index 634b1d934..aa404d760 100644 --- a/docs/news.rst +++ b/docs/news.rst @@ -28,9 +28,10 @@ Backward-incompatible changes requests have been scheduled. As a result, the order in which start requests are sent may change. See - :ref:`start-requests` for details and information on how to force start - request order or :ref:`pause start request iteration while there are - scheduled requests `. + :ref:`start-requests` for details. + + To restore the previous behavior, :ref:`use lazy start request scheduling + `. - An unhandled exception from the :meth:`~scrapy.spidermiddlewares.SpiderMiddleware.open_spider` method of a @@ -46,7 +47,7 @@ Backward-incompatible changes - The ``slot`` attribute has been renamed to ``_slot`` and should not be used. - To access the running scheduler, previously at ``_slot.scheduler``, use + To access the running scheduler, previously at ``slot.scheduler``, use the :attr:`~scrapy.core.engine.ExecutionEngine.scheduler` attribute of the running engine instead. @@ -101,14 +102,11 @@ New features (:issue:`456`, :issue:`3477`, :issue:`4467`, :issue:`5627`, :issue:`6729`) -- :class:`Crawler.signals ` has a new - :meth:`~scrapy.signalmanager.SignalManager.wait_for` method. +- Added official support for :ref:`idle ` and + :ref:`front load ` start request scheduling. -- Added a new :signal:`scheduler_empty` signal. - -- Exposed a new method of :class:`Crawler.engine - `: - :meth:`~scrapy.core.engine.ExecutionEngine.needs_backout`. + (:issue:`740`, :issue:`1051`, :issue:`1443`, :issue:`3237`, :issue:`5282`, + :issue:`6715`) - You can now raise :exc:`~scrapy.exceptions.CloseSpider` from :meth:`~scrapy.Spider.start` and from @@ -129,6 +127,20 @@ New features (:issue:`5426`) +- :class:`Crawler.signals ` has a new + :meth:`~scrapy.signalmanager.SignalManager.wait_for` method. + +- Added new signals: :signal:`spider_start_blocking`, + :signal:`scheduler_empty`. + +- In :class:`~scrapy.core.engine.ExecutionEngine`: + + - Added a :attr:`~scrapy.core.engine.ExecutionEngine.scheduler` + attribute. + + - Added a :meth:`~scrapy.core.engine.ExecutionEngine.needs_backout` + method. + Bug fixes ~~~~~~~~~ diff --git a/docs/topics/api.rst b/docs/topics/api.rst index 54b4b4830..9c7d77bfe 100644 --- a/docs/topics/api.rst +++ b/docs/topics/api.rst @@ -274,6 +274,8 @@ class (which they all inherit from). Close the given spider. After this is called, no more specific stats can be accessed or collected. +.. _engine: + Engine API ========== diff --git a/docs/topics/optimize.rst b/docs/topics/optimize.rst index 59d3a1089..39bd13747 100644 --- a/docs/topics/optimize.rst +++ b/docs/topics/optimize.rst @@ -73,6 +73,8 @@ For broad crawls, consider these adjustments: Lowering resource usage ======================= +.. _optimize-memory: + Lowering memory usage --------------------- @@ -86,7 +88,7 @@ Lowering memory usage requests. - If you have many :ref:`start requests `, use :ref:`lazy - scheduling `. + ` or :ref:`idle ` scheduling. - Set :setting:`JOBDIR` to offload all scheduled requests to disk. diff --git a/docs/topics/signals.rst b/docs/topics/signals.rst index 66cb87fc5..2a8f9e79c 100644 --- a/docs/topics/signals.rst +++ b/docs/topics/signals.rst @@ -112,7 +112,7 @@ engine_started .. signal:: engine_started .. function:: engine_started() - Sent when the Scrapy engine has started crawling. + Sent when the the :ref:`engine ` has started crawling. This signal supports returning deferreds from its handlers. @@ -126,23 +126,22 @@ engine_stopped .. signal:: engine_stopped .. function:: engine_stopped() - Sent when the Scrapy engine is stopped (for example, when a crawling - process has finished). + Sent when the the :ref:`engine ` is stopped (for example, when a + crawling process has finished). This signal supports returning deferreds from its handlers. +spider_start_blocking +~~~~~~~~~~~~~~~~~~~~~ + +.. signal:: spider_start_blocking +.. autofunction:: spider_start_blocking() + scheduler_empty ~~~~~~~~~~~~~~~ .. signal:: scheduler_empty -.. function:: scheduler_empty() - - Sent whenever the engine asks for a pending request from the - :ref:`scheduler ` (i.e. calls its - :meth:`~scrapy.core.scheduler.BaseScheduler.next_request` method) and the - scheduler returns none. - - See :ref:`start-requests-lazy` for an example. +.. autofunction:: scheduler_empty() Item signals ------------ @@ -299,6 +298,8 @@ spider_idle :param spider: the spider which has gone idle :type spider: :class:`~scrapy.Spider` object + .. seealso:: :signal:`spider_start_blocking` + .. note:: Scheduling some requests in your :signal:`spider_idle` handler does **not** guarantee that it can prevent the spider from being closed, although it sometimes can. That's because the spider may still remain idle diff --git a/docs/topics/spiders.rst b/docs/topics/spiders.rst index c24dd7dd9..0c05b34ea 100644 --- a/docs/topics/spiders.rst +++ b/docs/topics/spiders.rst @@ -466,17 +466,21 @@ while yielding your start requests: self.crawler.engine.scheduler.unpause() .. note:: If you use a :ref:`custom scheduler `, make sure it - supports pausing and unpausing, like - :class:`~scrapy.core.scheduler.Scheduler` does. + supports pausing like :class:`~scrapy.core.scheduler.Scheduler`. + +Delaying start request scheduling +--------------------------------- + +When :ref:`optimizing memory usage `, it can sometimes be +useful to delay start request scheduling :ref:`until the scheduler is empty +` or :ref:`until the engine is idle +`. .. _start-requests-lazy: -Lazy start request scheduling ------------------------------ - -To use lazy start request scheduling, where the iteration of start requests is -paused until the scheduler is empty, override the :meth:`~scrapy.Spider.start` -method as follows: +To use **lazy** start request scheduling, where the iteration of start requests +is paused until the :ref:`scheduler ` is empty, override the +:meth:`~scrapy.Spider.start` method as follows: .. code-block:: python @@ -486,9 +490,24 @@ method as follows: await self.crawler.signals.wait_for(signals.scheduler_empty) yield item_or_request -This can help minimize the number of requests in the scheduler at any given -time, to minimize resource usage (memory or disk, depending on -:setting:`JOBDIR`). +.. _start-requests-idle: + +To use **idle** start request scheduling, where the iteration of start requests +is paused until the :ref:`engine ` needs a start request to continue, +override the :meth:`~scrapy.Spider.start` method as follows: + +.. code-block:: python + + async def start(self): + async for item_or_request in super().start(): + await self.crawler.signals.wait_for(signals.spider_start_blocking) + yield item_or_request + +.. warning:: While lazy scheduling only affects request order, **idle + scheduling can slow down your crawl**. It is functionally equivalent to + running a spider multiple times in a row, one per start request. + +.. seealso:: :class:`~scrapy.crawler.Crawler`, :ref:`topics-signals`. .. _builtin-spiders: diff --git a/scrapy/core/engine.py b/scrapy/core/engine.py index ba7e4e049..db85aedfc 100644 --- a/scrapy/core/engine.py +++ b/scrapy/core/engine.py @@ -218,7 +218,6 @@ class ExecutionEngine: self._slot.nextcall.schedule() def _scheduler_has_pending_requests(self) -> bool: - assert self._slot is not None # typing assert self.scheduler is not None # typing try: return self.scheduler.has_pending_requests() @@ -282,7 +281,8 @@ class ExecutionEngine: """Returns ``True`` if no more requests can be sent at the moment, or ``False`` otherwise. - See :ref:`start-requests-lazy` for an example. + Can be used, for example, for :ref:`lazy start request scheduling + `. """ assert self._slot is not None # typing assert self.scraper.slot is not None # typing @@ -375,9 +375,12 @@ class ExecutionEngine: return False if self.downloader.active: # downloader has pending requests return False - if self._start is not None: # not all start requests are handled + if self._scheduler_has_pending_requests(): return False - return not self._scheduler_has_pending_requests() + if self._start is not None: # not all start requests are handled + self.signals.send_catch_log(signals.spider_start_blocking) + return False + return True def crawl(self, request: Request) -> None: """Inject the request into the spider <-> downloader pipeline""" diff --git a/scrapy/signalmanager.py b/scrapy/signalmanager.py index f8c50b5e3..51859b727 100644 --- a/scrapy/signalmanager.py +++ b/scrapy/signalmanager.py @@ -78,7 +78,8 @@ class SignalManager: async def wait_for(self, signal): """Await the next *signal*. - See :ref:`start-requests-lazy` for an example. + Can be used, for example, for :ref:`lazy start request scheduling + `. """ d = Deferred() diff --git a/scrapy/signals.py b/scrapy/signals.py index bdeec1ba0..d01d46785 100644 --- a/scrapy/signals.py +++ b/scrapy/signals.py @@ -7,7 +7,24 @@ signals here without documenting them there. engine_started = object() engine_stopped = object() + +#: Sent when the :ref:`engine ` is waiting for :meth:`Spider.start +#: ` to either finish or yield its next start item or +#: request, and is otherwise idle (:signal:`spider_idle`). +#: +#: Can be used, for example, for :ref:`idle start request scheduling +#: `. +spider_start_blocking = object() + +#: Sent whenever the :ref:`engine ` asks for a pending request from the +#: :ref:`scheduler ` (i.e. calls its +#: :meth:`~scrapy.core.scheduler.BaseScheduler.next_request` method) and the +#: scheduler returns none. +#: +#: Can be used, for example, for :ref:`lazy start request scheduling +#: `. scheduler_empty = object() + spider_opened = object() spider_idle = object() spider_closed = object() diff --git a/tests/test_engine_loop.py b/tests/test_engine_loop.py index ef947f441..f6391b5d1 100644 --- a/tests/test_engine_loop.py +++ b/tests/test_engine_loop.py @@ -2,8 +2,8 @@ from __future__ import annotations from collections import deque from logging import ERROR +from typing import TYPE_CHECKING -import pytest from testfixtures import LogCapture from twisted.internet.defer import Deferred from twisted.trial.unittest import TestCase @@ -16,6 +16,9 @@ from scrapy.utils.test import get_crawler from .mockserver import MockServer from .test_scheduler import MemoryScheduler +if TYPE_CHECKING: + from scrapy.http import Response + async def sleep(seconds: float = ExecutionEngine._MIN_BACK_IN_SECONDS) -> None: from twisted.internet import reactor @@ -205,11 +208,14 @@ class RequestSendOrderTestCase(TestCase): def tearDownClass(cls): cls.mockserver.__exit__(None, None, None) # increase if flaky - def _request(self, num, response_seconds, download_slots=1): + def request(self, num, response_seconds, download_slots=1): url = self.mockserver.url(f"/delay?n={response_seconds}&{num}") meta = {"download_slot": str(num % download_slots)} return Request(url, meta=meta) + def get_num(self, request_or_response: Request | Response): + return int(request_or_response.url.rsplit("&", maxsplit=1)[1]) + @deferred_f_from_coro_f async def _test_request_order( self, @@ -219,35 +225,37 @@ class RequestSendOrderTestCase(TestCase): response_seconds=None, download_slots=1, start_fn=None, + parse_fn=None, ): cb_nums = cb_nums or [] settings = settings or {} response_seconds = response_seconds or self.seconds + cb_requests = deque( + [self.request(num, response_seconds, download_slots) for num in cb_nums] + ) + if start_fn is None: async def start_fn(spider): for num in start_nums: - yield self._request(num, response_seconds, download_slots) + yield self.request(num, response_seconds, download_slots) + + if parse_fn is None: + + def parse_fn(spider, response): + while cb_requests: + yield cb_requests.popleft() class TestSpider(Spider): name = "test" - cb_requests = deque( - [ - self._request(num, response_seconds, download_slots) - for num in cb_nums - ] - ) start = start_fn - - def parse(self, response): - while self.cb_requests: - yield self.cb_requests.popleft() + parse = parse_fn actual_nums = [] def track_num(request, spider): - actual_nums.append(int(request.url.rsplit("&", maxsplit=1)[1])) + actual_nums.append(self.get_num(request)) crawler = get_crawler(TestSpider, settings_dict=settings) crawler.signals.connect(track_num, signals.request_reached_downloader) @@ -268,7 +276,7 @@ class RequestSendOrderTestCase(TestCase): async def start(spider): for num in start_nums: - request = self._request(num, response_seconds, download_slots) + request = self.request(num, response_seconds, download_slots) yield request.replace(priority=1) await maybe_deferred_to_future( @@ -291,7 +299,7 @@ class RequestSendOrderTestCase(TestCase): async def start(spider): priority = len(start_nums) for num in start_nums: - request = self._request(num, response_seconds, download_slots) + request = self.request(num, response_seconds, download_slots) yield request.replace(priority=priority) priority -= 1 @@ -307,36 +315,30 @@ class RequestSendOrderTestCase(TestCase): @deferred_f_from_coro_f async def test_front_load(self): - class TestSpider(Spider): - name = "test" + start_nums = [2, 1] + response_seconds = 0 + download_slots = 1 - async def start(self): - assert self.crawler.engine is not None # typing - assert isinstance( - self.crawler.engine.scheduler, MemoryScheduler - ) # typing - self.crawler.engine.scheduler.pause() - # By pausing the scheduler, a is scheduled before b is sent, - # and since the scheduler uses a LIFO queue, a is sent first. - yield Request("data:,b") - yield Request("data:,a") - self.crawler.engine.scheduler.unpause() + async def start(spider): + # typing: + assert spider.crawler.engine is not None + assert isinstance(spider.crawler.engine.scheduler, MemoryScheduler) - def parse(self, response): - pass + spider.crawler.engine.scheduler.pause() + # By pausing the scheduler, a is scheduled before b is sent, + # and since the scheduler uses a LIFO queue, a is sent first. + yield self.request(2, response_seconds, download_slots) + yield self.request(1, response_seconds, download_slots) + spider.crawler.engine.scheduler.unpause() - actual_urls = [] - - def track_url(request, spider): - actual_urls.append(request.url) - - settings = {"SCHEDULER": MemoryScheduler} - crawler = get_crawler(TestSpider, settings_dict=settings) - crawler.signals.connect(track_url, signals.request_reached_downloader) - await maybe_deferred_to_future(crawler.crawl()) - assert crawler.stats.get_value("finish_reason") == "finished" - expected_urls = ["data:,a", "data:,b"] - assert actual_urls == expected_urls, f"{actual_urls=} != {expected_urls=}" + await maybe_deferred_to_future( + self._test_request_order( + start_nums=start_nums, + settings={"SCHEDULER": MemoryScheduler}, + response_seconds=response_seconds, + start_fn=start, + ) + ) @deferred_f_from_coro_f async def test_lazy(self): @@ -349,7 +351,7 @@ class RequestSendOrderTestCase(TestCase): for num in start_nums: if spider.crawler.engine.needs_backout(): await spider.crawler.signals.wait_for(signals.scheduler_empty) - request = self._request(num, response_seconds, download_slots) + request = self.request(num, response_seconds, download_slots) yield request await maybe_deferred_to_future( @@ -367,40 +369,48 @@ class RequestSendOrderTestCase(TestCase): ) ) - @pytest.mark.skip(reason="not implemented yet") @deferred_f_from_coro_f async def test_idle(self): - def _url(id): - return self.mockserver.url(f"/delay?n={self.delay}&{id}") + # The requests already in the scheduler (a) take priority over start + # requests. + # Callback requests (b, c) take priority over start requests as well. a + # yields b, b yields c. + # Once there are no more ongoing requests, the first start request (d) + # is sent. Then the requests from its callback (e) take priority. This + # is recursive, the requests from the callback of the callback (f) take + # priority as well. + # Only once there are no more ongoing requests again is the second + # start request (g) sent. + + nums = [1, 2, 3, 4, 5, 6, 7] + response_seconds = 0 + download_slots = 1 + + def _request(num): + return self.request(num, response_seconds, download_slots) class TestScheduler(MemoryScheduler): - queue = [_url("a")] + queue = [_request(1)] - class TestSpider(Spider): - name = "test" - start_urls = [_url("b"), _url("d")] - queue = deque([_url("c")]) + async def start(spider): + for request in [_request(4), _request(7)]: + await spider.crawler.signals.wait_for(signals.spider_start_blocking) + yield request - def parse(self, response): - try: - url = self.queue.popleft() - except IndexError: - pass - else: - yield Request(url) + def parse(spider, response): + num = self.get_num(response) + if num in {1, 2, 4, 5}: + yield _request(num + 1) - actual_urls = [] - - def track_url(request, spider): - actual_urls.append(request.url) - - settings = {"SCHEDULER": TestScheduler} - crawler = get_crawler(TestSpider, settings_dict=settings) - crawler.signals.connect(track_url, signals.request_reached_downloader) - await maybe_deferred_to_future(crawler.crawl()) - assert crawler.stats.get_value("finish_reason") == "finished" - expected_urls = [_url(letter) for letter in "abcd"] - assert actual_urls == expected_urls, f"{actual_urls=} != {expected_urls=}" + await maybe_deferred_to_future( + self._test_request_order( + start_nums=nums, + settings={"SCHEDULER": TestScheduler}, + response_seconds=response_seconds, + start_fn=start, + parse_fn=parse, + ) + ) # Delay handling @@ -413,7 +423,7 @@ class RequestSendOrderTestCase(TestCase): seconds = ExecutionEngine._MIN_BACK_IN_SECONDS def _request(num): - return self._request(num, seconds) + return self.request(num, seconds) async def start(spider): from twisted.internet import reactor