Allow multiple handlers in separate thread (#299)

Author: elacuesta

scrapy_playwright/_utils.py

@@ -2,7 +2,7 @@
 import logging
 import platform
 import threading
-from typing import Awaitable, Iterator, Optional, Tuple, Union
+from typing import Awaitable, Dict, Iterator, Optional, Tuple, Union
 
 import scrapy
 from playwright.async_api import Error, Page, Request, Response
@@ -103,68 +103,58 @@ async def _get_header_value(
         return None
 
 
-if platform.system() == "Windows":
-
-    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
-        use ProactorEventLoop which is supported by Playwright on Windows.
-        """
-
-        _loop: asyncio.AbstractEventLoop
-        _thread: threading.Thread
-        _coro_queue: asyncio.Queue = asyncio.Queue()
-        _stop_event: asyncio.Event = asyncio.Event()
-
-        @classmethod
-        async def _handle_coro(cls, coro, future) -> None:
-            try:
-                future.set_result(await coro)
-            except Exception as exc:
-                future.set_exception(exc)
-
-        @classmethod
-        async def _process_queue(cls) -> None:
-            while not cls._stop_event.is_set():
-                coro, future = await cls._coro_queue.get()
-                asyncio.create_task(cls._handle_coro(coro, future))
-                cls._coro_queue.task_done()
-
-        @classmethod
-        def _deferred_from_coro(cls, coro) -> Deferred:
-            future: asyncio.Future = asyncio.Future()
-            asyncio.run_coroutine_threadsafe(cls._coro_queue.put((coro, future)), cls._loop)
-            return scrapy.utils.defer.deferred_from_coro(future)
-
-        @classmethod
-        def start(cls) -> None:
-            policy = asyncio.WindowsProactorEventLoopPolicy()  # type: ignore[attr-defined]
+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
+    use ProactorEventLoop which is supported by Playwright on Windows.
+    """
+
+    _loop: asyncio.AbstractEventLoop
+    _thread: threading.Thread
+    _coro_queue: asyncio.Queue = asyncio.Queue()
+    _stop_events: Dict[int, asyncio.Event] = {}
+
+    @classmethod
+    async def _handle_coro(cls, coro, future) -> None:
+        try:
+            future.set_result(await coro)
+        except Exception as exc:
+            future.set_exception(exc)
+
+    @classmethod
+    async def _process_queue(cls) -> None:
+        while any(not ev.is_set() for ev in cls._stop_events.values()):
+            coro, future = await cls._coro_queue.get()
+            asyncio.create_task(cls._handle_coro(coro, future))
+            cls._coro_queue.task_done()
+
+    @classmethod
+    def _deferred_from_coro(cls, coro) -> Deferred:
+        future: asyncio.Future = asyncio.Future()
+        asyncio.run_coroutine_threadsafe(cls._coro_queue.put((coro, future)), cls._loop)
+        return scrapy.utils.defer.deferred_from_coro(future)
+
+    @classmethod
+    def start(cls, caller_id: int) -> None:
+        cls._stop_events[caller_id] = asyncio.Event()
+        if not getattr(cls, "_loop", None):
+            policy = asyncio.DefaultEventLoopPolicy()
+            if platform.system() == "Windows":
+                policy = asyncio.WindowsProactorEventLoopPolicy()  # type: ignore[attr-defined]
             cls._loop = policy.new_event_loop()
             asyncio.set_event_loop(cls._loop)
 
+        if not getattr(cls, "_thread", None):
             cls._thread = threading.Thread(target=cls._loop.run_forever, daemon=True)
             cls._thread.start()
             logger.info("Started loop on separate thread: %s", cls._loop)
-
             asyncio.run_coroutine_threadsafe(cls._process_queue(), cls._loop)
 
-        @classmethod
-        def stop(cls) -> None:
-            cls._stop_event.set()
+    @classmethod
+    def stop(cls, caller_id: int) -> None:
+        """Wait until all handlers are closed to stop the event loop and join the thread."""
+        cls._stop_events[caller_id].set()
+        if all(ev.is_set() for ev in cls._stop_events.values()):
             asyncio.run_coroutine_threadsafe(cls._coro_queue.join(), cls._loop)
             cls._loop.call_soon_threadsafe(cls._loop.stop)
             cls._thread.join()
-
-    _deferred_from_coro = _ThreadedLoopAdapter._deferred_from_coro
-else:
-
-    class _ThreadedLoopAdapter:  # type: ignore[no-redef]
-        @classmethod
-        def start(cls) -> None:
-            pass
-
-        @classmethod
-        def stop(cls) -> None:
-            pass
-
-    _deferred_from_coro = scrapy.utils.defer.deferred_from_coro

scrapy_playwright/handler.py

@@ -1,5 +1,6 @@
 import asyncio
 import logging
+import platform
 from contextlib import suppress
 from dataclasses import dataclass, field as dataclass_field
 from ipaddress import ip_address
@@ -26,6 +27,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
 from scrapy.utils.misc import load_object
 from scrapy.utils.reactor import verify_installed_reactor
 from twisted.internet.defer import Deferred, inlineCallbacks
@@ -34,7 +36,6 @@
 from scrapy_playwright.page import PageMethod
 from scrapy_playwright._utils import (
     _ThreadedLoopAdapter,
-    _deferred_from_coro,
     _encode_body,
     _get_float_setting,
     _get_header_value,
@@ -91,6 +92,7 @@ class Config:
     startup_context_kwargs: dict
     navigation_timeout: Optional[float]
     restart_disconnected_browser: bool
+    use_threaded_loop: bool
 
     @classmethod
     def from_settings(cls, settings: Settings) -> "Config":
@@ -114,6 +116,8 @@ def from_settings(cls, settings: Settings) -> "Config":
             restart_disconnected_browser=settings.getbool(
                 "PLAYWRIGHT_RESTART_DISCONNECTED_BROWSER", default=True
             ),
+            use_threaded_loop=platform.system() == "Windows"
+            or settings.getbool("_PLAYWRIGHT_THREADED_LOOP", False),
         )
         cfg.cdp_kwargs.pop("endpoint_url", None)
         cfg.connect_kwargs.pop("ws_endpoint", None)
@@ -130,13 +134,14 @@ class ScrapyPlaywrightDownloadHandler(HTTPDownloadHandler):
 
     def __init__(self, crawler: Crawler) -> None:
         super().__init__(settings=crawler.settings, crawler=crawler)
-        _ThreadedLoopAdapter.start()
         verify_installed_reactor("twisted.internet.asyncioreactor.AsyncioSelectorReactor")
         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))
+
         self.browser_launch_lock = asyncio.Lock()
         self.context_launch_lock = asyncio.Lock()
         self.context_wrappers: Dict[str, BrowserContextWrapper] = {}
@@ -162,9 +167,14 @@ def __init__(self, crawler: Crawler) -> None:
     def from_crawler(cls: Type[PlaywrightHandler], crawler: Crawler) -> PlaywrightHandler:
         return cls(crawler)
 
+    def _deferred_from_coro(self, coro: Awaitable) -> Deferred:
+        if self.config.use_threaded_loop:
+            return _ThreadedLoopAdapter._deferred_from_coro(coro)
+        return deferred_from_coro(coro)
+
     def _engine_started(self) -> Deferred:
         """Launch the browser. Use the engine_started signal as it supports returning deferreds."""
-        return _deferred_from_coro(self._launch())
+        return self._deferred_from_coro(self._launch())
 
     async def _launch(self) -> None:
         """Launch Playwright manager and configured startup context(s)."""
@@ -333,8 +343,9 @@ def _set_max_concurrent_context_count(self):
     def close(self) -> Deferred:
         logger.info("Closing download handler")
         yield super().close()
-        yield _deferred_from_coro(self._close())
-        _ThreadedLoopAdapter.stop()
+        yield self._deferred_from_coro(self._close())
+        if self.config.use_threaded_loop:
+            _ThreadedLoopAdapter.stop(id(self))
 
     async def _close(self) -> None:
         await asyncio.gather(*[ctx.context.close() for ctx in self.context_wrappers.values()])
@@ -349,7 +360,7 @@ async def _close(self) -> None:
 
     def download_request(self, request: Request, spider: Spider) -> Deferred:
         if request.meta.get("playwright"):
-            return _deferred_from_coro(self._download_request(request, spider))
+            return self._deferred_from_coro(self._download_request(request, spider))
         return super().download_request(request, spider)
 
     async def _download_request(self, request: Request, spider: Spider) -> Response:

tests/__init__.py

@@ -24,10 +24,11 @@ def allow_windows(test_method):
 
         @wraps(test_method)
         async def wrapped(self, *args, **kwargs):
-            _ThreadedLoopAdapter.start()
+            caller_id = 1234
+            _ThreadedLoopAdapter.start(caller_id)
             coro = test_method(self, *args, **kwargs)
             asyncio.run_coroutine_threadsafe(coro=coro, loop=_ThreadedLoopAdapter._loop).result()
-            _ThreadedLoopAdapter.stop()
+            _ThreadedLoopAdapter.stop(caller_id)
 
         return wrapped