Add parallel_async.

scrapy/core/scraper.py

@@ -1,6 +1,6 @@
 """This module implements the Scraper component which parses responses and
 extracts information from them"""
-
+import collections
 import logging
 from collections import deque
 
@@ -12,7 +12,16 @@ from scrapy import signals
 from scrapy.core.spidermw import SpiderMiddlewareManager
 from scrapy.exceptions import CloseSpider, DropItem, IgnoreRequest
 from scrapy.http import Request, Response
-from scrapy.utils.defer import defer_fail, defer_succeed, iter_errback, parallel
+from scrapy.utils.defer import (
+    aiter_errback,
+    defer_fail,
+    defer_succeed,
+    deferred_from_coro,
+    iter_errback,
+    parallel,
+    parallel_async,
+)
+
 from scrapy.utils.log import failure_to_exc_info, logformatter_adapter
 from scrapy.utils.misc import load_object, warn_on_generator_with_return_value
 from scrapy.utils.spider import iterate_spider_output
@@ -180,9 +189,14 @@ class Scraper:
     def handle_spider_output(self, result, request, response, spider):
         if not result:
             return defer_succeed(None)
-        it = iter_errback(result, self.handle_spider_error, request, response, spider)
-        dfd = parallel(it, self.concurrent_items, self._process_spidermw_output,
-                       request, response, spider)
+        if isinstance(result, collections.abc.AsyncIterable):
+            it = aiter_errback(result, self.handle_spider_error, request, response, spider)
+            dfd = deferred_from_coro(parallel_async(it, self.concurrent_items, self._process_spidermw_output,
+                                     request, response, spider))
+        else:
+            it = iter_errback(result, self.handle_spider_error, request, response, spider)
+            dfd = parallel(it, self.concurrent_items, self._process_spidermw_output,
+                           request, response, spider)
         return dfd
 
     def _process_spidermw_output(self, output, request, response, spider):

scrapy/utils/defer.py

@@ -75,6 +75,116 @@ def parallel(iterable, count, callable, *args, **named):
     return defer.DeferredList([coop.coiterate(work) for _ in range(count)])
 
 
+class _AsyncCooperatorAdapter:
+    """ A class that wraps an async iterator into a normal iterator suitable
+    for using in Cooperator.coiterate(). As it's only needed for parallel_async(),
+    it calls the callable directly in the callback, instead of providing a more
+    generic interface.
+
+    On the outside, this class behaves as an iterator that yields Deferreds.
+    Each Deferred is fired with the result of the callable which was called on
+    the next result from aiterator. It raises StopIteration when aiterator is
+    exhausted, as expected.
+
+    Cooperator calls __next__() multiple times and waits on the Deferreds
+    returned from it. As async generators (since Python 3.8) don't support
+    awaiting on __anext__() several times in parallel, we need to serialize
+    this. It's done by storing the Deferreds returned from __next__() and
+    firing the oldest one when a result from __anext__() is available.
+
+    The workflow:
+    1. When __next__() is called for the first time, it creates a Deferred, stores it
+    in self.waiting_deferreds and returns it. It also makes a Deferred that will wait
+    for self.aiterator.__anext__() and puts it into self.anext_deferred.
+    2. If __next__() is called again before self.anext_deferred fires, more Deferreds
+    are added to self.waiting_deferreds.
+    3. When self.anext_deferred fires, it either calls _callback() or _errback(). Both
+    clear self.anext_deferred.
+    3.1. _callback() calls the callable passing the result value that it takes, pops a
+    Deferred from self.waiting_deferreds, and if the callable result was a Deferred, it
+    chains those Deferreds so that the waiting Deferred will fire when the result
+    Deferred does, otherwise it fires it directly. This causes one awaiting task to
+    receive a result. If self.waiting_deferreds is still not empty, new __anext__() is
+    called and self.anext_deferred is populated.
+    3.2. _errback() checks the exception class. If it's StopAsyncIteration it means
+    self.aiterator is exhausted and so it sets self.finished and fires all
+    self.waiting_deferreds. Other exceptions are propagated.
+    4. If __next__() is called after __anext__() was handled, then if self.finished is
+    True, it raises StopIteration, otherwise it acts like in step 2, but if
+    self.anext_deferred is now empty is also populates it with a new __anext__().
+
+    Note that CooperativeTask ignores the value returned from the Deferred that it waits
+    for, so we fire them with None when needed.
+
+    It may be possible to write an async iterator-aware replacement for
+    Cooperator/CooperativeTask and use it instead of this adapter to achieve the same
+    goal.
+    """
+    def __init__(self, aiterator, callable, *callable_args, **callable_kwargs):
+        self.aiterator = aiterator
+        self.callable = callable
+        self.callable_args = callable_args
+        self.callable_kwargs = callable_kwargs
+        self.finished = False
+        self.waiting_deferreds = []
+        self.anext_deferred = None
+
+    def _callback(self, result):
+        # 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
+        result = self.callable(result, *self.callable_args, **self.callable_kwargs)
+        d = self.waiting_deferreds.pop(0)
+        if d.called:
+            raise ValueError('Deferred in waiting_deferreds already called')
+        if isinstance(result, defer.Deferred):
+            result.chainDeferred(d)
+        else:
+            d.callback(None)
+        if self.waiting_deferreds:
+            self._call_anext()
+
+    def _errback(self, failure):
+        # This gets called on any exceptions in aiterator.__anext__().
+        # It handles StopAsyncIteration by stopping the iteration and reraises all others.
+        self.anext_deferred = None
+        failure.trap(StopAsyncIteration)
+        self.finished = True
+        for d in self.waiting_deferreds:
+            if d.called:
+                raise ValueError('Deferred in waiting_deferreds already called')
+            d.callback(None)
+
+    def _call_anext(self):
+        # This starts waiting for the next result from aiterator.
+        # If aiterator is exhausted, _errback will be called.
+        self.anext_deferred = deferred_from_coro(self.aiterator.__anext__())
+        self.anext_deferred.addCallbacks(self._callback, self._errback)
+
+    def __iter__(self):
+        return self
+
+    def __next__(self):
+        # This puts a new Deferred into self.waiting_deferreds and returns it.
+        # It also calls __anext__() if needed.
+        if self.finished:
+            raise StopIteration
+        d = defer.Deferred()
+        self.waiting_deferreds.append(d)
+        if not self.anext_deferred:
+            self._call_anext()
+        return d
+
+
+def parallel_async(async_iterable, count, callable, *args, **named):
+    """ Like parallel but for async iterables """
+    coop = task.Cooperator()
+    work = _AsyncCooperatorAdapter(async_iterable, callable, *args, **named)
+    dl = defer.DeferredList([coop.coiterate(work) for _ in range(count)])
+    return dl
+
+
 def process_chain(callbacks, input, *a, **kw):
     """Return a Deferred built by chaining the given callbacks"""
     d = defer.Deferred()