replace asyncio with trio (#1240)

* migrate to trio

* test(yamux): serialize writes in TrioStreamAdapter to fix Windows BusyResourceError

Trio allows only one task to use a stream at a time. In test_yamux_fin_on_window_update
(and test_yamux_rst_on_window_update) the test injects a raw frame via
secured_conn.write() while the server Yamux task also writes to the same
stream, causing trio.BusyResourceError on Windows where scheduling exposes
the race. Add a write lock so all writes are serialized and the tests pass
on all platforms.

Co-authored-by: Cursor <cursoragent@cursor.com>

* add newsfragment for #174

---------

Co-authored-by: Nilav <nilav@Nilav-MacBook-Air.local>
Co-authored-by: acul71 <34693171+acul71@users.noreply.github.com>
Co-authored-by: acul71 <34693171+acul71@users.noreply.github.com>
Co-authored-by: Cursor <cursoragent@cursor.com>

Author: 4 people

examples/autotls_browser/certificate_manager.py

@@ -6,7 +6,6 @@
 validation, and lifecycle management.
 """
 
-import asyncio
 from datetime import datetime, timedelta
 import logging
 from pathlib import Path
@@ -17,6 +16,7 @@
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import rsa
 from cryptography.x509.oid import NameOID
+import trio
 
 from libp2p.peer.id import ID
 
@@ -51,7 +51,18 @@ def __init__(
         self.renewal_threshold_hours = renewal_threshold_hours
 
         self._certificates: dict[tuple[ID, str], dict[Any, Any]] = {}
-        self._renewal_tasks: dict[tuple[ID, str], asyncio.Task[Any]] = {}
+        self._renewal_scopes: dict[tuple[ID, str], trio.CancelScope] = {}
+        self._nursery: trio.Nursery | None = None
+
+    async def start(self, nursery: trio.Nursery) -> None:
+        """Attach manager to a long-lived nursery."""
+        self._nursery = nursery
+
+    async def shutdown(self) -> None:
+        """Cancel all renewal jobs."""
+        for scope in self._renewal_scopes.values():
+            scope.cancel()
+        logger.info("Certificate manager shutdown complete")
 
     async def get_certificate(
         self,
@@ -183,13 +194,15 @@ async def _schedule_renewal(
         peer_id: ID,
         domain: str,
         cert_data: dict[Any, Any],
+        _current_scope: trio.CancelScope | None = None,
     ) -> None:
         """Schedule certificate renewal."""
         key = (peer_id, domain)
 
         # Cancel existing renewal task
-        if key in self._renewal_tasks:
-            self._renewal_tasks[key].cancel()
+        existing_scope = self._renewal_scopes.get(key)
+        if existing_scope is not None and existing_scope is not _current_scope:
+            existing_scope.cancel()
 
         # Calculate renewal time
         expires_at = datetime.fromisoformat(cert_data["expires_at"])
@@ -204,24 +217,41 @@ async def _schedule_renewal(
             f"Scheduling certificate renewal for {peer_id} in {delay:.0f} seconds"
         )
 
-        async def renew_certificate() -> None:
-            try:
-                await asyncio.sleep(delay)
-
-                logger.info(f"Renewing certificate for {peer_id} on {domain}")
-                new_cert_data = await self.get_certificate(
-                    peer_id, domain, force_renew=True
-                )
-
-                # Update cached certificate
-                self._certificates[key] = new_cert_data  # type: ignore
+        scope = trio.CancelScope()
+        self._renewal_scopes[key] = scope
 
-            except asyncio.CancelledError:
-                logger.debug(f"Certificate renewal cancelled for {peer_id}")
-            except Exception as e:
-                logger.error(f"Certificate renewal failed for {peer_id}: {e}")
-
-        self._renewal_tasks[key] = asyncio.create_task(renew_certificate())
+        async def renew_certificate() -> None:
+            with scope:
+                try:
+                    await trio.sleep(delay)
+
+                    logger.info(f"Renewing certificate for {peer_id} on {domain}")
+                    new_cert_data = await self._generate_certificate(peer_id, domain)
+                    await self._store_certificate_to_storage(
+                        peer_id, domain, new_cert_data
+                    )
+                    self._certificates[key] = new_cert_data
+
+                    await self._schedule_renewal(
+                        peer_id,
+                        domain,
+                        new_cert_data,
+                        _current_scope=scope,
+                    )
+
+                except trio.Cancelled:
+                    logger.debug(f"Certificate renewal cancelled for {peer_id}")
+                    raise
+                except Exception as e:
+                    logger.error(f"Certificate renewal failed for {peer_id}: {e}")
+                finally:
+                    if self._renewal_scopes.get(key) is scope:
+                        self._renewal_scopes.pop(key, None)
+
+        if self._nursery is not None:
+            self._nursery.start_soon(renew_certificate)
+        else:
+            trio.lowlevel.spawn_system_task(renew_certificate)
 
     def _get_cert_path(self, peer_id: ID, domain: str) -> Path:
         """Get certificate file path."""
@@ -323,9 +353,9 @@ async def cleanup_expired_certificates(self) -> None:
             del self._certificates[key]
 
             # Cancel renewal task
-            if key in self._renewal_tasks:
-                self._renewal_tasks[key].cancel()
-                del self._renewal_tasks[key]
+            scope = self._renewal_scopes.pop(key, None)
+            if scope is not None:
+                scope.cancel()
 
         if expired_keys:
             logger.info(f"Cleaned up {len(expired_keys)} expired certificates")
@@ -375,26 +405,13 @@ async def revoke_certificate(
             del self._certificates[key]
 
         # Cancel renewal task
-        if key in self._renewal_tasks:
-            self._renewal_tasks[key].cancel()
-            del self._renewal_tasks[key]
+        scope = self._renewal_scopes.pop(key, None)
+        if scope is not None:
+            scope.cancel()
 
         # Remove from storage
         cert_path = self._get_cert_path(peer_id, domain)
         if cert_path.exists():
             cert_path.unlink()
 
         logger.info(f"Revoked certificate for {peer_id} on {domain}")
-
-    async def shutdown(self) -> None:
-        """Shutdown certificate manager."""
-        # Cancel all renewal tasks
-        for task in self._renewal_tasks.values():
-            if not task.done():
-                task.cancel()
-
-        # Wait for tasks to complete
-        if self._renewal_tasks:
-            await asyncio.gather(*self._renewal_tasks.values(), return_exceptions=True)
-
-        logger.info("Certificate manager shutdown complete")

examples/persistent_peerstore/example_usage.py

@@ -6,11 +6,11 @@
 with Protocol Buffer serialization instead of unsafe pickle.
 """
 
-import asyncio
 from pathlib import Path
 import tempfile
 
 from multiaddr import Multiaddr
+import trio
 
 from libp2p.peer.id import ID
 from libp2p.peer.persistent import (
@@ -111,7 +111,7 @@ def security_example():
 
 if __name__ == "__main__":
     # Run async example
-    asyncio.run(async_example())
+    trio.run(async_example)
 
     # Run sync example
     sync_example()

examples/transport_integration_demo.py

@@ -9,11 +9,12 @@
 4. How the new system automatically selects the right transport
 """
 
-import asyncio
 import logging
 from pathlib import Path
 import sys
 
+import trio
+
 # Add the libp2p directory to the path so we can import it
 sys.path.insert(0, str(Path(__file__).parent.parent))
 
@@ -200,7 +201,7 @@ async def main():
 
 if __name__ == "__main__":
     try:
-        asyncio.run(main())
+        trio.run(main)
     except KeyboardInterrupt:
         print("\n👋 Demo interrupted by user")
     except Exception as e:

libp2p/tools/async_service/base.py

@@ -3,7 +3,6 @@
 from abc import (
     abstractmethod,
 )
-import asyncio
 from collections import (
     Counter,
 )
@@ -22,6 +21,8 @@
 )
 import uuid
 
+import trio
+
 from ._utils import (
     is_verbose_logging_enabled,
 )
@@ -350,7 +351,7 @@ async def _run_and_manage_task(self, task: TaskAPI) -> None:
                             child,
                             new_parent or "root",
                         )
-        except asyncio.CancelledError:
+        except trio.Cancelled:
             logger.debug("%s: task %s raised CancelledError.", self, task)
             raise
         except Exception as err:

libp2p/transport/websocket/autotls.py

@@ -8,7 +8,6 @@
 Based on patterns from JavaScript and Go libp2p implementations.
 """
 
-import asyncio
 from collections.abc import Callable
 from datetime import datetime, timedelta, timezone
 import logging
@@ -21,6 +20,7 @@
 from cryptography.hazmat.primitives import hashes, serialization
 from cryptography.hazmat.primitives.asymmetric import rsa
 from cryptography.x509.oid import NameOID
+import trio
 
 from libp2p.peer.id import ID
 
@@ -218,28 +218,26 @@ def __init__(
         self.on_certificate_renew = on_certificate_renew
 
         self._active_certificates: dict[tuple[ID, str], TLSCertificate] = {}
-        self._renewal_tasks: dict[tuple[ID, str], asyncio.Task[None]] = {}
-        self._shutdown_event = asyncio.Event()
+        self._renewal_scopes: dict[tuple[ID, str], trio.CancelScope] = {}
+        self._nursery: trio.Nursery | None = None
 
-    async def start(self) -> None:
+    async def start(self, nursery: trio.Nursery | None = None) -> None:
         """Start the AutoTLS manager."""
         logger.info("Starting AutoTLS manager")
+        if nursery is not None:
+            self._nursery = nursery
         # Manager is ready to handle certificate requests
         pass
 
     async def stop(self) -> None:
         """Stop the AutoTLS manager."""
         logger.info("Stopping AutoTLS manager")
-        self._shutdown_event.set()
 
         # Cancel all renewal tasks
-        for task in self._renewal_tasks.values():
-            if not task.done():
-                task.cancel()
+        for scope in self._renewal_scopes.values():
+            scope.cancel()
 
-        # Wait for tasks to complete
-        if self._renewal_tasks:
-            await asyncio.gather(*self._renewal_tasks.values(), return_exceptions=True)
+        logger.info("AutoTLS manager stopped")
 
     async def get_certificate(
         self,
@@ -358,13 +356,15 @@ async def _schedule_renewal(
         peer_id: ID,
         domain: str,
         cert: TLSCertificate,
+        _current_scope: trio.CancelScope | None = None,
     ) -> None:
         """Schedule certificate renewal."""
         key = (peer_id, domain)
 
         # Cancel existing renewal task
-        if key in self._renewal_tasks:
-            self._renewal_tasks[key].cancel()
+        existing_scope = self._renewal_scopes.get(key)
+        if existing_scope is not None and existing_scope is not _current_scope:
+            existing_scope.cancel()
 
         # Calculate renewal time
         renewal_time = cert.expires_at - timedelta(hours=self.renewal_threshold_hours)
@@ -378,26 +378,47 @@ async def _schedule_renewal(
             f"Scheduling certificate renewal for {peer_id} in {delay:.0f} seconds"
         )
 
-        async def renew_certificate() -> None:
-            try:
-                await asyncio.sleep(delay)
-
-                if self._shutdown_event.is_set():
-                    return
-
-                logger.info(f"Renewing certificate for {peer_id} on {domain}")
-                new_cert = await self.get_certificate(peer_id, domain, force_renew=True)
-
-                # Notify renewal
-                if self.on_certificate_renew:
-                    self.on_certificate_renew(new_cert)
-
-            except asyncio.CancelledError:
-                logger.debug(f"Certificate renewal cancelled for {peer_id}")
-            except Exception as e:
-                logger.error(f"Certificate renewal failed for {peer_id}: {e}")
+        scope = trio.CancelScope()
+        self._renewal_scopes[key] = scope
 
-        self._renewal_tasks[key] = asyncio.create_task(renew_certificate())
+        async def renew_certificate() -> None:
+            with scope:
+                try:
+                    await trio.sleep(delay)
+
+                    logger.info(f"Renewing certificate for {peer_id} on {domain}")
+                    new_cert = await self._renew_certificate_now(peer_id, domain)
+
+                    # Notify renewal
+                    if self.on_certificate_renew:
+                        self.on_certificate_renew(new_cert)
+
+                    await self._schedule_renewal(
+                        peer_id,
+                        domain,
+                        new_cert,
+                        _current_scope=scope,
+                    )
+                except trio.Cancelled:
+                    logger.debug(f"Certificate renewal cancelled for {peer_id}")
+                    raise
+                except Exception as e:
+                    logger.error(f"Certificate renewal failed for {peer_id}: {e}")
+                finally:
+                    if self._renewal_scopes.get(key) is scope:
+                        self._renewal_scopes.pop(key, None)
+
+        if self._nursery is not None:
+            self._nursery.start_soon(renew_certificate)
+        else:
+            trio.lowlevel.spawn_system_task(renew_certificate)
+
+    async def _renew_certificate_now(self, peer_id: ID, domain: str) -> TLSCertificate:
+        key = (peer_id, domain)
+        cert = await self._generate_certificate(peer_id, domain)
+        await self.storage.store_certificate(cert)
+        self._active_certificates[key] = cert
+        return cert
 
     def get_ssl_context(self, peer_id: ID | None, domain: str) -> ssl.SSLContext | None:
         """Get SSL context for peer ID and domain."""

newsfragments/174.feature.rst

@@ -0,0 +1 @@
+py-libp2p now uses trio exclusively for async operations, removing remaining asyncio usage.

newsfragments/301.feature.rst

@@ -0,0 +1 @@
+Migrated remaining asyncio usage with trio, so the codebase uses a single async runtime.

pyproject.toml

@@ -99,7 +99,6 @@ test = [
     "factory-boy>=2.12.0,<3.0.0",
     "p2pclient>=0.2.1",
     "pytest>=7.0.0",
-    "pytest-asyncio>=0.21.0",
     "pytest-timeout>=2.4.0",
     "pytest-trio>=0.5.2",
     "pytest-xdist>=2.4.0",