From de614929fdc7e63c0ae460578a02fb2d862dcfd0 Mon Sep 17 00:00:00 2001 From: Adrian Chaves Date: Sun, 9 Aug 2026 14:19:39 +0200 Subject: [PATCH 1/2] Start item-processing workers on demand instead of one per CONCURRENT_ITEMS --- scrapy/core/scraper.py | 2 +- scrapy/utils/asyncio.py | 36 ++++++++++---------- scrapy/utils/defer.py | 66 ++++++++++++++++++++++++++++--------- tests/test_utils_asyncio.py | 13 ++++++++ tests/test_utils_defer.py | 28 ++++++++++++++++ 5 files changed, 110 insertions(+), 35 deletions(-) diff --git a/scrapy/core/scraper.py b/scrapy/core/scraper.py index 58e37ce5e..012478406 100644 --- a/scrapy/core/scraper.py +++ b/scrapy/core/scraper.py @@ -117,7 +117,7 @@ class Scraper: ]: self._check_deprecated_itemproc_method(method) - self.concurrent_items: int = crawler.settings.getint("CONCURRENT_ITEMS") + self.concurrent_items: int = max(1, crawler.settings.getint("CONCURRENT_ITEMS")) self.crawler: Crawler = crawler self.signals: SignalManager = crawler.signals assert crawler.logformatter diff --git a/scrapy/utils/asyncio.py b/scrapy/utils/asyncio.py index 7c7697f56..7ff0bf44d 100644 --- a/scrapy/utils/asyncio.py +++ b/scrapy/utils/asyncio.py @@ -102,32 +102,30 @@ async def _parallel_asyncio( """Execute a callable over the objects in the given iterable, in parallel, using no more than ``count`` concurrent calls. + Tasks are created on demand, one per item, so that a *count* much larger + than the amount of work does not cost anything. + This function is only used in :meth:`scrapy.core.scraper.Scraper.handle_spider_output_async` and so it assumes that neither *callable* nor iterating *iterable* will raise an exception. """ - queue: asyncio.Queue[_T | None] = asyncio.Queue(count * 2) + semaphore = asyncio.Semaphore(count) + tasks: set[asyncio.Task[None]] = set() - async def worker() -> None: - while True: - item = await queue.get() - if item is None: - break - try: - await callable_(item, *args, **kwargs) - finally: - queue.task_done() + async def work(item: _T) -> None: + try: + await callable_(item, *args, **kwargs) + finally: + semaphore.release() - async def fill_queue() -> None: - async for item in as_async_generator(iterable): - await queue.put(item) - for _ in range(count): - await queue.put(None) - - fill_task = asyncio.create_task(fill_queue()) - work_tasks = [asyncio.create_task(worker()) for _ in range(count)] - await asyncio.wait([fill_task, *work_tasks]) + async for item in as_async_generator(iterable): + await semaphore.acquire() + task = asyncio.create_task(work(item)) + tasks.add(task) + task.add_done_callback(tasks.discard) + if tasks: + await asyncio.wait(tasks) class AsyncioLoopingCall: diff --git a/scrapy/utils/defer.py b/scrapy/utils/defer.py index 7c6235f29..f3ec03e85 100644 --- a/scrapy/utils/defer.py +++ b/scrapy/utils/defer.py @@ -149,21 +149,56 @@ def mustbe_deferred( return defer_result(result) +class _ParallelTasks: + """Consumes *work* with no more than *count* concurrent cooperative tasks. + + Tasks are started on demand: :meth:`start` must be called every time a new + item becomes available, so that a *count* much larger than the amount of + work does not cost anything. + + :attr:`finished` fires once *work* is exhausted and every task is done. + """ + + def __init__(self, work: Iterator[Any], count: int): + self._coop = Cooperator() + self._work = work + self._count = count + self._running = 0 + self.finished: Deferred[None] = Deferred() + + def start(self) -> None: + if self._running >= self._count: + return + self._running += 1 + self._coop.coiterate(self._work).addBoth(self._task_done) + + def _task_done(self, _: Any) -> None: + self._running -= 1 + # Only a running task can start another one, so once none is left no + # more work can come. + if not self._running: + self.finished.callback(None) + + def parallel( iterable: Iterable[_T], count: int, callable: Callable[Concatenate[_T, _P], _T2], # noqa: A002 *args: _P.args, **named: _P.kwargs, -) -> Deferred[list[tuple[bool, Iterator[_T2]]]]: +) -> Deferred[None]: """Execute a callable over the objects in the given iterable, in parallel, using no more than ``count`` concurrent calls. - - Taken from: https://jcalderone.livejournal.com/24285.html """ - coop = Cooperator() - work: Iterator[_T2] = (callable(elem, *args, **named) for elem in iterable) - return DeferredList([coop.coiterate(work) for _ in range(count)]) + + def work() -> Iterator[_T2]: + for elem in iterable: + tasks.start() + yield callable(elem, *args, **named) + + tasks = _ParallelTasks(work(), count) + tasks.start() + return tasks.finished class _AsyncCooperatorAdapter(Iterator[Deferred[Any]], Generic[_T]): @@ -226,12 +261,16 @@ class _AsyncCooperatorAdapter(Iterator[Deferred[Any]], Generic[_T]): self.finished: bool = False self.waiting_deferreds: deque[Deferred[Any]] = deque() self.anext_deferred: Deferred[_T] | None = None + # Called whenever aiterator produces a value, so that parallel_async() + # can start a task for it. + self.on_value: Callable[[], None] = lambda: None def _callback(self, result: _T) -> None: # This gets called when the result from aiterator.__anext__() is available. # It calls the callable on it and sends the result to the oldest waiting Deferred # (by chaining if the result is a Deferred too or by firing if not). self.anext_deferred = None + self.on_value() callable_result = self.callable( result, *self.callable_args, **self.callable_kwargs ) @@ -276,16 +315,13 @@ def parallel_async( callable: Callable[Concatenate[_T, _P], Deferred[Any] | None], # noqa: A002 *args: _P.args, **named: _P.kwargs, -) -> Deferred[list[tuple[bool, Iterator[Deferred[Any]]]]]: +) -> Deferred[None]: """Like ``parallel`` but for async iterators""" - coop = Cooperator() - work: Iterator[Deferred[Any]] = _AsyncCooperatorAdapter( - async_iterable, callable, *args, **named - ) - dl: Deferred[list[tuple[bool, Iterator[Deferred[Any]]]]] = DeferredList( - [coop.coiterate(work) for _ in range(count)] - ) - return dl + work = _AsyncCooperatorAdapter(async_iterable, callable, *args, **named) + tasks = _ParallelTasks(work, count) + work.on_value = tasks.start + tasks.start() + return tasks.finished def process_chain( diff --git a/tests/test_utils_asyncio.py b/tests/test_utils_asyncio.py index 4bd54acd4..59fa17594 100644 --- a/tests/test_utils_asyncio.py +++ b/tests/test_utils_asyncio.py @@ -115,6 +115,19 @@ class TestParallelAsyncio: assert parallel_count[0] == 0 assert max_parallel_count[0] <= self.CONCURRENT_ITEMS + @coroutine_test + async def test_count_higher_than_work(self): + results: list[int] = [] + task_counts: list[int] = [] + + async def callable_(o: int) -> None: + task_counts.append(len(asyncio.all_tasks())) + results.append(o) + + await _parallel_asyncio(range(3), 1_000_000, callable_) + assert results == [0, 1, 2] + assert max(task_counts) < 100 + @pytest.mark.only_asyncio class TestAsyncioLoopingCall: diff --git a/tests/test_utils_defer.py b/tests/test_utils_defer.py index 175a4fe03..bb174e66f 100644 --- a/tests/test_utils_defer.py +++ b/tests/test_utils_defer.py @@ -4,9 +4,11 @@ import asyncio import random from asyncio import Future from typing import TYPE_CHECKING, Any +from unittest import mock import pytest from twisted.internet.defer import Deferred, inlineCallbacks +from twisted.internet.task import Cooperator from scrapy.utils.asyncgen import as_async_generator, collect_asyncgen from scrapy.utils.defer import ( @@ -17,6 +19,7 @@ from scrapy.utils.defer import ( iter_errback, maybe_deferred_to_future, mustbe_deferred, + parallel, parallel_async, ) from tests.utils.decorators import coroutine_test, inline_callbacks_test @@ -129,6 +132,20 @@ class TestAsyncDefTestsuite: raise RuntimeError("This is expected to be raised") +@pytest.mark.requires_reactor # parallel() requires a reactor +class TestParallel: + @inline_callbacks_test + def test_count_higher_than_work(self) -> Generator[Deferred[Any], Any, None]: + results: list[int] = [] + with mock.patch.object( + Cooperator, "coiterate", autospec=True, side_effect=Cooperator.coiterate + ) as coiterate: + yield parallel(range(3), 1_000_000, results.append) + assert results == [0, 1, 2] + # One task per item, plus the one that finds no more work. + assert coiterate.call_count <= 4 + + @pytest.mark.requires_reactor # parallel_async() requires a reactor class TestParallelAsync: """This tests _AsyncCooperatorAdapter by testing parallel_async which is its only usage. @@ -239,6 +256,17 @@ class TestParallelAsync: assert parallel_count[0] == 0 assert max_parallel_count[0] <= self.CONCURRENT_ITEMS, max_parallel_count[0] + @inline_callbacks_test + def test_count_higher_than_work(self) -> Generator[Deferred[Any], Any, None]: + results: list[int] = [] + with mock.patch.object( + Cooperator, "coiterate", autospec=True, side_effect=Cooperator.coiterate + ) as coiterate: + yield parallel_async(self.get_async_iterable(3), 1_000_000, results.append) + assert sorted(results) == [0, 1, 2] + # One task per item, plus the one that finds no more work. + assert coiterate.call_count <= 4 + class TestDeferredFromCoro: def test_deferred(self): From b777ebb881e42b313141072c43a43162910e3a16 Mon Sep 17 00:00:00 2001 From: Adrian Chaves Date: Sun, 9 Aug 2026 16:22:04 +0200 Subject: [PATCH 2/2] Add a timeout to tests --- .github/workflows/tests-ubuntu.yml | 5 ++++- tox.ini | 1 + 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/.github/workflows/tests-ubuntu.yml b/.github/workflows/tests-ubuntu.yml index cd726a2fe..318ef6b24 100644 --- a/.github/workflows/tests-ubuntu.yml +++ b/.github/workflows/tests-ubuntu.yml @@ -18,8 +18,11 @@ jobs: tests: name: tests (${{ matrix.python-version }}, ${{ matrix.env.TOXENV }}) runs-on: ubuntu-latest + timeout-minutes: 30 env: - PYTEST_ADDOPTS: ${{ matrix.coverage && '-n auto' || '-n auto --no-cov' }} + # A hanging test would otherwise keep the job running until GitHub kills + # it, without telling which test hung. + PYTEST_ADDOPTS: --timeout=120 ${{ matrix.coverage && '-n auto' || '-n auto --no-cov' }} # Make uv use the interpreter that actions/setup-python installed instead # of downloading one of its own. UV_PYTHON_PREFERENCE: only-system diff --git a/tox.ini b/tox.ini index 94c04f75f..bf5c2660f 100644 --- a/tox.ini +++ b/tox.ini @@ -44,6 +44,7 @@ deps = pygments pytest pytest-cov >= 7.0.0 + pytest-timeout pytest-xdist sybil >= 1.3.0 # https://github.com/cjw296/sybil/issues/20#issuecomment-605433422 pytest-twisted >= 1.14.3