more helpful warning for leaky backpressure; removed isinstance

Author: griffinmilsap

examples/ezmsg_leaky_subscriber.py

@@ -13,7 +13,8 @@
 # rather than processing a growing backlog of stale messages.
 
 import asyncio
-from collections.abc import AsyncGenerator
+import typing
+
 from dataclasses import dataclass, field
 
 import ezmsg.core as ez
@@ -40,9 +41,7 @@ class FastPublisher(ez.Unit):
     OUTPUT = ez.OutputStream(TimestampedMessage, num_buffers=32)
 
     @ez.publisher(OUTPUT)
-    async def publish(self) -> AsyncGenerator:
-        # Small delay to ensure subscriber is ready
-        await asyncio.sleep(0.5)
+    async def publish(self) -> typing.AsyncGenerator:
 
         for seq in range(self.SETTINGS.num_messages):
             msg = TimestampedMessage(seq=seq)
@@ -60,8 +59,8 @@ class SlowSubscriberSettings(ez.Settings):
 
 
 class SlowSubscriberState(ez.State):
+    received_seqs: list
     received_count: int = 0
-    received_seqs: list = None
     total_latency: float = 0.0
 
 
@@ -76,9 +75,8 @@ class SlowSubscriber(ez.Unit):
     SETTINGS = SlowSubscriberSettings
     STATE = SlowSubscriberState
 
-    # Leaky input stream with max queue of 3 messages
-    # When the queue fills up, oldest messages are dropped
-    INPUT = ez.InputStream(TimestampedMessage, leaky=True, max_queue=3)
+    # Leaky input stream; oldest messages are dropped
+    INPUT = ez.InputStream(TimestampedMessage, leaky=True)
 
     async def initialize(self) -> None:
         self.STATE.received_seqs = []

src/ezmsg/core/backendprocess.py

@@ -416,15 +416,14 @@ async def handle_subscriber(
     # allowing publishers to continue without blocking during slow processing.
     # Non-leaky subscribers use recv_zero_copy() to hold backpressure during
     # processing, which provides zero-copy performance but applies backpressure.
-    is_leaky = isinstance(sub._incoming, LeakyQueue)
 
     while True:
         if not callables:
             sub.close()
             await sub.wait_closed()
             break
 
-        if is_leaky:
+        if sub.leaky:
             msg = await sub.recv()
             try:
                 for callable in list(callables):

src/ezmsg/core/messagechannel.py

@@ -164,6 +164,8 @@ async def create(
         writer.write(Command.CHANNEL.value)
         writer.write(encode_str(id_str))
 
+        topic = await read_str(reader)
+
         shm = None
         shm_name = await read_str(reader)
         try:
@@ -184,6 +186,7 @@ async def create(
         assert num_buffers > 0, "publisher reports invalid num_buffers"
 
         chan = cls(UUID(id_str), pub_id, num_buffers, shm, graph_address, _guard=cls._SENTINEL)
+        chan.topic = topic
 
         chan._graph_task = asyncio.create_task(
             chan._graph_connection(graph_reader, graph_writer),

src/ezmsg/core/pubclient.py

@@ -329,6 +329,7 @@ async def _channel_connect(
         if cmd == Command.CHANNEL.value:
             channel_id_str = await read_str(reader)
             channel_id = UUID(channel_id_str)
+            writer.write(encode_str(self.topic))
             writer.write(encode_str(self._shm.name))
             shm_ok = await reader.read(1) == Command.SHM_OK.value
             pid = await read_int(reader)

src/ezmsg/core/subclient.py

@@ -35,6 +35,7 @@ class Subscriber:
 
     id: UUID
     topic: str
+    leaky: bool
 
     _graph_address: AddressType | None
     _graph_task: asyncio.Task[None]
@@ -124,11 +125,12 @@ def __init__(
             )
         self.id = id
         self.topic = topic
+        self.leaky = leaky
         self._graph_address = graph_address
 
         self._cur_pubs = set()
         self._channels = dict()
-        if leaky:
+        if self.leaky:
             self._incoming = LeakyQueue(
                 1 if max_queue is None else max_queue, self._handle_dropped_notification
             )
@@ -227,15 +229,12 @@ async def _graph_connection(
                             pub_id, self.id, self._incoming, self._graph_address
                         )
 
-                        if (
-                            isinstance(self._incoming, LeakyQueue)
-                            and self._incoming.maxsize >= channel.num_buffers
-                        ):
+                        if self.leaky and self._incoming.maxsize >= channel.num_buffers:
                             logger.warning(
                                 f"Leaky Subscriber {self.topic} may cause "
-                                f"backpressure in Publisher {channel.topic}."
+                                f"backpressure in Publisher {channel.topic}. "
                                 f"Subscriber's max queue size ({self._incoming.maxsize}) >= "
-                                f"Publisher's num_buffers ({channel.num_buffers})"
+                                f"Publisher's num_buffers ({channel.num_buffers})."
                             )
 
                         self._channels[pub_id] = channel