diff --git a/README.md b/README.md index 41b5272..349453d 100644 --- a/README.md +++ b/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. diff --git a/launch/interprocess/zeromq/python_multithread.launch b/launch/interprocess/zeromq/python_multithread.launch new file mode 100644 index 0000000..361f904 --- /dev/null +++ b/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 diff --git a/src/interprocess/zeromq/python/CMakeLists.txt b/src/interprocess/zeromq/python/CMakeLists.txt index 381d98f..816de8f 100644 --- a/src/interprocess/zeromq/python/CMakeLists.txt +++ b/src/interprocess/zeromq/python/CMakeLists.txt @@ -1 +1,2 @@ add_subdirectory(basic_publisher_subscriber) +add_subdirectory(basic_multithread) diff --git a/src/interprocess/zeromq/python/basic_multithread/CMakeLists.txt b/src/interprocess/zeromq/python/basic_multithread/CMakeLists.txt new file mode 100644 index 0000000..3a52b4a --- /dev/null +++ b/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) diff --git a/src/interprocess/zeromq/python/basic_multithread/interface.yml b/src/interprocess/zeromq/python/basic_multithread/interface.yml new file mode 100644 index 0000000..339447c --- /dev/null +++ b/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 diff --git a/src/interprocess/zeromq/python/basic_multithread/multithread.py b/src/interprocess/zeromq/python/basic_multithread/multithread.py new file mode 100755 index 0000000..a8293bf --- /dev/null +++ b/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)) diff --git a/src/interprocess/zeromq/python/basic_multithread/multithread_config.proto b/src/interprocess/zeromq/python/basic_multithread/multithread_config.proto new file mode 100644 index 0000000..2c3ceaf --- /dev/null +++ b/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]; +}