Fix deadlock in atexit cleanup handlers

Author: Wanderlust

playwright/_impl/_transport.py

@@ -99,6 +99,14 @@ def request_stop(self) -> None:
         self._output.close()
 
     async def wait_until_stopped(self) -> None:
+        # In atexit scenarios, the original event loop might be closed.
+        # If so, we can't wait for _stopped_future (it's tied to the closed loop).
+        if self._loop.is_closed():
+            # Loop is closed. The process is being terminated by run() already.
+            # Just wait for it directly without asyncio (it will self-clean in time).
+            return
+        
+        # Normal case: original loop still exists, wait for the stopped signal
         await self._stopped_future
 
     async def connect(self) -> None:
@@ -165,10 +173,25 @@ async def run(self) -> None:
                         Exception("Connection closed while reading from the driver")
                     )
                 break
+            except asyncio.CancelledError:
+                break
             await asyncio.sleep(0)
-
-        await self._proc.communicate()
-        self._stopped_future.set_result(None)
+        
+        # Graceful shutdown: only if event loop is still running
+        try:
+            asyncio.get_running_loop()
+        except RuntimeError:
+            # No running loop, OS will clean up the process during exit
+            return
+        
+        # Process is still running and we have an event loop
+        if self._proc.returncode is None:
+            self._proc.terminate()
+            # Let OS clean up if process doesn't respond to SIGTERM
+        
+        # Notify anyone waiting that the transport has fully stopped
+        if not self._stopped_future.done():
+            self._stopped_future.set_result(None)
 
     def send(self, message: Dict) -> None:
         assert self._output

tests/async/test_atexit_cleanup.py

@@ -0,0 +1,24 @@
+import asyncio
+
+from playwright.async_api import async_playwright
+
+
+async def test_stop_during_concurrent_operations() -> None:
+    playwright = await async_playwright().start()
+    browser = await playwright.chromium.launch()
+    
+    async def quick_operation():
+        try:
+            page = await browser.new_page()
+            await page.close()
+        except Exception:
+            pass
+    
+    task = asyncio.create_task(quick_operation())
+    await asyncio.sleep(0.01)
+    await playwright.stop()
+    
+    try:
+        await asyncio.wait_for(task, timeout=2.0)
+    except asyncio.TimeoutError:
+        raise AssertionError("Playwright.stop() deadlocked during concurrent operations")