From fee20b785866dc64bd5f1cff5fa2c50416332e50 Mon Sep 17 00:00:00 2001 From: Andrey Rakhmatullin Date: Fri, 27 Mar 2026 16:21:25 +0500 Subject: [PATCH] AsyncCrawlerProcess._start_asyncio() improvements. (#7366) --- scrapy/crawler.py | 169 ++++++++++++++++-- .../reactorless_sleeping.py | 20 +++ tests/test_crawler_subprocess.py | 41 +++-- 3 files changed, 199 insertions(+), 31 deletions(-) create mode 100644 tests/AsyncCrawlerProcess/reactorless_sleeping.py diff --git a/scrapy/crawler.py b/scrapy/crawler.py index 2e07a3b87..2ae45fe40 100644 --- a/scrapy/crawler.py +++ b/scrapy/crawler.py @@ -605,22 +605,30 @@ class CrawlerProcessBase(CrawlerRunnerBase): from twisted.internet import reactor install_shutdown_handlers(self._signal_kill) - signame = signal_names[signum] - logger.info( - "Received %(signame)s, shutting down gracefully. Send again to force ", - {"signame": signame}, - ) + self._log_shutdown(signum) reactor.callFromThread(self._graceful_stop_reactor) def _signal_kill(self, signum: int, _: Any) -> None: from twisted.internet import reactor install_shutdown_handlers(signal.SIG_IGN) + self._log_kill(signum) + reactor.callFromThread(self._stop_reactor) + + @staticmethod + def _log_shutdown(signum: int) -> None: + signame = signal_names[signum] + logger.info( + "Received %(signame)s, shutting down gracefully. Send again to force ", + {"signame": signame}, + ) + + @staticmethod + def _log_kill(signum: int) -> None: signame = signal_names[signum] logger.info( "Received %(signame)s twice, forcing unclean shutdown", {"signame": signame} ) - reactor.callFromThread(self._stop_reactor) def _setup_reactor(self, install_signal_handlers: bool) -> None: from twisted.internet import reactor @@ -767,6 +775,7 @@ class AsyncCrawlerProcess(CrawlerProcessBase, AsyncCrawlerRunner): ): super().__init__(settings, install_root_handler) logger.debug("Using AsyncCrawlerProcess") + self._reactorless_loop: asyncio.AbstractEventLoop | None = None # We want the asyncio event loop to be installed early, so that it's # always the correct one. And as we do that, we can also install the # reactor here. @@ -778,7 +787,7 @@ class AsyncCrawlerProcess(CrawlerProcessBase, AsyncCrawlerRunner): raise RuntimeError( "TWISTED_ENABLED is False but a Twisted reactor is installed." ) - set_asyncio_event_loop(loop_path) + self._reactorless_loop = set_asyncio_event_loop(loop_path) install_reactor_import_hook() elif is_reactor_installed(): # The user could install a reactor before this class is instantiated. @@ -790,6 +799,7 @@ class AsyncCrawlerProcess(CrawlerProcessBase, AsyncCrawlerRunner): else: install_reactor(_asyncio_reactor_path, loop_path) self._initialized_reactor = True + self._reactorless_main_task: asyncio.Future[None] | None = None def _stop_dfd(self) -> Deferred[Any]: return deferred_from_coro(self.stop()) @@ -820,20 +830,149 @@ class AsyncCrawlerProcess(CrawlerProcessBase, AsyncCrawlerRunner): def _start_asyncio( self, stop_after_crawl: bool, install_signal_handlers: bool ) -> None: - # Very basic and will need multiple improvements. - # TODO https://docs.python.org/3/library/asyncio-runner.html#handling-keyboard-interruption - # TODO various exception handling - # TODO consider asyncio.run() + # We cannot use asyncio.run() here, because we can't let it handle the + # loop lifetime: _crawl() needs a loop (which we create in __init__()), + # because crawl() returns a Task. + # So we reproduce a part of asyncio.runners.Runner that is useful to us. + + # Normal workflow: + # 1. _start_asyncio() creates a task for self.join() and calls _run_loop() + # 2. _run_loop() calls loop.run_until_complete(main_task) + # 3. Crawling tasks start and finish + # 4. join() completes, loop.run_until_complete() and thus _run_loop() return + # 5. _start_asyncio() calls _close_loop() + # 6. _close_loop() does finalization and calls loop.close() + + # Normal workflow with stop_after_crawl=False: + # 1. _start_asyncio() creates a simple future and calls _run_loop() + # 2. _run_loop() calls loop.run_until_complete(main_task) + # 3. Crawling tasks start and finish + # 4. _run_loop() blocks until the loop is stopped externally or the + # future is cancelled via Ctrl-C + # 5. (after _run_loop() returns) _start_asyncio() calls _close_loop() + # 6. _close_loop() does finalization and calls loop.close() + + # Workflow with Ctrl-C pressed once: + # 1. While loop.run_until_complete() blocks, _signal_shutdown_reactorless() + # is called + # 2. _signal_shutdown_reactorless() calls _shutdown_graceful_reactorless() + # (via call_soon_threadsafe()) + # 3. _shutdown_graceful_reactorless() calls stop() + # 4. For stop_after_crawl=True: crawl tasks finish, join() completes, + # loop.run_until_complete() and thus _run_loop() return + # For stop_after_crawl=False: _shutdown_graceful_reactorless() waits + # for crawl tasks via join(), then cancels the main task, + # loop.run_until_complete() raises CancelledError, _run_loop() returns + # 5. _start_asyncio() calls _close_loop() + # 6. _close_loop() does finalization and calls loop.close() + + # Workflow with Ctrl-C pressed twice: + # 1. While loop.run_until_complete() blocks, _signal_shutdown_reactorless() + # is called + # 2. _signal_shutdown_reactorless() calls _shutdown_graceful_reactorless() + # (via call_soon_threadsafe()) and installs _signal_kill_reactorless() + # as the next handler + # 3. Before _shutdown_graceful_reactorless() completes, + # _signal_kill_reactorless() is called + # 4. _signal_kill_reactorless() cancels the main task + # (via call_soon_threadsafe()) + # 5. loop.run_until_complete() raises CancelledError, _run_loop() returns + # 6. _start_asyncio() calls _close_loop() + # 7. _close_loop() cancels all pending tasks (including + # _shutdown_graceful_reactorless()), does finalization and calls loop.close() + + loop = self._reactorless_loop + assert loop - loop = asyncio.get_event_loop() if stop_after_crawl: - join_task = loop.create_task(self.join()) - join_task.add_done_callback(lambda _: loop.stop()) + self._reactorless_main_task = loop.create_task(self.join()) + else: + self._reactorless_main_task = loop.create_future() + self._stop_after_crawl = stop_after_crawl + try: - loop.run_forever() # blocking call + self._run_loop(install_signal_handlers) # blocking call + except asyncio.CancelledError: + pass finally: + self._close_loop() + + def _run_loop(self, install_signal_handlers: bool) -> None: + # similar to asyncio.runners.Runner.run() + if install_signal_handlers: + install_shutdown_handlers(self._signal_shutdown_reactorless) + assert self._reactorless_loop + assert self._reactorless_main_task + self._reactorless_loop.run_until_complete(self._reactorless_main_task) + + def _close_loop(self) -> None: + # Similar to asyncio.runners.Runner.close() + loop = self._reactorless_loop + assert loop + try: + self._cancel_all_tasks(loop) loop.run_until_complete(loop.shutdown_asyncgens()) + loop.run_until_complete(loop.shutdown_default_executor()) + finally: + self._reactorless_main_task = None + asyncio.set_event_loop(None) loop.close() + self._reactorless_loop = None + + @staticmethod + def _cancel_all_tasks(loop: asyncio.AbstractEventLoop) -> None: + # copy of asyncio.runners._cancel_all_tasks() + to_cancel = asyncio.all_tasks(loop) + if not to_cancel: + return + + for task in to_cancel: + task.cancel() + + loop.run_until_complete(asyncio.gather(*to_cancel, return_exceptions=True)) + + for task in to_cancel: + if task.cancelled(): + continue + if task.exception() is not None: + loop.call_exception_handler( + { + "message": "unhandled exception during AsyncCrawlerProcess shutdown", + "exception": task.exception(), + "task": task, + } + ) + + def _signal_shutdown_reactorless(self, signum: int, _: Any) -> None: + install_shutdown_handlers(self._signal_kill_reactorless) + self._log_shutdown(signum) + if (loop := self._reactorless_loop) is None: + return + + def _create_shutdown_task() -> None: + coro = self._shutdown_graceful_reactorless() + try: + loop.create_task(coro) + except RuntimeError: + coro.close() + + loop.call_soon_threadsafe(_create_shutdown_task) + + async def _shutdown_graceful_reactorless(self) -> None: + await self.stop() + if not self._stop_after_crawl: + # wait until crawl tasks finish and cancel the future + await self.join() + if self._reactorless_main_task and not self._reactorless_main_task.done(): + self._reactorless_main_task.cancel() + + def _signal_kill_reactorless(self, signum: int, _: Any) -> None: + install_shutdown_handlers(signal.SIG_IGN) + self._log_kill(signum) + if (loop := self._reactorless_loop) is None: + return + if (task := self._reactorless_main_task) is not None: + loop.call_soon_threadsafe(task.cancel) def _start_twisted( self, stop_after_crawl: bool, install_signal_handlers: bool diff --git a/tests/AsyncCrawlerProcess/reactorless_sleeping.py b/tests/AsyncCrawlerProcess/reactorless_sleeping.py new file mode 100644 index 000000000..6bfe64686 --- /dev/null +++ b/tests/AsyncCrawlerProcess/reactorless_sleeping.py @@ -0,0 +1,20 @@ +import asyncio +import sys + +import scrapy +from scrapy.crawler import AsyncCrawlerProcess + + +class SleepingSpider(scrapy.Spider): + name = "sleeping" + + start_urls = ["data:,;"] + + async def parse(self, response): + await asyncio.sleep(int(sys.argv[1])) + + +process = AsyncCrawlerProcess(settings={"TWISTED_ENABLED": False}) + +process.crawl(SleepingSpider) +process.start() diff --git a/tests/test_crawler_subprocess.py b/tests/test_crawler_subprocess.py index 8d7b8f871..b07ddbe09 100644 --- a/tests/test_crawler_subprocess.py +++ b/tests/test_crawler_subprocess.py @@ -12,12 +12,10 @@ from typing import TYPE_CHECKING import pytest from packaging.version import parse as parse_version from pexpect.popen_spawn import PopenSpawn -from twisted.internet.defer import Deferred from w3lib import __version__ as w3lib_version -from scrapy.utils.asyncio import call_later -from tests.utils import get_script_run_env -from tests.utils.decorators import inline_callbacks_test +from tests.utils import async_sleep, get_script_run_env +from tests.utils.decorators import coroutine_test if TYPE_CHECKING: from tests.mockserver.http import MockServer @@ -207,33 +205,37 @@ class TestCrawlerProcessSubprocessBase(ScriptRunnerMixin): assert "Spider closed (finished)" in log assert "The value of FOO is 42" in log - def test_shutdown_graceful(self): - sig = signal.SIGINT if sys.platform != "win32" else signal.SIGBREAK - args = self.get_script_args("sleeping.py", "3") + def _test_shutdown_graceful(self, script: str = "sleeping.py") -> None: + sig = signal.SIGINT if sys.platform != "win32" else signal.SIGBREAK # type: ignore[attr-defined] + args = self.get_script_args(script, "3") p = PopenSpawn(args, timeout=5, env=get_script_run_env()) p.expect_exact("Spider opened") p.expect_exact("Crawled (200)") p.kill(sig) p.expect_exact("shutting down gracefully") p.expect_exact("Spider closed (shutdown)") - p.wait() + p.wait() # type: ignore[no-untyped-call] - @inline_callbacks_test - def test_shutdown_forced(self): - sig = signal.SIGINT if sys.platform != "win32" else signal.SIGBREAK - args = self.get_script_args("sleeping.py", "10") + def test_shutdown_graceful(self) -> None: + self._test_shutdown_graceful() + + async def _test_shutdown_forced(self, script: str = "sleeping.py") -> None: + sig = signal.SIGINT if sys.platform != "win32" else signal.SIGBREAK # type: ignore[attr-defined] + args = self.get_script_args(script, "10") p = PopenSpawn(args, timeout=5, env=get_script_run_env()) p.expect_exact("Spider opened") p.expect_exact("Crawled (200)") p.kill(sig) p.expect_exact("shutting down gracefully") # sending the second signal too fast often causes problems - d = Deferred() - call_later(0.01, d.callback, None) - yield d + await async_sleep(0.01) p.kill(sig) p.expect_exact("forcing unclean shutdown") - p.wait() + p.wait() # type: ignore[no-untyped-call] + + @coroutine_test + async def test_shutdown_forced(self) -> None: + await self._test_shutdown_forced() class TestCrawlerProcessSubprocess(TestCrawlerProcessSubprocessBase): @@ -397,6 +399,13 @@ class TestAsyncCrawlerProcessSubprocess(TestCrawlerProcessSubprocessBase): in log ) + def test_shutdown_graceful(self) -> None: + self._test_shutdown_graceful("reactorless_sleeping.py") + + @coroutine_test + async def test_shutdown_forced(self) -> None: + await self._test_shutdown_forced("reactorless_sleeping.py") + class TestCrawlerRunnerSubprocessBase(ScriptRunnerMixin): """Common tests between CrawlerRunner and AsyncCrawlerRunner,