nvdemux: remove manager callback altogether

Author: mangelajo

packages/jumpstarter-driver-pyserial/jumpstarter_driver_pyserial/nvdemux/driver.py

@@ -3,7 +3,7 @@
 from dataclasses import dataclass, field
 from typing import Optional
 
-from anyio import Event, fail_after, sleep
+from anyio import sleep
 from anyio._backends._asyncio import StreamReaderWrapper, StreamWriterWrapper
 from serial_asyncio import open_serial_connection
 
@@ -49,7 +49,6 @@ class NVDemuxSerial(Driver):
     poll_interval: float = field(default=1.0)
 
     # Internal state (not init params)
-    _ready: Event = field(init=False, default_factory=Event)
     _registered: bool = field(init=False, default=False)
 
     def __post_init__(self):
@@ -65,7 +64,6 @@ def __post_init__(self):
                 device=self.device,
                 chip=self.chip,
                 target=self.target,
-                callback=self._on_target_ready,
                 poll_interval=self.poll_interval,
             )
             self._registered = True
@@ -78,16 +76,6 @@ def __post_init__(self):
     def client(cls) -> str:
         return "jumpstarter_driver_pyserial.client.PySerialClient"
 
-    def _on_target_ready(self, target: str, pts_path: str):
-        """Callback invoked by DemuxerManager when target becomes ready.
-
-        Args:
-            target: The target channel that became ready
-            pts_path: The pts path for this target
-        """
-        self.logger.info("Target '%s' ready with pts path: %s", target, pts_path)
-        self._ready.set()
-
     def close(self):
         """Unregister from the DemuxerManager."""
         if self._registered:
@@ -105,23 +93,16 @@ async def connect(self):
         Waits for the demuxer to be ready (device connected and pts path discovered)
         before opening the serial connection.
         """
-        # Wait for ready state with timeout
-        try:
-            with fail_after(self.timeout):
-                await self._ready.wait()
-        except TimeoutError:
-            raise TimeoutError(
-                f"Timeout waiting for demuxer to become ready (device pattern: {self.device})"
-            ) from None
-
-        # Get the current pts path from manager (retry until timeout)
+        # Poll for pts path until available or timeout
         manager = DemuxerManager.get_instance()
         pts_start = time.monotonic()
         pts_path = manager.get_pts_path(str(self.uuid))
         while not pts_path:
             elapsed = time.monotonic() - pts_start
             if elapsed >= self.timeout:
-                raise TimeoutError("Demuxer ready but no pts path available after retrying")
+                raise TimeoutError(
+                    f"Timeout waiting for demuxer to become ready (device pattern: {self.device})"
+                )
             await sleep(self.poll_interval)
             pts_path = manager.get_pts_path(str(self.uuid))
 

packages/jumpstarter-driver-pyserial/jumpstarter_driver_pyserial/nvdemux/driver_test.py

@@ -1,7 +1,6 @@
 """Tests for NVDemuxSerial driver."""
 
 import tempfile
-import time
 from unittest.mock import MagicMock, patch
 
 from .driver import NVDemuxSerial
@@ -35,36 +34,6 @@ def test_nvdemux_registration():
                 driver.close()
 
 
-def test_nvdemux_callback_sets_ready():
-    """Test that callback from manager sets the ready event."""
-    with tempfile.NamedTemporaryFile() as device_file:
-        with patch("jumpstarter_driver_pyserial.nvdemux.driver.DemuxerManager") as mock_manager_class:
-            mock_manager = MagicMock()
-            mock_manager_class.get_instance.return_value = mock_manager
-
-            driver = NVDemuxSerial(
-                demuxer_path="/usr/bin/demuxer",
-                device=device_file.name,
-                target="CCPLEX: 0",
-                timeout=0.1,
-            )
-
-            try:
-                # Get the callback that was registered
-                callback = mock_manager.register_driver.call_args[1]["callback"]
-
-                # Initially not ready
-                assert not driver._ready.is_set()
-
-                # Call the callback
-                callback("CCPLEX: 0", "/dev/pts/5")
-
-                # Should now be ready
-                assert driver._ready.is_set()
-            finally:
-                driver.close()
-
-
 def test_nvdemux_gets_pts_from_manager():
     """Test that connect() gets pts path from manager."""
     with tempfile.NamedTemporaryFile() as device_file:
@@ -81,10 +50,6 @@ def test_nvdemux_gets_pts_from_manager():
             )
 
             try:
-                # Trigger callback to set ready
-                callback = mock_manager.register_driver.call_args[1]["callback"]
-                callback("CCPLEX: 0", "/dev/pts/5")
-
                 # Should call get_pts_path when checking pts availability
                 # (We can't test connect() easily without mocking serial, but we can test the logic)
                 pts_path = mock_manager.get_pts_path(str(driver.uuid))
@@ -114,28 +79,6 @@ def test_nvdemux_unregisters_on_close():
             mock_manager.unregister_driver.assert_called_once_with(driver_id)
 
 
-def test_nvdemux_timeout_no_callback():
-    """Test timeout when callback is never invoked."""
-    with tempfile.NamedTemporaryFile() as device_file:
-        with patch("jumpstarter_driver_pyserial.nvdemux.driver.DemuxerManager") as mock_manager_class:
-            mock_manager = MagicMock()
-            mock_manager_class.get_instance.return_value = mock_manager
-
-            driver = NVDemuxSerial(
-                demuxer_path="/usr/bin/demuxer",
-                device=device_file.name,
-                target="CCPLEX: 0",
-                timeout=0.1,
-            )
-
-            try:
-                # Callback is never invoked, so ready should not be set
-                time.sleep(0.2)
-                assert not driver._ready.is_set()
-            finally:
-                driver.close()
-
-
 def test_nvdemux_default_values():
     """Test default parameter values."""
     with tempfile.NamedTemporaryFile() as device_file:

packages/jumpstarter-driver-pyserial/jumpstarter_driver_pyserial/nvdemux/manager.py

@@ -80,7 +80,6 @@ class DriverInfo:
 
     driver_id: str
     target: str
-    callback: Callable[[str, str], None]  # (target, pts_path) -> None
 
 
 class DemuxerManager:
@@ -207,22 +206,13 @@ def _validate_config(self, demuxer_path: str, device: str, chip: str, target: st
             if existing_driver.target == target:
                 raise ValueError(f"Target '{target}' already registered by another driver")
 
-    def _get_ready_callback(self, target: str, callback: Callable[[str, str], None]) -> tuple[str, str] | None:
-        """Return callback args if target is already ready."""
-        if target in self._ready_targets:
-            pts_path = self._pts_map.get(target)
-            if pts_path:
-                return target, pts_path
-        return None
-
     def register_driver(
         self,
         driver_id: str,
         demuxer_path: str,
         device: str,
         chip: str,
         target: str,
-        callback: Callable[[str, str], None],
         poll_interval: float = 1.0,
     ) -> None:
         """Register a driver instance with the manager.
@@ -233,13 +223,11 @@ def register_driver(
             device: Device path or glob pattern
             chip: Chip type (T234 or T264)
             target: Target channel (e.g., "CCPLEX: 0")
-            callback: Function to call when target becomes ready
             poll_interval: Polling interval for device reconnection
 
         Raises:
             ValueError: If configuration doesn't match existing process
         """
-        notify_args: tuple[str, str] | None = None
         with self._lock:
             # Validate configuration matches existing process
             if self._drivers:
@@ -252,25 +240,15 @@ def register_driver(
                 self._poll_interval = poll_interval
 
             # Register the driver
-            driver_info = DriverInfo(driver_id=driver_id, target=target, callback=callback)
+            driver_info = DriverInfo(driver_id=driver_id, target=target)
             self._drivers[driver_id] = driver_info
 
             logger.debug("Registered driver %s for target '%s'", driver_id, target)
 
-            # If target is already ready, notify immediately
-            notify_args = self._get_ready_callback(target, callback)
-
             # Start monitor thread only once
             if not self._monitor_thread or not self._monitor_thread.is_alive():
                 self._start_monitor()
 
-        if notify_args:
-            # Invoke callbacks outside the lock to avoid deadlocks/reentrancy.
-            try:
-                callback(*notify_args)
-            except Exception as e:
-                logger.error("Error in driver callback: %s", e)
-
     def unregister_driver(self, driver_id: str) -> None:
         """Unregister a driver instance.
 
@@ -527,24 +505,10 @@ def _read_demuxer_output(self):
                 if pts_path and target:
                     logger.debug("Found pts path for target '%s': %s", target, pts_path)
 
-                    callback_to_invoke = None
                     with self._lock:
                         self._pts_map[target] = pts_path
                         self._ready_targets.add(target)
 
-                        # Find driver callback for this specific target
-                        for driver_info in self._drivers.values():
-                            if driver_info.target == target:
-                                callback_to_invoke = driver_info.callback
-                                break  # Only one driver per target
-
-                    if callback_to_invoke:
-                        # Invoke callbacks outside the lock to avoid deadlocks/reentrancy.
-                        try:
-                            callback_to_invoke(target, pts_path)
-                        except Exception as e:
-                            logger.error("Error in driver callback: %s", e)
-
         except Exception as e:
             logger.error("Error reading demuxer output: %s", e)