reference parity

Author: pavel-kirienko

src/pycyphal/_node.py

@@ -271,6 +271,22 @@ class Coupling:
     substitutions: list[tuple[str, int]]
 
 
+@dataclass
+class SharedSubjectListener:
+    """One transport listener shared by all topics bound to the same subject-ID."""
+
+    handle: Closable
+    owners: set[Topic] = field(default_factory=set)
+
+
+@dataclass
+class SharedSubjectWriter:
+    """One transport writer shared by all topics bound to the same subject-ID."""
+
+    handle: SubjectWriter
+    owners: set[Topic] = field(default_factory=set)
+
+
 @dataclass
 class PublishTracker:
     """Tracks a pending reliable publication awaiting ACKs."""
@@ -365,34 +381,32 @@ def tag_seqno(self, tag: int) -> int:
 
     def ensure_writer(self) -> SubjectWriter:
         if self.pub_writer is None:
-            self.pub_writer = self._node.transport.subject_advertise(self.subject_id)
-            _logger.info("Writer created for '%s' sid=%d", self._name, self.subject_id)
+            sid = self.subject_id
+            self.pub_writer = self._node.acquire_subject_writer(self, sid)
+            _logger.info("Writer acquired for '%s' sid=%d", self._name, sid)
         return self.pub_writer
 
     def ensure_listener(self) -> None:
         if self.sub_listener is None and self.couplings:
             sid = self.subject_id
-
-            def handler(arrival: TransportArrival) -> None:
-                self._node.on_subject_arrival(sid, arrival)
-
-            self.sub_listener = self._node.transport.subject_listen(sid, handler)
-            _logger.info("Listener created for '%s' sid=%d", self._name, sid)
+            self.sub_listener = self._node.acquire_subject_listener(self, sid)
+            _logger.info("Listener acquired for '%s' sid=%d", self._name, sid)
 
     def sync_listener(self) -> None:
         if self.couplings:
             self.ensure_listener()
         elif self.sub_listener is not None:
-            self.sub_listener.close()
+            self._node.release_subject_listener(self, self.subject_id)
             self.sub_listener = None
             _logger.info("Listener released for '%s'", self._name)
 
     def release_transport_handles(self) -> None:
+        sid = self.subject_id
         if self.pub_writer is not None:
-            self.pub_writer.close()
+            self._node.release_subject_writer(self, sid)
             self.pub_writer = None
         if self.sub_listener is not None:
-            self.sub_listener.close()
+            self._node.release_subject_listener(self, sid)
             self.sub_listener = None
 
     def compute_is_implicit(self) -> bool:
@@ -465,6 +479,8 @@ def broadcast_handler(arrival: TransportArrival) -> None:
         # Gossip shard state: lazily created per shard.
         self.gossip_shard_writers: dict[int, SubjectWriter] = {}
         self.gossip_shard_listeners: dict[int, Closable] = {}
+        self.shared_subject_writers: dict[int, SharedSubjectWriter] = {}
+        self.shared_subject_listeners: dict[int, SharedSubjectListener] = {}
 
         # Register unicast handler.
         transport.unicast_listen(self.on_unicast_arrival)
@@ -600,13 +616,11 @@ def topic_allocate(self, topic: TopicImpl, new_evictions: int, now: float) -> No
             elif left_wins(t.lage(now), t.hash, collider.lage(now), collider.hash):
                 # Our topic wins: take the slot, evict the collider.
                 t.release_transport_handles()
-                # Preserve the collider's writer if we can reuse it.
-                if collider.pub_writer is not None:
-                    t.pub_writer = collider.pub_writer
-                    collider.pub_writer = None
                 t.evictions = ev
                 del self.topics_by_subject_id[new_sid]
                 self.topics_by_subject_id[new_sid] = t
+                if collider.pub_writer is not None:
+                    t.pub_writer = self.acquire_subject_writer(t, new_sid)
                 t.sync_listener()
                 self.schedule_gossip_urgent(t)
                 # Schedule collider for reallocation.
@@ -708,6 +722,48 @@ def handler(arrival: TransportArrival) -> None:
             _logger.debug("Gossip shard writer/listener for sid=%d", shard_sid)
         return writer
 
+    def acquire_subject_writer(self, topic: TopicImpl, subject_id: int) -> SubjectWriter:
+        entry = self.shared_subject_writers.get(subject_id)
+        if entry is None:
+            entry = SharedSubjectWriter(handle=self.transport.subject_advertise(subject_id))
+            self.shared_subject_writers[subject_id] = entry
+            _logger.debug("Shared subject writer created sid=%d", subject_id)
+        entry.owners.add(topic)
+        return entry.handle
+
+    def release_subject_writer(self, topic: TopicImpl, subject_id: int) -> None:
+        entry = self.shared_subject_writers.get(subject_id)
+        if entry is None:
+            return
+        entry.owners.discard(topic)
+        if not entry.owners:
+            entry.handle.close()
+            del self.shared_subject_writers[subject_id]
+            _logger.debug("Shared subject writer released sid=%d", subject_id)
+
+    def acquire_subject_listener(self, topic: TopicImpl, subject_id: int) -> Closable:
+        entry = self.shared_subject_listeners.get(subject_id)
+        if entry is None:
+
+            def handler(arrival: TransportArrival) -> None:
+                self.on_subject_arrival(subject_id, arrival)
+
+            entry = SharedSubjectListener(handle=self.transport.subject_listen(subject_id, handler))
+            self.shared_subject_listeners[subject_id] = entry
+            _logger.debug("Shared subject listener created sid=%d", subject_id)
+        entry.owners.add(topic)
+        return entry.handle
+
+    def release_subject_listener(self, topic: TopicImpl, subject_id: int) -> None:
+        entry = self.shared_subject_listeners.get(subject_id)
+        if entry is None:
+            return
+        entry.owners.discard(topic)
+        if not entry.owners:
+            entry.handle.close()
+            del self.shared_subject_listeners[subject_id]
+            _logger.debug("Shared subject listener released sid=%d", subject_id)
+
     def schedule_gossip(self, topic: TopicImpl) -> None:
         """Start periodic gossip for an explicit topic."""
         if topic.gossip_task is not None:
@@ -1304,8 +1360,14 @@ def close(self) -> None:
             topic.release_transport_handles()
         self.broadcast_writer.close()
         self.broadcast_listener.close()
+        for shared_writer in list(self.shared_subject_writers.values()):
+            shared_writer.handle.close()
+        self.shared_subject_writers.clear()
+        for shared_listener in list(self.shared_subject_listeners.values()):
+            shared_listener.handle.close()
+        self.shared_subject_listeners.clear()
         for w in self.gossip_shard_writers.values():
             w.close()
-        for listener in self.gossip_shard_listeners.values():
-            listener.close()
+        for gossip_listener in self.gossip_shard_listeners.values():
+            gossip_listener.close()
         self.transport.close()

src/pycyphal/_transport.py

@@ -54,6 +54,7 @@ def subject_listen(self, subject_id: int, handler: Callable[[TransportArrival],
         """
         Subscribe to a subject to receive messages from it until the returned closable handle is closed.
         The session layer may request at most one listener per subject at any given time, similar to the reference impl.
+        Duplicate requests for the same subject should raise ValueError.
         Unlike the reference implementation, our listeners do not have the extent setting -- the extent mostly matters
         for high-reliability/real-time applications; this Python implementation assumes infinite extent.
         """
@@ -63,7 +64,8 @@ def subject_listen(self, subject_id: int, handler: Callable[[TransportArrival],
     def subject_advertise(self, subject_id: int) -> SubjectWriter:
         """
         Begin sending messages on a subject.
-        The session layer may request at most one listener per subject at any given time, similar to the reference impl.
+        The session layer may request at most one writer per subject at any given time, similar to the reference impl.
+        Duplicate requests for the same subject should raise ValueError.
         """
         raise NotImplementedError
 

src/pycyphal/udp.py

@@ -352,7 +352,10 @@ async def __call__(self, deadline: Instant, priority: Priority, message: bytes |
         _logger.debug("Subject tx done sid=%d tid=%d", self._subject_id, transfer_id)
 
     def close(self) -> None:
+        if self._closed:
+            return
         self._closed = True
+        self._transport.remove_subject_writer(self._subject_id, self)
         _logger.debug("Subject writer closed for subject %d", self._subject_id)
 
 
@@ -441,7 +444,8 @@ def __init__(
             self._self_endpoints.add(sock.getsockname()[:2])
 
         # Subject state
-        self._subject_handlers: dict[int, list[Callable[[TransportArrival], None]]] = {}
+        self._subject_handlers: dict[int, Callable[[TransportArrival], None]] = {}
+        self._subject_writers: dict[int, _UDPSubjectWriter] = {}
         self._mcast_socks: dict[tuple[int, int], socket.socket] = {}
         self._reassemblers: dict[int, _RxReassembler] = {}
 
@@ -521,23 +525,24 @@ def __repr__(self) -> str:
 
     def remove_subject_listener(self, subject_id: int, handler: Callable[[TransportArrival], None]) -> None:
         """
-        Remove a handler for a subject; clean up sockets/tasks if no handlers remain.
-        Internal use only.
+        Remove the handler for a subject; clean up sockets/tasks if none remains. Internal use only.
         """
-        handlers = self._subject_handlers.get(subject_id, [])
-        if handler in handlers:
-            handlers.remove(handler)
-        if not handlers:
-            self._subject_handlers.pop(subject_id, None)
-            self._reassemblers.pop(subject_id, None)
-            for i in range(len(self._interfaces)):
-                key = (subject_id, i)
-                task = self._mcast_rx_tasks.pop(key, None)
-                if task is not None:
-                    task.cancel()
-                sock = self._mcast_socks.pop(key, None)
-                if sock is not None:
-                    sock.close()
+        if self._subject_handlers.get(subject_id) is not handler:
+            return
+        self._subject_handlers.pop(subject_id, None)
+        self._reassemblers.pop(subject_id, None)
+        for i in range(len(self._interfaces)):
+            key = (subject_id, i)
+            task = self._mcast_rx_tasks.pop(key, None)
+            if task is not None:
+                task.cancel()
+            sock = self._mcast_socks.pop(key, None)
+            if sock is not None:
+                sock.close()
+
+    def remove_subject_writer(self, subject_id: int, writer: _UDPSubjectWriter) -> None:
+        if self._subject_writers.get(subject_id) is writer:
+            self._subject_writers.pop(subject_id, None)
 
     # -- Async sendto helper --
 
@@ -558,21 +563,25 @@ def subject_id_modulus(self) -> int:
         return self._subject_id_modulus_val
 
     def subject_listen(self, subject_id: int, handler: Callable[[TransportArrival], None]) -> Closable:
-        if subject_id not in self._subject_handlers:
-            _logger.info("Subscribing to subject %d", subject_id)
-            self._subject_handlers[subject_id] = []
-            for i, iface in enumerate(self._interfaces):
-                key = (subject_id, i)
-                sock = self._create_mcast_socket(subject_id, iface)
-                self._mcast_socks[key] = sock
-                task = self._loop.create_task(self._mcast_rx_loop(sock, subject_id, i))
-                self._mcast_rx_tasks[key] = task
-        self._subject_handlers[subject_id].append(handler)
+        if subject_id in self._subject_handlers:
+            raise ValueError(f"Subject {subject_id} already has an active listener")
+        _logger.info("Subscribing to subject %d", subject_id)
+        self._subject_handlers[subject_id] = handler
+        for i, iface in enumerate(self._interfaces):
+            key = (subject_id, i)
+            sock = self._create_mcast_socket(subject_id, iface)
+            self._mcast_socks[key] = sock
+            task = self._loop.create_task(self._mcast_rx_loop(sock, subject_id, i))
+            self._mcast_rx_tasks[key] = task
         return _UDPSubjectListener(self, subject_id, handler)
 
     def subject_advertise(self, subject_id: int) -> SubjectWriter:
+        if subject_id in self._subject_writers:
+            raise ValueError(f"Subject {subject_id} already has an active writer")
         _logger.info("Advertising subject %d", subject_id)
-        return _UDPSubjectWriter(self, subject_id)
+        writer = _UDPSubjectWriter(self, subject_id)
+        self._subject_writers[subject_id] = writer
+        return writer
 
     def unicast_listen(self, handler: Callable[[TransportArrival], None]) -> None:
         self._unicast_handler = handler
@@ -627,6 +636,7 @@ def close(self) -> None:
         self._mcast_socks.clear()
         self._tx_socks.clear()
         self._subject_handlers.clear()
+        self._subject_writers.clear()
         self._reassemblers.clear()
 
     # -- Internal async RX loops --
@@ -723,5 +733,6 @@ def _process_subject_datagram(
             arrival = TransportArrival(
                 timestamp=Instant.now(), priority=Priority(priority), remote_id=sender_uid, message=message
             )
-            for handler in self._subject_handlers.get(subject_id, []):
+            handler = self._subject_handlers.get(subject_id)
+            if handler is not None:
                 handler(arrival)

tests/mock_transport.py

@@ -36,11 +36,15 @@ async def __call__(self, deadline: Instant, priority: Priority, message: bytes |
         if self.transport.network is not None:
             self.transport.network.deliver_subject(self.subject_id, arrival, sender=self.transport)
         else:
-            for handler in self.transport.subject_handlers.get(self.subject_id, []):
+            handler = self.transport.subject_handlers.get(self.subject_id)
+            if handler is not None:
                 handler(arrival)
 
     def close(self) -> None:
+        if self.closed:
+            return
         self.closed = True
+        self.transport.remove_subject_writer(self.subject_id, self)
 
 
 class MockSubjectListener(Closable):
@@ -51,22 +55,22 @@ def __init__(self, transport: MockTransport, subject_id: int, handler: Callable[
         self.closed = False
 
     def close(self) -> None:
+        if self.closed:
+            return
         self.closed = True
-        handlers = self.transport.subject_handlers.get(self.subject_id, [])
-        if self.handler in handlers:
-            handlers.remove(self.handler)
-        if not handlers:
-            self.transport.subject_handlers.pop(self.subject_id, None)
+        self.transport.remove_subject_listener(self.subject_id, self.handler)
 
 
 class MockTransport(Transport):
     def __init__(self, node_id: int = 0, modulus: int = DEFAULT_MODULUS, network: MockNetwork | None = None) -> None:
         self.node_id = node_id
         self._modulus = modulus
         self.network = network
-        self.subject_handlers: dict[int, list[Callable[[TransportArrival], None]]] = {}
+        self.subject_handlers: dict[int, Callable[[TransportArrival], None]] = {}
+        self.subject_listener_creations: dict[int, int] = {}
         self.unicast_handler: Callable[[TransportArrival], None] | None = None
         self.writers: dict[int, MockSubjectWriter] = {}
+        self.subject_writer_creations: dict[int, int] = {}
         self.unicast_log: list[tuple[int, bytes]] = []
         self.closed = False
         self.fail_unicast = False
@@ -82,16 +86,28 @@ def subject_id_modulus(self) -> int:
         return self._modulus
 
     def subject_listen(self, subject_id: int, handler: Callable[[TransportArrival], None]) -> Closable:
-        if subject_id not in self.subject_handlers:
-            self.subject_handlers[subject_id] = []
-        self.subject_handlers[subject_id].append(handler)
+        if subject_id in self.subject_handlers:
+            raise ValueError(f"Subject {subject_id} already has an active listener")
+        self.subject_handlers[subject_id] = handler
+        self.subject_listener_creations[subject_id] = self.subject_listener_creations.get(subject_id, 0) + 1
         return MockSubjectListener(self, subject_id, handler)
 
     def subject_advertise(self, subject_id: int) -> MockSubjectWriter:
+        if subject_id in self.writers:
+            raise ValueError(f"Subject {subject_id} already has an active writer")
         writer = MockSubjectWriter(self, subject_id)
         self.writers[subject_id] = writer
+        self.subject_writer_creations[subject_id] = self.subject_writer_creations.get(subject_id, 0) + 1
         return writer
 
+    def remove_subject_listener(self, subject_id: int, handler: Callable[[TransportArrival], None]) -> None:
+        if self.subject_handlers.get(subject_id) is handler:
+            self.subject_handlers.pop(subject_id, None)
+
+    def remove_subject_writer(self, subject_id: int, writer: MockSubjectWriter) -> None:
+        if self.writers.get(subject_id) is writer:
+            self.writers.pop(subject_id, None)
+
     def unicast_listen(self, handler: Callable[[TransportArrival], None]) -> None:
         self.unicast_handler = handler
 
@@ -118,7 +134,8 @@ def close(self) -> None:
         self.closed = True
 
     def deliver_subject(self, subject_id: int, arrival: TransportArrival) -> None:
-        for handler in self.subject_handlers.get(subject_id, []):
+        handler = self.subject_handlers.get(subject_id)
+        if handler is not None:
             handler(arrival)
 
     def deliver_unicast(self, arrival: TransportArrival) -> None: