Merge branch 'main' into claude/interthread-python-wrapper-r75m1y

Author: tsaubergine

README.md

@@ -152,7 +152,7 @@ goby_launch -x julia_publisher_subscriber.launch
 
 This launches `gobyd`, the Julia subscriber, and the Julia publisher, each in their own XTerm window. When finished, type CTRL-C into the original terminal (the one from which `goby_launch` was run).
 
-The publisher sends a `NavigationReport` at 10 Hz on the `groups::julia_nav` group, which the subscriber prints as it receives them. Note that the Julia code identifies a group by its string value: `GOBY_DEFINE_GROUP` in `src/messages/groups.h` sets that string to the fully-qualified C++ name, so the group named in `interface.yml` and the string passed to `Goby.publish()`/`Goby.subscribe()` are the same text.
+The publisher sends a `NavigationReport` at 10 Hz on the `groups::julia_nav` group, which the subscriber prints as it receives them. The group is named through the module `goby_add_julia_app()` generates from `interface.yml` beside each application (`basic_julia_publisher_goby.jl`), so `groups.julia_nav` and `interprocess()` stand in for the group string and the layer constant. The expression then lives in one place, and a mistyped group is an `UndefVarError` naming it rather than a `GOBY_JULIA_FAIL` that terminates the application once the publish is reached.
 
 Unlike the C++ applications, the Julia applications take the path to their configuration file as their only argument, so `basic_julia_publisher.pb.cfg` and `basic_julia_subscriber.pb.cfg` are provided alongside the launch file:
 
@@ -161,6 +161,23 @@ basic_julia_publisher <config.pb.cfg>
 basic_julia_subscriber <config.pb.cfg>
 ```
 
+### Multi-Threaded Publish/Subscribe in Julia (ZeroMQ)
+
+`src/interprocess/zeromq/julia/basic_multithread` is the Julia equivalent of the interthread publisher/subscriber, and shows the layers available to a Julia task.
+
+`Goby.run(app, Main, [PublisherTask, SubscriberTask])` runs each task module on its own Julia thread. The publisher task sends `NavigationReport`s over `INTERTHREAD`, which is implemented in Julia over `Channel`s rather than in the generated C++ -- so its groups are plain strings chosen by the application rather than groups declared in `interface.yml`, and its messages can be any Julia value (the depth alert `Main` receives is a `NamedTuple`).
+
+The subscriber task publishes the deep reports out to `INTERPROCESS` itself. Goby.jl hands those to the task that owns the C++ application, the way a C++ thread's `InterProcessForwarder` does, so no task needs a portal of its own and none has to route through `Main`. Subscribing works the same way: the publisher task subscribes to the same group and sees each report that goes out, alongside the separate `basic_julia_subscriber` process.
+
+Julia fixes its thread count at startup, so the application needs one thread per task module plus three, for `Main`, the loop timer and the C++ application. `goby_add_julia_app()`'s `THREADS` gives that to the generated launcher.
+
+To run:
+
+```
+cd launch/interprocess/zeromq
+goby_launch -x julia_multithread.launch
+```
+
 ### GPS Driver
 A working example using a standard NMEA-0183 GPS is given in `src/interprocess/zeromq/gps_driver`.
 

launch/interprocess/zeromq/julia_multithread.launch

@@ -2,5 +2,5 @@
 gobyd
 # the launcher starts julia with the thread count declared by THREADS in CMakeLists.txt
 basic_julia_multithread basic_julia_multithread.pb.cfg
-# the reports the Julia application forwards to interprocess, seen from outside it
+# the deep reports the Julia application's subscriber task publishes, seen from outside it
 basic_julia_subscriber basic_julia_subscriber.pb.cfg

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

@@ -1,6 +1,9 @@
-# Only the interprocess publication is declared. INTERTHREAD is implemented in Julia, using
-# channels between tasks, 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.
+# Only the interprocess traffic is declared. INTERTHREAD is implemented in Julia, using channels
+# between tasks, 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.
+#
+# The application both publishes and subscribes to groups::julia_nav: the subscriber task sends
+# deep reports out, and the publisher task sees each one that goes.
 application:
   name: BasicJuliaMultithread
   cpp_type: goby.zeromq.SingleThreadApplication
@@ -13,3 +16,7 @@ interprocess:
     - group: groups::julia_nav
       scheme: PROTOBUF
       type: goby3_examples::protobuf::NavigationReport
+  subscribes:
+    - group: groups::julia_nav
+      scheme: PROTOBUF
+      type: goby3_examples::protobuf::NavigationReport

src/interprocess/zeromq/julia/basic_multithread/multithread.jl

@@ -1,20 +1,23 @@
-# Julia equivalent of src/interthread/basic_publisher_subscriber, showing the INTERTHREAD layer.
+# Julia equivalent of src/interthread/basic_publisher_subscriber, showing the INTERTHREAD layer
+# and pub/sub from a task rather than from Main.
 #
-# A publisher task produces NavigationReports, a subscriber task consumes them and raises an
-# alert when the vehicle is deep, and Main forwards each report on to INTERPROCESS so the rest
-# of the Goby system sees it.
+# A publisher task produces NavigationReports on INTERTHREAD; a subscriber task consumes them,
+# and publishes the deep ones straight out to INTERPROCESS. Those reach every application on the
+# system that subscribed to the group (basic_julia_subscriber, in julia_multithread.launch), this
+# one included: the publisher task subscribed to the same group and sees each one.
 #
-# Three things this example exists to show:
+# Four things this example exists to show:
 #
 #   1. INTERTHREAD is pure Julia. Goby.jl carries these messages between tasks over Channels,
 #      so they never enter the generated C++ and interthread groups are not declared in
 #      interface.yml -- any string will do.
 #   2. Because of that, an interthread message can be any Julia value, not only a protobuf
-#      message. The alert below is a NamedTuple.
-#   3. INTERPROCESS is only available from Main, so a task that needs the outside world hands
-#      the message to Main over INTERTHREAD, as the subscriber task does here. Publishing to it
-#      from a task instead raises an AssertionError naming the task, which stops the
-#      application.
+#      message. The alert Main receives below is a NamedTuple.
+#   3. INTERPROCESS works from any task. Goby.jl hands the message to the task that owns the C++
+#      application, the way a C++ thread's InterProcessForwarder does, so a task neither needs a
+#      portal of its own nor has to route through Main.
+#   4. Subscribing from a task works the same way: the subscription is registered on the task
+#      that owns the application, and messages come back to the task that asked for them.
 #
 # Julia fixes its thread count at startup, and Goby.run() needs one thread per task module plus
 # three (Main, the loop timer, and the C++ application). Two task modules therefore need five,
@@ -26,12 +29,17 @@ include(joinpath(@__DIR__, "..", "goby_application_init.jl"))
 init_cxxwrap("basic_julia_multithread")
 include_protos(@__MODULE__, "goby3_examples")
 
+# the groups and layer accessors generated from interface.yml, written beside this script by
+# goby_add_julia_app()
+include(joinpath(@__DIR__, "basic_julia_multithread_goby.jl"))
+using .BasicJuliaMultithreadGoby: groups, interprocess
+
 # interthread groups are Julia-side only, so they are named here rather than in interface.yml
 const NAV_GROUP = "nav"
 const DEPTH_ALERT_GROUP = "depth_alert"
 
 # ---------------------------------------------------------------------------
-# Publisher task: runs on its own thread, publishing at the rate in goby_cfg.
+# Publisher task: produces reports, and watches the ones that reach interprocess.
 # ---------------------------------------------------------------------------
 module PublisherTask
 
@@ -48,6 +56,15 @@ function loop()
     Goby.publish(Main.app, Goby.INTERTHREAD, Main.NAV_GROUP, nav)
 end
 
+function published_nav(nav::Main.goby3_examples.protobuf.NavigationReport)
+    println("[publisher] on the wire: z=$(round(nav.z, digits = 1))")
+end
+
+# subscriptions belong in start(), which Goby.run() calls on this task's own thread
+function start()
+    Goby.subscribe(Main.app, Main.interprocess(), Main.groups.julia_nav, published_nav)
+end
+
 # read by Goby.run() to drive loop() from the timer thread
 goby_cfg = Dict(
     :loop_frequency => 1,   # Hz
@@ -57,7 +74,7 @@ goby_cfg = Dict(
 end # module PublisherTask
 
 # ---------------------------------------------------------------------------
-# Subscriber task: consumes reports and reports deep ones back to Main.
+# Subscriber task: consumes reports, and sends the deep ones to the outside.
 # ---------------------------------------------------------------------------
 module SubscriberTask
 
@@ -67,38 +84,35 @@ function incoming_nav(nav::Main.goby3_examples.protobuf.NavigationReport)
     println("[subscriber] Rx: z=$(round(nav.z, digits = 1))")
 
     if -nav.z > Main.alert_depth
+        # straight out of this task: no portal here, and no hop through Main
+        Goby.publish(Main.app, Main.interprocess(), Main.groups.julia_nav, nav)
+
         # an interthread message is any Julia value; this one never becomes protobuf
         Goby.publish(Main.app, Goby.INTERTHREAD, Main.DEPTH_ALERT_GROUP,
                      (depth = -nav.z, x = nav.x, y = nav.y))
     end
 end
 
-# subscriptions belong in start(), which Goby.run() calls on this task's own thread
 function start()
     Goby.subscribe(Main.app, Goby.INTERTHREAD, Main.NAV_GROUP, incoming_nav)
 end
 
 end # module SubscriberTask
 
 # ---------------------------------------------------------------------------
-# Main: bridges the tasks to the rest of the Goby system.
+# Main: reports the alerts the subscriber task raises.
 # ---------------------------------------------------------------------------
 
-function forward_nav(nav::goby3_examples.protobuf.NavigationReport)
-    Goby.publish(app, Goby.INTERPROCESS, "groups::julia_nav", nav)
-end
-
 function incoming_depth_alert(alert)
     println("[main] alert: $(round(alert.depth, digits = 1)) m at " *
             "x=$(round(alert.x, digits = 1)) y=$(round(alert.y, digits = 1))")
 end
 
 function start()
-    Goby.subscribe(app, Goby.INTERTHREAD, NAV_GROUP, forward_nav)
     Goby.subscribe(app, Goby.INTERTHREAD, DEPTH_ALERT_GROUP, incoming_depth_alert)
 end
 
-# how often the C++ application drains the channel Main publishes to it through
+# how often the C++ application drains the channels the tasks publish to it through
 goby_cfg = Dict(:cxx_channel_check_frequency => 10)
 
 config_str = Goby.read_cli_cfg()

src/interprocess/zeromq/julia/basic_publisher/publisher.jl

@@ -9,6 +9,11 @@ include(joinpath(@__DIR__, "..", "goby_application_init.jl"))
 init_cxxwrap("basic_julia_publisher")
 include_protos(@__MODULE__, "goby3_examples")
 
+# the groups and layer accessors generated from interface.yml, written beside this script by
+# goby_add_julia_app()
+include(joinpath(@__DIR__, "basic_julia_publisher_goby.jl"))
+using .BasicJuliaPublisherGoby: groups, interprocess
+
 # ---------------------------------------------------------------------------
 # Publisher loop – called at the frequency specified in goby_cfg below.
 # ---------------------------------------------------------------------------
@@ -20,9 +25,7 @@ function loop()
     )
 
     println("Tx: x=$(nav.x) y=$(nav.y) z=$(nav.z)")
-    # the group is given by its string value, which GOBY_DEFINE_GROUP sets to
-    # the fully-qualified C++ name (see src/messages/groups.h)
-    Goby.publish(Main.app, Goby.INTERPROCESS, "groups::julia_nav", nav)
+    Goby.publish(Main.app, interprocess(), groups.julia_nav, nav)
 end
 
 config_str = Goby.read_cli_cfg()

src/interprocess/zeromq/julia/basic_subscriber/subscriber.jl

@@ -10,6 +10,11 @@ include(joinpath(@__DIR__, "..", "goby_application_init.jl"))
 init_cxxwrap("basic_julia_subscriber")
 include_protos(@__MODULE__, "goby3_examples")
 
+# the groups and layer accessors generated from interface.yml, written beside this script by
+# goby_add_julia_app()
+include(joinpath(@__DIR__, "basic_julia_subscriber_goby.jl"))
+using .BasicJuliaSubscriberGoby: groups, interprocess
+
 # ---------------------------------------------------------------------------
 # Called by Goby for each NavigationReport received. The message type is
 # inferred by Goby.subscribe() from this argument type.
@@ -20,9 +25,7 @@ end
 
 # start() is called by Goby.run() before entering the event loop.
 function start()
-    # the group is given by its string value, which GOBY_DEFINE_GROUP sets to
-    # the fully-qualified C++ name (see src/messages/groups.h)
-    Goby.subscribe(Main.app, Goby.INTERPROCESS, "groups::julia_nav", receive_incoming_msg)
+    Goby.subscribe(Main.app, interprocess(), groups.julia_nav, receive_incoming_msg)
 end
 
 config_str = Goby.read_cli_cfg()