Add _parallel_asyncio(). (#6852)

scrapy/core/scraper.py

@@ -21,6 +21,7 @@ from scrapy.exceptions import (
     ScrapyDeprecationWarning,
 )
 from scrapy.http import Request, Response
+from scrapy.utils.asyncio import _parallel_asyncio, is_asyncio_available
 from scrapy.utils.defer import (
     _defer_sleep,
     aiter_errback,
@@ -328,11 +329,21 @@ class Scraper:
         response: Response,
     ) -> None:
         """Pass items/requests produced by a callback to ``_process_spidermw_output()`` in parallel."""
+        it: Iterable[_T] | AsyncIterator[_T]
+        if is_asyncio_available():
+            if isinstance(result, AsyncIterator):
+                it = aiter_errback(result, self.handle_spider_error, request, response)
+            else:
+                it = iter_errback(result, self.handle_spider_error, request, response)
+            await _parallel_asyncio(
+                it, self.concurrent_items, self._process_spidermw_output_async, response
+            )
+            return
         if isinstance(result, AsyncIterator):
-            ait = aiter_errback(result, self.handle_spider_error, request, response)
+            it = aiter_errback(result, self.handle_spider_error, request, response)
             await maybe_deferred_to_future(
                 parallel_async(
-                    ait,
+                    it,
                     self.concurrent_items,
                     self._process_spidermw_output,
                     response,
@@ -349,8 +360,19 @@ class Scraper:
             )
         )
 
-    @deferred_f_from_coro_f
-    async def _process_spidermw_output(self, output: Any, response: Response) -> None:
+    def _process_spidermw_output(
+        self, output: Any, response: Response
+    ) -> Deferred[None]:
+        """Process each Request/Item (given in the output parameter) returned
+        from the given spider.
+
+        Items are sent to the item pipelines, requests are scheduled.
+        """
+        return deferred_from_coro(self._process_spidermw_output_async(output, response))
+
+    async def _process_spidermw_output_async(
+        self, output: Any, response: Response
+    ) -> None:
         """Process each Request/Item (given in the output parameter) returned
         from the given spider.
 

scrapy/utils/asyncio.py

@@ -1,7 +1,23 @@
 """Utilities related to asyncio and its support in Scrapy."""
 
+from __future__ import annotations
+
+import asyncio
+from typing import TYPE_CHECKING, Any, TypeVar
+
+from scrapy.utils.asyncgen import as_async_generator
 from scrapy.utils.reactor import is_asyncio_reactor_installed, is_reactor_installed
 
+if TYPE_CHECKING:
+    from collections.abc import AsyncIterator, Callable, Coroutine, Iterable
+
+    # typing.Concatenate and typing.ParamSpec require Python 3.10
+    from typing_extensions import Concatenate, ParamSpec
+
+    _P = ParamSpec("_P")
+
+_T = TypeVar("_T")
+
 
 def is_asyncio_available() -> bool:
     """Check if it's possible to call asyncio code that relies on the asyncio event loop.
@@ -36,3 +52,41 @@ def is_asyncio_available() -> bool:
         )
 
     return is_asyncio_reactor_installed()
+
+
+async def _parallel_asyncio(
+    iterable: Iterable[_T] | AsyncIterator[_T],
+    count: int,
+    callable: Callable[Concatenate[_T, _P], Coroutine[Any, Any, None]],
+    *args: _P.args,
+    **kwargs: _P.kwargs,
+) -> None:
+    """Execute a callable over the objects in the given iterable, in parallel,
+    using no more than ``count`` concurrent calls.
+
+    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()
+
+    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 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])

tests/test_utils_asyncio.py

@@ -1,6 +1,18 @@
-import pytest
+from __future__ import annotations
 
-from scrapy.utils.asyncio import is_asyncio_available
+import asyncio
+import random
+from typing import TYPE_CHECKING
+
+import pytest
+from twisted.trial import unittest
+
+from scrapy.utils.asyncgen import as_async_generator
+from scrapy.utils.asyncio import _parallel_asyncio, is_asyncio_available
+from scrapy.utils.defer import deferred_f_from_coro_f
+
+if TYPE_CHECKING:
+    from collections.abc import AsyncGenerator
 
 
 @pytest.mark.usefixtures("reactor_pytest")
@@ -8,3 +20,80 @@ class TestAsyncio:
     def test_is_asyncio_available(self):
         # the result should depend only on the pytest --reactor argument
         assert is_asyncio_available() == (self.reactor_pytest != "default")
+
+
+@pytest.mark.only_asyncio
+class TestParallelAsyncio(unittest.TestCase):
+    """Test for scrapy.utils.asyncio.parallel_asyncio(), based on tests.test_utils_defer.TestParallelAsync."""
+
+    CONCURRENT_ITEMS = 50
+
+    @staticmethod
+    async def callable(o: int, results: list[int]) -> None:
+        if random.random() < 0.4:
+            # simulate async processing
+            await asyncio.sleep(random.random() / 8)
+        # simulate trivial sync processing
+        results.append(o)
+
+    async def callable_wrapped(
+        self,
+        o: int,
+        results: list[int],
+        parallel_count: list[int],
+        max_parallel_count: list[int],
+    ) -> None:
+        parallel_count[0] += 1
+        max_parallel_count[0] = max(max_parallel_count[0], parallel_count[0])
+        await self.callable(o, results)
+        assert parallel_count[0] > 0, parallel_count[0]
+        parallel_count[0] -= 1
+
+    @staticmethod
+    def get_async_iterable(length: int) -> AsyncGenerator[int, None]:
+        # simulate a simple callback without delays between results
+        return as_async_generator(range(length))
+
+    @staticmethod
+    async def get_async_iterable_with_delays(length: int) -> AsyncGenerator[int, None]:
+        # simulate a callback with delays between some of the results
+        for i in range(length):
+            if random.random() < 0.1:
+                await asyncio.sleep(random.random() / 20)
+            yield i
+
+    @deferred_f_from_coro_f
+    async def test_simple(self):
+        for length in [20, 50, 100]:
+            parallel_count = [0]
+            max_parallel_count = [0]
+            results = []
+            ait = self.get_async_iterable(length)
+            await _parallel_asyncio(
+                ait,
+                self.CONCURRENT_ITEMS,
+                self.callable_wrapped,
+                results,
+                parallel_count,
+                max_parallel_count,
+            )
+            assert list(range(length)) == sorted(results)
+            assert max_parallel_count[0] <= self.CONCURRENT_ITEMS
+
+    @deferred_f_from_coro_f
+    async def test_delays(self):
+        for length in [20, 50, 100]:
+            parallel_count = [0]
+            max_parallel_count = [0]
+            results = []
+            ait = self.get_async_iterable_with_delays(length)
+            await _parallel_asyncio(
+                ait,
+                self.CONCURRENT_ITEMS,
+                self.callable_wrapped,
+                results,
+                parallel_count,
+                max_parallel_count,
+            )
+            assert list(range(length)) == sorted(results)
+            assert max_parallel_count[0] <= self.CONCURRENT_ITEMS

tests/test_utils_defer.py

@@ -164,7 +164,7 @@ class TestAsyncDefTestsuite(unittest.TestCase):
         raise RuntimeError("This is expected to be raised")
 
 
-class TestAsyncCooperator(unittest.TestCase):
+class TestParallelAsync(unittest.TestCase):
     """This tests _AsyncCooperatorAdapter by testing parallel_async which is its only usage.
 
     parallel_async is called with the results of a callback (so an iterable of items, requests and None,
@@ -194,6 +194,27 @@ class TestAsyncCooperator(unittest.TestCase):
         results.append(o)
         return None
 
+    def callable_wrapped(
+        self,
+        o: int,
+        results: list[int],
+        parallel_count: list[int],
+        max_parallel_count: list[int],
+    ) -> Deferred[None] | None:
+        parallel_count[0] += 1
+        max_parallel_count[0] = max(max_parallel_count[0], parallel_count[0])
+        dfd = self.callable(o, results)
+
+        def decrement(_: Any = None) -> None:
+            assert parallel_count[0] > 0, parallel_count[0]
+            parallel_count[0] -= 1
+
+        if dfd is not None:
+            dfd.addBoth(decrement)
+        else:
+            decrement()
+        return dfd
+
     @staticmethod
     def get_async_iterable(length: int) -> AsyncGenerator[int, None]:
         # simulate a simple callback without delays between results
@@ -215,20 +236,42 @@ class TestAsyncCooperator(unittest.TestCase):
     @inlineCallbacks
     def test_simple(self):
         for length in [20, 50, 100]:
+            parallel_count = [0]
+            max_parallel_count = [0]
             results = []
             ait = self.get_async_iterable(length)
-            dl = parallel_async(ait, self.CONCURRENT_ITEMS, self.callable, results)
+            dl = parallel_async(
+                ait,
+                self.CONCURRENT_ITEMS,
+                self.callable_wrapped,
+                results,
+                parallel_count,
+                max_parallel_count,
+            )
             yield dl
             assert list(range(length)) == sorted(results)
+            assert parallel_count[0] == 0
+            assert max_parallel_count[0] <= self.CONCURRENT_ITEMS, max_parallel_count[0]
 
     @inlineCallbacks
     def test_delays(self):
         for length in [20, 50, 100]:
+            parallel_count = [0]
+            max_parallel_count = [0]
             results = []
             ait = self.get_async_iterable_with_delays(length)
-            dl = parallel_async(ait, self.CONCURRENT_ITEMS, self.callable, results)
+            dl = parallel_async(
+                ait,
+                self.CONCURRENT_ITEMS,
+                self.callable_wrapped,
+                results,
+                parallel_count,
+                max_parallel_count,
+            )
             yield dl
             assert list(range(length)) == sorted(results)
+            assert parallel_count[0] == 0
+            assert max_parallel_count[0] <= self.CONCURRENT_ITEMS, max_parallel_count[0]
 
 
 class TestDeferredFromCoro(unittest.TestCase):