Add a Python multithread (INTERTHREAD) example

README.md

@@ -68,6 +68,50 @@ class BasicPublisher(SingleThreadApplication):
 
 Command line and configuration file handling are the same as for a C++ application, so `goby3_example_python_basic_publisher --help`, `--example_config` and `--my_value 10` all work as expected.
 
+### Multi-Threaded Publish/Subscribe in Python (INTERTHREAD)
+
+The Python equivalent of the interthread example above: a publisher thread produces
+`NavigationReport`s, a watcher thread consumes them and forwards each one to the interprocess
+layer, and the application prints an alert when the vehicle is deep. The code is in
+`src/interprocess/zeromq/python/basic_multithread`.
+
+```
+cd launch/interprocess/zeromq
+goby_launch -x python_multithread.launch
+```
+
+Threads are written in Python and launched by the application, mirroring how a C++ application
+launches `SimpleThread`s. The thread is constructed on the thread it runs on, so it subscribes in
+its constructor as a C++ thread does:
+
+```python
+class DepthWatcher(Thread):
+    def __init__(self):
+        super().__init__()
+        self.interthread().subscribe(NAV_GROUP, self.incoming_nav)
+
+class BasicMultithread(SingleThreadApplication):
+    def __init__(self):
+        super().__init__()
+        self.launch_thread(DepthWatcher)
+```
+
+The interthread layer is implemented in Python rather than through the C++
+`InterThreadTransporter`, which has three consequences the example is built to show:
+
+- The application is a `SingleThreadApplication`: the C++ side never sees these threads, so
+  `MultiThreadApplication` is not needed.
+- Interthread groups are not declared in `interface.yml` -- any string will do -- because nothing
+  is generated for the layer.
+- An interthread message is any Python object, not only a protobuf message. The depth alert here
+  is a `dataclass` and never becomes protobuf.
+
+The interprocess layer still belongs to the application's thread, but a thread publishes to it
+directly anyway: the publication is handed to the application behind the scenes, the way a C++
+thread's `InterProcessForwarder` hands its traffic to the portal. The application picks that up in
+its loop, so launching a thread raises the rate it is serviced at (10 Hz by default,
+`interthread_poll_frequency_hertz` to change it).
+
 ### Basic Multi-Process Single-Threaded Publish/Subscribe (UDP Multicast)
 
 This example is equivalent to the ZeroMQ example above but uses the UDP Multicast (udpm) interprocess transport instead. No central broker (`gobyd`) is required.

launch/interprocess/zeromq/python_multithread.launch

@@ -0,0 +1,5 @@
+#!goby_launch -x
+gobyd
+goby3_example_python_multithread --alert_depth 300
+# the reports the watcher thread forwards to interprocess, seen from outside the application
+goby3_example_basic_interprocess_subscriber

src/interprocess/zeromq/python/CMakeLists.txt

@@ -1 +1,2 @@
 add_subdirectory(basic_publisher_subscriber)
+add_subdirectory(basic_multithread)

src/interprocess/zeromq/python/basic_multithread/CMakeLists.txt

@@ -0,0 +1,16 @@
+goby_add_library(TARGET goby3_example_python_multithread_config
+  PROTOS multithread_config.proto
+  PROTOC_OUT_DIR ${project_INC_DIR}/python_multithread_config
+  LINK_LIBRARIES ${Protobuf_LIBRARIES} dccl)
+
+goby_add_python_app(TARGET goby3_example_python_multithread
+  INTERFACE_YML interface.yml
+  MAIN multithread.py
+  PYTHON_PROTOS multithread_config.proto ${project_INC_DIR}/messages/nav.proto
+  PROTO_IMPORT_DIRS ${project_INC_DIR} ${CMAKE_CURRENT_SOURCE_DIR}
+  PROTO_MODULES multithread_config_pb2 messages.nav_pb2
+  # headers for the group and type named in interface.yml; the interthread groups need none
+  INCLUDES goby/zeromq/application/single_thread.h
+           messages/groups.h messages/nav.pb.h
+           python_multithread_config/multithread_config.pb.h
+  LINK_LIBRARIES goby_zeromq goby3_example_messages goby3_example_python_multithread_config)

src/interprocess/zeromq/python/basic_multithread/interface.yml

@@ -0,0 +1,15 @@
+# Only the interprocess publication is declared. INTERTHREAD is implemented in Python and never
+# reaches the generated C++, so an interthread group is a plain string chosen by the application
+# rather than a C++ group named here -- and the application needs no MultiThreadApplication.
+application:
+  name: BasicPythonMultithread
+  cpp_type: goby::zeromq::SingleThreadApplication
+  config:
+    scheme: PROTOBUF
+    type: goby3_example.protobuf.BasicPythonMultithreadConfig
+
+interprocess:
+  publishes:
+    - group: groups::nav
+      scheme: PROTOBUF
+      type: protobuf.NavigationReport

src/interprocess/zeromq/python/basic_multithread/multithread.py

@@ -0,0 +1,114 @@
+#!/usr/bin/env python3
+"""Python version of src/interthread/basic_publisher_subscriber, showing the INTERTHREAD layer.
+
+A publisher thread produces NavigationReports, a watcher thread consumes them and forwards each
+one to the rest of the Goby system, and the application prints an alert when the vehicle is deep.
+
+Three things this example exists to show:
+
+  1. INTERTHREAD is pure Python. Goby carries these messages between threads without marshalling
+     them, so they never enter the generated C++ and interthread groups are not declared in
+     interface.yml -- any string will do. The application is a SingleThreadApplication: the C++
+     side has no idea these threads exist.
+  2. Because of that, an interthread message can be any Python object, not only a protobuf
+     message. The alert below is a dataclass.
+  3. INTERPROCESS belongs to the application's thread, but a thread publishes to it directly
+     anyway: the publication is handed to the application behind the scenes, as a C++ thread's
+     InterProcessForwarder hands its traffic to the portal.
+
+Run it as any other Goby application:
+
+    goby3_example_python_multithread --alert_depth 300
+"""
+
+import random
+import sys
+from dataclasses import dataclass
+
+import goby
+
+# generated from interface.yml by goby_add_python_app(); provides the application and thread base
+# classes, and the groups declared in the interface file
+from basic_python_multithread_goby import SingleThreadApplication, Thread, groups
+
+# protoc --python_out of src/messages/nav.proto
+from messages import nav_pb2
+
+# interthread groups are Python-side only, so they are named here rather than in interface.yml
+NAV_GROUP = "nav"
+DEPTH_ALERT_GROUP = "depth_alert"
+
+
+@dataclass
+class DepthAlert:
+    """An interthread message that is not a protobuf message, and never becomes one."""
+
+    depth: float
+    x: float
+    y: float
+
+
+class NavPublisher(Thread):
+    """Produces reports on the interthread layer at its own rate."""
+
+    def __init__(self):
+        super().__init__(loop_frequency_hertz=1)
+
+    def loop(self):
+        nav = nav_pb2.NavigationReport(
+            x=95 + random.randrange(20),
+            y=195 + random.randrange(20),
+            z=-305 + random.randrange(10),
+        )
+
+        print(f"[publisher] Tx: z={nav.z:.1f}")
+        self.interthread().publish(NAV_GROUP, nav)
+
+
+class DepthWatcher(Thread):
+    """Consumes reports, forwards them outside the process, and alerts on deep ones."""
+
+    def __init__(self):
+        # no loop frequency: this thread is woken by its subscription rather than by a clock
+        super().__init__()
+
+        # subscribing in the constructor is what a C++ SimpleThread does, and works for the same
+        # reason: the thread is constructed on the thread it runs on, so the callback lands here
+        self.interthread().subscribe(NAV_GROUP, self.incoming_nav)
+
+    def incoming_nav(self, nav: nav_pb2.NavigationReport):
+        print(f"[watcher] Rx: z={nav.z:.1f}")
+
+        # the interprocess portal lives on the application's thread; publishing to it from here
+        # hands it over rather than touching the portal
+        self.interprocess().publish(groups.nav, nav)
+
+        if -nav.z > self.cfg.alert_depth:
+            self.interthread().publish(
+                DEPTH_ALERT_GROUP, DepthAlert(depth=-nav.z, x=nav.x, y=nav.y)
+            )
+
+
+class BasicMultithread(SingleThreadApplication):
+    def __init__(self):
+        # no loop frequency: launching a thread raises the rate the application is serviced at,
+        # which is what picks up the alerts below
+        super().__init__()
+
+        print(f"Alerting below {self.cfg.alert_depth} m")
+
+        self.interthread().subscribe(DEPTH_ALERT_GROUP, self.incoming_depth_alert)
+
+        self.launch_thread(NavPublisher)
+        self.launch_thread(DepthWatcher)
+
+    # the annotation names the type this subscription takes, as it does for a protobuf message
+    def incoming_depth_alert(self, alert: DepthAlert):
+        print(f"[main] alert: {alert.depth:.1f} m at x={alert.x:.1f} y={alert.y:.1f}")
+
+
+# reads command line parameters based on the BasicPythonMultithreadConfig definition, exactly as a
+# C++ application does (try "--help" to see the parameters, or "--example_config" for the correct
+# configuration file syntax). Edit "multithread_config.proto" to add parameters.
+if __name__ == "__main__":
+    sys.exit(goby.run(BasicMultithread))

src/interprocess/zeromq/python/basic_multithread/multithread_config.proto

@@ -0,0 +1,16 @@
+syntax = "proto2";
+import "goby/middleware/protobuf/app_config.proto";
+import "goby/zeromq/protobuf/interprocess_config.proto";
+
+package goby3_example.protobuf;
+
+message BasicPythonMultithreadConfig
+{
+    // required parameters for ApplicationBase3 class
+    optional goby.middleware.protobuf.AppConfig app = 1;
+    // required parameters for connecting to 'gobyd'
+    optional goby.zeromq.protobuf.InterProcessPortalConfig interprocess = 2;
+
+    // depth (m) below which the watcher thread raises an alert
+    optional double alert_depth = 3 [default = 300];
+}