Merge pull request #198 from ezmsg-org/griff/working

Backend Rework: Message Channels

Author: griffinmilsap

.git-blame-ignore-revs

@@ -2,3 +2,5 @@
 5b4c220154d33cac15a75d3d7d978a75e67f2b8a
 # Ran `black` formatter on entire codebase and fixed non-standard line-endings with `dos2unix`.
 4e4f20be40a73f2162a565732bae76ea0c812739
+# chore: ruff formatting
+5ca5711e7714042f23d1286abf946217d75318c2

docs/source/conf.py

@@ -1,7 +1,6 @@
 # Configuration file for the Sphinx documentation builder.
 
 import os
-import sys
 
 # -- Project information --------------------------
 
@@ -75,7 +74,7 @@
 html_static_path = ["_static"]
 
 # Timestamp is inserted at every page bottom in this strftime format.
-html_last_updated_fmt = '%Y-%m-%d'
+html_last_updated_fmt = "%Y-%m-%d"
 
 # -- Options for EPUB output --------------------------
 epub_show_urls = "footnote"
@@ -95,8 +94,10 @@ def linkcode_resolve(domain, info):
     else:
         return f"{code_url}src/{filename}.py"
 
+
 # -- Options for graphviz -----------------------------
 graphviz_output_format = "svg"
 
+
 def setup(app):
-    app.add_css_file("custom.css")
+    app.add_css_file("custom.css")

examples/ezmsg_attach.py

@@ -5,4 +5,4 @@
 if __name__ == "__main__":
     print("This example attaches to the system created/run by ezmsg_toy.py.")
     log = DebugLog()
-    ez.run(log, connections=(("TestSystem/PING/OUTPUT", log.INPUT),))
+    ez.run(LOG=log, connections=(("GLOBAL_PING_TOPIC", log.INPUT),))

examples/ezmsg_configs.py

@@ -239,5 +239,5 @@ def network(self) -> ez.NetworkDefinition:
     ]
 
     for system in test_systems:
-        ez.logger.info(f"Testing { system.__name__ }")
+        ez.logger.info(f"Testing {system.__name__}")
         ez.run(system())

examples/ezmsg_toy.py

@@ -192,8 +192,8 @@ def process_components(self):
 
     ez.run(
         SYSTEM=system,
-        # connections = [
-        #     ( system.PING.OUTPUT, 'PING_OUTPUT' ),
-        #     ( 'FOO_SUB', system.FOOSUB.INPUT )
-        # ]
+        connections=[
+            # Make PING.OUTPUT available on a topic ezmsg_attach.py
+            (system.PING.OUTPUT, "GLOBAL_PING_TOPIC"),
+        ],
     )

examples/lowlevel_api.py

@@ -0,0 +1,119 @@
+import asyncio
+
+import ezmsg.core as ez
+
+PORT = 12345
+MAX_COUNT = 100
+TOPIC = "/TEST"
+
+
+async def handle_pub(pub: ez.Publisher) -> None:
+    print("Publisher Task Launched")
+
+    count = 0
+
+    while True:
+        await pub.broadcast(f"{count=}")
+        await asyncio.sleep(0.1)
+        count += 1
+        if count >= MAX_COUNT:
+            break
+
+    print("Publisher Task Concluded")
+
+
+async def handle_sub(sub: ez.Subscriber) -> None:
+    print("Subscriber Task Launched")
+
+    rx_count = 0
+    while True:
+        async with sub.recv_zero_copy() as msg:
+            # Uncomment if you want to witness backpressure!
+            # await asyncio.sleep(0.15)
+            print(msg)
+
+        rx_count += 1
+        if rx_count >= MAX_COUNT:
+            break
+
+    print("Subscriber Task Concluded")
+
+
+async def host(host: str = "127.0.0.1"):
+    # Manually create a GraphServer
+    server = ez.GraphServer()
+    server.start((host, PORT))
+
+    print(f"Created GraphServer @ {server.address}")
+
+    try:
+        test_pub = await ez.Publisher.create(TOPIC, (host, PORT), host=host)
+        test_sub1 = await ez.Subscriber.create(TOPIC, (host, PORT))
+        test_sub2 = await ez.Subscriber.create(TOPIC, (host, PORT))
+
+        await asyncio.sleep(1.0)
+
+        pub_task = asyncio.Task(handle_pub(test_pub))
+        sub_task_1 = asyncio.Task(handle_sub(test_sub1))
+        sub_task_2 = asyncio.Task(handle_sub(test_sub2))
+
+        await asyncio.wait([pub_task, sub_task_1, sub_task_2])
+
+        test_pub.close()
+        test_sub1.close()
+        test_sub2.close()
+
+        for future in asyncio.as_completed(
+            [
+                test_pub.wait_closed(),
+                test_sub1.wait_closed(),
+                test_sub2.wait_closed(),
+            ]
+        ):
+            await future
+
+    finally:
+        server.stop()
+
+    print("Done")
+
+
+async def attach_client(host: str = "127.0.0.1"):
+
+    sub = await ez.Subscriber.create(TOPIC, (host, PORT))
+
+    try:
+        while True:
+            async with sub.recv_zero_copy() as msg:
+                # Uncomment if you want to see EXTREME backpressure!
+                # await asyncio.sleep(1.0)
+                print(msg)
+
+    except asyncio.CancelledError:
+        pass
+
+    finally:
+        sub.close()
+        await sub.wait_closed()
+        print("Detached")
+
+
+if __name__ == "__main__":
+    from dataclasses import dataclass
+    from argparse import ArgumentParser
+
+    parser = ArgumentParser()
+    parser.add_argument("--attach", action="store_true", help="attach to running graph")
+    parser.add_argument("--host", default="0.0.0.0", help="hostname for graphserver")
+
+    @dataclass
+    class Args:
+        attach: bool
+        host: str
+
+    args = Args(**vars(parser.parse_args()))
+
+    if args.attach:
+        asyncio.run(attach_client(host=args.host))
+    else:
+        asyncio.run(host(host=args.host))

pyproject.toml

@@ -24,6 +24,7 @@ dev = [
   {include-group = "lint"},
   {include-group = "test"},
   "pre-commit>=4.3.0",
+  "viztracer>=1.0.4",
 ]
 lint = [
     "flake8>=7.3.0",
@@ -35,6 +36,7 @@ test = [
     "pytest-asyncio>=1.1.0",
     "pytest-cov>=6.2.1",
     "xarray>=2025.6.1",
+    "psutil>=7.1.0",
 ]
 docs = [
   {include-group = "axisarray"},
@@ -51,6 +53,7 @@ axisarray = [
 
 [project.scripts]
 ezmsg = "ezmsg.core.command:cmdline"
+ezmsg-perf = "ezmsg.util.perf.command:command"
 
 [project.optional-dependencies]
 axisarray = [

src/ezmsg/core/__init__.py

@@ -24,6 +24,8 @@
     "GraphServer",
     "GraphContext",
     "run_command",
+    "Publisher",
+    "Subscriber",
     # All following are deprecated
     "System",
     "run_system",
@@ -42,6 +44,8 @@
 from .graphserver import GraphServer
 from .graphcontext import GraphContext
 from .command import run_command
+from .pubclient import Publisher
+from .subclient import Subscriber
 
 # Following imports are deprecated
 from .backend import run_system

src/ezmsg/core/addressable.py

@@ -4,17 +4,18 @@
 class Addressable:
     """
     Base class for objects that can be addressed within the ezmsg system.
-    
-    Addressable objects have a hierarchical address structure consisting of 
+
+    Addressable objects have a hierarchical address structure consisting of
     a location path and a name, similar to a filesystem path.
     """
+
     _name: str | None
     _location: list[str] | None
 
     def __init__(self) -> None:
         """
         Initialize an Addressable object.
-        
+
         The name and location are initially None and must be set before
         the object can be properly addressed. This is achieved through
         the ``_set_name()`` and ``_set_location()`` methods.
@@ -38,7 +39,7 @@ def _set_location(self, location: list[str] | None = None):
     def name(self) -> str:
         """
         Get the name of this addressable object.
-        
+
         :return: The object's name
         :rtype: str
         :raises AssertionError: If name has not been set
@@ -51,7 +52,7 @@ def name(self) -> str:
     def location(self) -> list[str]:
         """
         Get the location path of this addressable object.
-        
+
         :return: List of path components representing the object's location
         :rtype: list[str]
         :raises AssertionError: If location has not been set
@@ -64,10 +65,10 @@ def location(self) -> list[str]:
     def address(self) -> str:
         """
         Get the full address of this object.
-        
+
         The address is constructed by joining the location path and name
         with forward slashes, similar to a filesystem path.
-        
+
         :return: The full address string
         :rtype: str
         """