Merge pull request #78 from mwolter805/fix/asyncio-lifecycle

fix: track background tasks, defer Queue/Lock construction, use get_running_loop

src/meshcore/ble_cx.py

@@ -51,6 +51,14 @@ class BLEConnection:
         self.pin = pin
         self.rx_char = None
         self._disconnect_callback = None
+        self._background_tasks: set[asyncio.Task] = set()
+
+    def _spawn_background(self, coro) -> asyncio.Task:
+        """Create a tracked background task (prevents GC of fire-and-forget tasks)."""
+        task = asyncio.create_task(coro)
+        self._background_tasks.add(task)
+        task.add_done_callback(self._background_tasks.discard)
+        return task
 
     async def connect(self):
         """
@@ -155,7 +163,7 @@ class BLEConnection:
         self.device = self._user_provided_device
 
         if self._disconnect_callback:
-            asyncio.create_task(self._disconnect_callback("ble_disconnect"))
+            self._spawn_background(self._disconnect_callback("ble_disconnect"))
 
     def set_disconnect_callback(self, callback):
         """Set callback to handle disconnections."""
@@ -166,7 +174,7 @@ class BLEConnection:
 
     def handle_rx(self, _: BleakGATTCharacteristic, data: bytearray):
         if self.reader is not None:
-            asyncio.create_task(self.reader.handle_rx(data))
+            self._spawn_background(self.reader.handle_rx(data))
 
     async def send(self, data):
         if not self.client:

src/meshcore/commands/base.py

@@ -58,17 +58,32 @@ def _validate_destination(dst: DestinationType, prefix_length: int = 6) -> bytes
 
 
 class CommandHandlerBase:
+    """Base class for command handlers.
+
+    .. note::
+        The internal ``asyncio.Lock`` is created lazily on first access
+        so that it binds to the correct running event loop (required for
+        Python 3.9/3.10 compatibility).
+    """
+
     DEFAULT_TIMEOUT = 5.0
 
     def __init__(self, default_timeout: Optional[float] = None):
         self._sender_func: Optional[Callable[[bytes], Coroutine[Any, Any, None]]] = None
         self._reader: Optional[MessageReader] = None
         self.dispatcher: Optional[EventDispatcher] = None
-        self._mesh_request_lock = asyncio.Lock()
+        self.__mesh_request_lock: Optional[asyncio.Lock] = None
         self.default_timeout = (
             default_timeout if default_timeout is not None else self.DEFAULT_TIMEOUT
         )
 
+    @property
+    def _mesh_request_lock(self) -> asyncio.Lock:
+        """Lazy-init lock so it binds to the running loop, not import-time."""
+        if self.__mesh_request_lock is None:
+            self.__mesh_request_lock = asyncio.Lock()
+        return self.__mesh_request_lock
+
     def set_connection(self, connection: Any) -> None:
         async def sender(data: bytes) -> None:
             await connection.send(data)
@@ -170,7 +185,7 @@ class CommandHandlerBase:
             futures: List[asyncio.Future] = []
             subscriptions = []
 
-            loop = asyncio.get_event_loop()
+            loop = asyncio.get_running_loop()
             for event_type in expected_events:
                 future = loop.create_future()
 

src/meshcore/events.py

@@ -129,11 +129,28 @@ class Subscription:
 
 
 class EventDispatcher:
+    """Event dispatch engine.
+
+    .. note::
+        ``start()`` must be called before dispatching or processing events.
+        The internal ``asyncio.Queue`` is created lazily inside ``start()``
+        so that it binds to the correct running event loop (required for
+        Python 3.9/3.10 compatibility).
+    """
+
     def __init__(self):
-        self.queue: asyncio.Queue[Event] = asyncio.Queue()
+        self.queue: Optional[asyncio.Queue[Event]] = None
         self.subscriptions: List[Subscription] = []
         self.running = False
         self._task = None
+        self._background_tasks: set[asyncio.Task] = set()
+
+    def _spawn_background(self, coro) -> asyncio.Task:
+        """Create a tracked background task (prevents GC of fire-and-forget tasks)."""
+        task = asyncio.create_task(coro)
+        self._background_tasks.add(task)
+        task.add_done_callback(self._background_tasks.discard)
+        return task
 
     def subscribe(
         self,
@@ -166,6 +183,10 @@ class EventDispatcher:
             self.subscriptions.remove(subscription)
 
     async def dispatch(self, event: Event):
+        if self.queue is None:
+            raise RuntimeError(
+                "EventDispatcher.start() must be called before dispatching events"
+            )
         await self.queue.put(event)
 
     async def _process_events(self):
@@ -197,7 +218,7 @@ class EventDispatcher:
                     # returns - avoids the race where create_task schedules the callback after
                     # the waiter has already timed out with done=set().
                     if asyncio.iscoroutinefunction(subscription.callback):
-                        asyncio.create_task(self._execute_callback(subscription.callback, event.clone()))
+                        self._spawn_background(self._execute_callback(subscription.callback, event.clone()))
                     else:
                         try:
                             subscription.callback(event.clone())
@@ -220,6 +241,8 @@ class EventDispatcher:
 
     async def start(self):
         if not self.running:
+            if self.queue is None:
+                self.queue = asyncio.Queue()
             self.running = True
             self._task = asyncio.create_task(self._process_events())
 
@@ -227,7 +250,12 @@ class EventDispatcher:
         if self.running:
             self.running = False
             if self._task:
-                await self.queue.join()
+                if self.queue is not None:
+                    await self.queue.join()
+                # Wait for any in-flight async callbacks to complete before
+                # tearing down (F07: task_done fires before callbacks finish).
+                if self._background_tasks:
+                    await asyncio.gather(*self._background_tasks, return_exceptions=True)
                 self._task.cancel()
                 try:
                     await self._task

src/meshcore/serial_cx.py

@@ -20,11 +20,19 @@ class SerialConnection:
         self._disconnect_callback = None
         self.cx_dly = cx_dly
         self._connected_event = asyncio.Event()
+        self._background_tasks: set[asyncio.Task] = set()
 
         self.frame_expected_size = 0
         self.inframe = b""
         self.header = b""
 
+    def _spawn_background(self, coro) -> asyncio.Task:
+        """Create a tracked background task (prevents GC of fire-and-forget tasks)."""
+        task = asyncio.create_task(coro)
+        self._background_tasks.add(task)
+        task.add_done_callback(self._background_tasks.discard)
+        return task
+
     class MCSerialClientProtocol(asyncio.Protocol):
         def __init__(self, cx):
             self.cx = cx
@@ -44,7 +52,7 @@ class SerialConnection:
             self.cx._connected_event.clear()
 
             if self.cx._disconnect_callback:
-                asyncio.create_task(self.cx._disconnect_callback("serial_disconnect"))
+                self.cx._spawn_background(self.cx._disconnect_callback("serial_disconnect"))
 
         def pause_writing(self):
             logger.debug("pause writing")
@@ -114,7 +122,7 @@ class SerialConnection:
         data = data[upbound:]
         if self.reader is not None:
             # feed meshcore reader
-            asyncio.create_task(self.reader.handle_rx(self.inframe))
+            self._spawn_background(self.reader.handle_rx(self.inframe))
         # reset inframe
         self.inframe = b""
         self.header = b""

src/meshcore/tcp_cx.py

@@ -24,6 +24,14 @@ class TCPConnection:
         self.frame_expected_size = 0
         self.header = b""
         self.inframe = b""
+        self._background_tasks: set[asyncio.Task] = set()
+
+    def _spawn_background(self, coro) -> asyncio.Task:
+        """Create a tracked background task (prevents GC of fire-and-forget tasks)."""
+        task = asyncio.create_task(coro)
+        self._background_tasks.add(task)
+        task.add_done_callback(self._background_tasks.discard)
+        return task
 
     class MCClientProtocol(asyncio.Protocol):
         def __init__(self, cx):
@@ -47,7 +55,7 @@ class TCPConnection:
         def connection_lost(self, exc):
             logger.debug("TCP server closed the connection")
             if self.cx._disconnect_callback:
-                asyncio.create_task(self.cx._disconnect_callback("tcp_disconnect"))
+                self.cx._spawn_background(self.cx._disconnect_callback("tcp_disconnect"))
 
     async def connect(self):
         """
@@ -108,7 +116,7 @@ class TCPConnection:
         data = data[upbound:]
         if self.reader is not None:
             # feed meshcore reader
-            asyncio.create_task(self.reader.handle_rx(self.inframe))
+            self._spawn_background(self.reader.handle_rx(self.inframe))
         # reset inframe
         self.inframe = b""
         self.header = b""