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

Re-implement idle support

This commit is contained in:
Adrián Chaves 2025-03-31 15:18:38 +02:00
parent a04e85e08b
commit 5c9a931e8b
9 changed files with 178 additions and 111 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
docs/news.rst
View File

@ -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 <start-requests-lazy>`.
:ref:`start-requests` for details.
To restore the previous behavior, :ref:`use lazy start request scheduling
<start-requests-lazy>`.
- 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 <scrapy.signalmanager.SignalManager>` has a new
:meth:`~scrapy.signalmanager.SignalManager.wait_for` method.
- Added official support for :ref:`idle <start-requests-idle>` and
:ref:`front load <start-requests-front-load>` start request scheduling.
- Added a new :signal:`scheduler_empty` signal.
- Exposed a new method of :class:`Crawler.engine
<scrapy.core.engine.ExecutionEngine>`:
: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 <scrapy.signalmanager.SignalManager>` 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
~~~~~~~~~

2
docs/topics/api.rst
View File

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

4
docs/topics/optimize.rst
View File

@ -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 <start-requests>`, use :ref:`lazy
scheduling <start-requests-lazy>`.
<start-requests-lazy>` or :ref:`idle <start-requests-idle>` scheduling.
- Set :setting:`JOBDIR` to offload all scheduled requests to disk.

23
docs/topics/signals.rst
View File

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

41
docs/topics/spiders.rst
View File

@ -466,17 +466,21 @@ while yielding your start requests:
self.crawler.engine.scheduler.unpause()
.. note:: If you use a :ref:`custom scheduler <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 <optimize-memory>`, it can sometimes be
useful to delay start request scheduling :ref:`until the scheduler is empty
<start-requests-lazy>` or :ref:`until the engine is idle
<start-requests-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 <topics-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 <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:

11
scrapy/core/engine.py
View File

@ -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
<start-requests-lazy>`.
"""
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"""

3
scrapy/signalmanager.py
View File

@ -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
<start-requests-lazy>`.
"""
d = Deferred()

17
scrapy/signals.py
View File

@ -7,7 +7,24 @@ signals here without documenting them there.
engine_started = object()
engine_stopped = object()
#: Sent when the :ref:`engine <engine>` is waiting for :meth:`Spider.start
#: <scrapy.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
#: <start-requests-idle>`.
spider_start_blocking = object()
#: Sent whenever the :ref:`engine <engine>` asks for a pending request from the
#: :ref:`scheduler <topics-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
#: <start-requests-lazy>`.
scheduler_empty = object()
spider_opened = object()
spider_idle = object()
spider_closed = object()

154
tests/test_engine_loop.py
View File

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