Handle futures in threaded loop

Author: elacuesta

pylintrc

@@ -9,6 +9,7 @@ disable=
     too-few-public-methods,
     too-many-arguments,
     too-many-instance-attributes,
+    too-many-lines,
     # tests
     duplicate-code,
     import-outside-toplevel,

scrapy_playwright/_utils.py

@@ -2,6 +2,7 @@
 import logging
 import platform
 import threading
+from dataclasses import dataclass
 from typing import Awaitable, Dict, Iterator, Optional, Tuple, Union
 
 import scrapy
@@ -104,6 +105,13 @@ async def _get_header_value(
         return None
 
 
+@dataclass
+class _QueueItem:
+    coro: Awaitable
+    promise: Deferred | asyncio.Future
+    loop: asyncio.AbstractEventLoop | None = None
+
+
 class _ThreadedLoopAdapter:
     """Utility class to start an asyncio event loop in a new thread and redirect coroutines.
     This allows to run Playwright in a different loop than the Scrapy crawler, allowing to
@@ -116,29 +124,54 @@ class _ThreadedLoopAdapter:
     _stop_events: Dict[int, asyncio.Event] = {}
 
     @classmethod
-    async def _handle_coro(cls, coro: Awaitable, dfd: Deferred) -> None:
+    async def _handle_coro_deferred(cls, queue_item: _QueueItem) -> None:
         from twisted.internet import reactor
 
+        dfd: Deferred = queue_item.promise
+
         try:
-            result = await coro
+            result = await queue_item.coro
         except Exception as exc:
             reactor.callFromThread(dfd.errback, failure.Failure(exc))
         else:
             reactor.callFromThread(dfd.callback, result)
 
+    @classmethod
+    async def _handle_coro_future(cls, queue_item: _QueueItem) -> None:
+        future: asyncio.Future = queue_item.promise
+        assert queue_item.loop is not None  # typing
+        try:
+            result = await queue_item.coro
+        except Exception as exc:
+            queue_item.loop.call_soon_threadsafe(future.set_exception, exc)
+        else:
+            queue_item.loop.call_soon_threadsafe(future.set_result, result)
+
     @classmethod
     async def _process_queue(cls) -> None:
         while any(not ev.is_set() for ev in cls._stop_events.values()):
-            coro, dfd = await cls._coro_queue.get()
-            asyncio.create_task(cls._handle_coro(coro, dfd))
+            queue_item = await cls._coro_queue.get()
+            if isinstance(queue_item.promise, asyncio.Future):
+                asyncio.create_task(cls._handle_coro_future(queue_item))
+            elif isinstance(queue_item.promise, Deferred):
+                asyncio.create_task(cls._handle_coro_deferred(queue_item))
             cls._coro_queue.task_done()
 
     @classmethod
     def _deferred_from_coro(cls, coro) -> Deferred:
         dfd: Deferred = Deferred()
-        asyncio.run_coroutine_threadsafe(cls._coro_queue.put((coro, dfd)), cls._loop)
+        queue_item = _QueueItem(coro=coro, promise=dfd)
+        asyncio.run_coroutine_threadsafe(cls._coro_queue.put(queue_item), cls._loop)
         return dfd
 
+    @classmethod
+    def _future_from_coro(cls, coro) -> asyncio.Future:
+        target_loop = asyncio.get_running_loop()  # Scrapy thread loop
+        future: asyncio.Future = asyncio.Future()
+        queue_item = _QueueItem(coro=coro, promise=future, loop=target_loop)
+        asyncio.run_coroutine_threadsafe(cls._coro_queue.put(queue_item), cls._loop)
+        return future
+
     @classmethod
     def start(cls, download_handler_id: int) -> None:
         """Start the event loop in a new thread if not already started.

scrapy_playwright/handler.py

@@ -31,7 +31,7 @@
 from scrapy.http.headers import Headers
 from scrapy.responsetypes import responsetypes
 from scrapy.settings import Settings
-from scrapy.utils.defer import deferred_from_coro, maybe_deferred_to_future
+from scrapy.utils.defer import deferred_from_coro
 from scrapy.utils.misc import load_object
 from scrapy.utils.reactor import verify_installed_reactor
 from twisted.internet.defer import Deferred, inlineCallbacks
@@ -144,18 +144,21 @@ def __init__(self, crawler: Crawler) -> None:
         verify_installed_reactor("twisted.internet.asyncioreactor.AsyncioSelectorReactor")
         if _SCRAPY_ASYNC_API:
             super().__init__(crawler=crawler)
-            crawler.signals.connect(self._launch, signals.engine_started)
         else:
             super().__init__(  # pylint: disable=unexpected-keyword-arg
                 settings=crawler.settings, crawler=crawler
             )
-            crawler.signals.connect(self._engine_started, signals.engine_started)
         self.stats = crawler.stats
         self.config = Config.from_settings(crawler.settings)
 
         if self.config.use_threaded_loop:
             _ThreadedLoopAdapter.start(id(self))
 
+        if _SCRAPY_ASYNC_API:
+            crawler.signals.connect(self._maybe_launch_in_thread, signals.engine_started)
+        else:
+            crawler.signals.connect(self._engine_started, signals.engine_started)
+
         self.browser_launch_lock = asyncio.Lock()
         self.context_launch_lock = asyncio.Lock()
         self.context_wrappers: Dict[str, BrowserContextWrapper] = {}
@@ -186,10 +189,17 @@ def _deferred_from_coro(self, coro: Awaitable) -> Deferred:
             return _ThreadedLoopAdapter._deferred_from_coro(coro)
         return deferred_from_coro(coro)
 
+    def _future_from_coro(self, coro: Awaitable) -> asyncio.Future:
+        if self.config.use_threaded_loop:
+            return _ThreadedLoopAdapter._future_from_coro(coro)
+        return asyncio.ensure_future(coro)
+
     def _engine_started(self) -> Deferred:
-        """Launch the browser. Use the engine_started signal as it supports returning deferreds."""
         return self._deferred_from_coro(self._launch())
 
+    async def _maybe_launch_in_thread(self) -> None:
+        await self._future_from_coro(self._launch())
+
     async def _launch(self) -> None:
         """Launch Playwright manager and configured startup context(s)."""
         logger.info("Starting download handler")
@@ -362,7 +372,7 @@ def _set_max_concurrent_context_count(self):
         async def close(self) -> None:
             logger.info("Closing download handler")
             await super().close()
-            await self._close()
+            await self._future_from_coro(self._close())
             if self.config.use_threaded_loop:
                 _ThreadedLoopAdapter.stop(id(self))
 
@@ -392,9 +402,8 @@ async def _close(self) -> None:
 
         async def download_request(self, request: Request) -> Response:
             if request.meta.get("playwright"):
-                return await maybe_deferred_to_future(
-                    self._deferred_from_coro(self._download_request(request, self._crawler.spider))
-                )
+                coro = self._download_request(request)
+                return await self._future_from_coro(coro)
             return await super().download_request(  # pylint: disable=no-value-for-parameter
                 request
             )
@@ -410,7 +419,8 @@ def download_request(  # type: ignore[misc] # pylint: disable=invalid-overridden
                 request=request, spider=spider
             )
 
-    async def _download_request(self, request: Request, spider: Spider) -> Response:
+    async def _download_request(self, request: Request, spider: Spider | None = None) -> Response:
+        spider = spider or self._crawler.spider
         counter = 0
         while True:
             try: