: v1/logging: avoid spinning up LogForwardActor mesh when redundant (#1722)

Summary:

this diff changes `LoggingMeshClient::spawn` so that log forwarding is no longer unconditional. previously we always spawned a `LogForwardActor` mesh across the `ProcMesh` and stored it in the client. now we read `MESH_ENABLE_LOG_FORWARDING` and only create that mesh if the flag is set. when forwarding is disabled we don’t spawn any `LogForwardActor` actors at all and we record `forwarder_mesh` as `None`. `logger_mesh` (the per-proc `LoggerRuntimeActor` that manages python logging state) is still always spawned. `client_actor` is still spawned locally in the caller's process.

because `forwarder_mesh` is now an `Option`, the rest of the API is updated to match the new contract. `set_mode` now either propagates `stream_to_client` to the forwarders when they exist, silently accepts"don’t stream" when they don’t, or returns an error if the caller tries to enable streaming in a configuration where we never created forwarders. `flush` now no-ops successfully in the case where there is no forwarding mesh instead of assuming that those actors are always present.

the diff also adds the minimal test harness needed to exercise this logic end-to-end. there is a dedicated bootstrap binary in `monarch_hyperactor` (`monarch_hyperactor_test_bootstrap`) which initializes python and then runs the mesh bootstrap protocol, so remote procs can safely construct `LoggerRuntimeActor `without panicking on `Python::with_gil`. we also add `ensure_python` and `AwaitPyExt `on the rust side so the tokio tests can stand up a `ProcMesh`, call the python-facing `spawn()`, await the returned `PyPythonTask`, and inspect the resulting `LoggingMeshClient`. finally, there is a test that brings up a tiny mesh, forces the forwarding flag off and on, and in each case asserts that `forwarder_mesh` is `None` or `Some(..)` accordingly. follow-up diffs will extend coverage to `set_mode`, `flush`, and shutdown semantics, now that the plumbing exists.

Differential Revision: D85919326

Author: shayne-fletcher

hyperactor_mesh/src/testresource.rs

@@ -27,15 +27,15 @@ where
     }
 
     assert!(
-        name.starts_with("monarch/hyperactor_mesh/"),
-        "invalid resource {}: must start with \"monarch/hyperactor_mesh/\"",
+        name.starts_with("monarch/monarch_hyperactor/"),
+        "invalid resource {}: must start with \"monarch/monarch_hyperactor/\"",
         name
     );
 
     let path: PathBuf = name
         .replace(
-            "monarch/hyperactor_mesh/",
-            "../target/debug/hyperactor_mesh_test_",
+            "monarch/monarch_hyperactor/",
+            "../target/debug/monarch_hyperactor_test_",
         )
         .into();
 

monarch_hyperactor/Cargo.toml

@@ -1,4 +1,4 @@
-# @generated by autocargo from //monarch/monarch_hyperactor:[monarch_hyperactor,process_allocator-oss,test_monarch_hyperactor]
+# @generated by autocargo from //monarch/monarch_hyperactor:[monarch_hyperactor,monarch_hyperactor_test_bootstrap,process_allocator-oss,test_monarch_hyperactor]
 
 [package]
 name = "monarch_hyperactor"
@@ -7,6 +7,11 @@ authors = ["Meta"]
 edition = "2021"
 license = "BSD-3-Clause"
 
+[[bin]]
+name = "monarch_hyperactor_test_bootstrap"
+path = "test/bootstrap.rs"
+edition = "2024"
+
 [[bin]]
 name = "process_allocator"
 edition = "2024"
@@ -52,6 +57,7 @@ tokio-util = { version = "0.7.15", features = ["full"] }
 tracing = { version = "0.1.41", features = ["attributes", "valuable"] }
 
 [dev-dependencies]
+buck-resources = "1"
 dir-diff = "0.3"
 
 [features]

monarch_hyperactor/src/pytokio.rs

@@ -50,6 +50,8 @@ use hyperactor::config::CONFIG;
 use hyperactor::config::ConfigAttr;
 use monarch_types::SerializablePyErr;
 use pyo3::IntoPyObjectExt;
+#[cfg(test)]
+use pyo3::PyClass;
 use pyo3::exceptions::PyRuntimeError;
 use pyo3::exceptions::PyStopIteration;
 use pyo3::exceptions::PyTimeoutError;
@@ -547,3 +549,52 @@ pub fn register_python_bindings(hyperactor_mod: &Bound<'_, PyModule>) -> PyResul
 
     Ok(())
 }
+
+/// Ensure the embedded Python interpreter is initialized exactly
+/// once.
+///
+/// Safe to call from multiple threads, multiple times.
+#[cfg(test)]
+pub(crate) fn ensure_python() {
+    static INIT: std::sync::OnceLock<()> = std::sync::OnceLock::new();
+    INIT.get_or_init(|| {
+        pyo3::prepare_freethreaded_python();
+    });
+}
+
+#[cfg(test)]
+// Helper: let us "await" a `PyPythonTask` in Rust.
+//
+// Semantics:
+//   - consume the `PyPythonTask`,
+//   - take the inner future,
+//   - `.await` it on tokio to get `Py<PyAny>`,
+//   - turn that into `Py<T>`.
+pub(crate) trait AwaitPyExt {
+    async fn await_py<T: PyClass>(self) -> Result<Py<T>, PyErr>;
+}
+
+#[cfg(test)]
+impl AwaitPyExt for PyPythonTask {
+    async fn await_py<T: PyClass>(mut self) -> Result<Py<T>, PyErr> {
+        // Take ownership of the inner future.
+        let fut = self
+            .take_task()
+            .expect("PyPythonTask already consumed in await_py");
+
+        // Await a Result<Py<PyAny>, PyErr>.
+        let py_any: Py<PyAny> = fut.await?;
+
+        // Convert Py<PyAny> -> Py<T>.
+        Python::with_gil(|py| {
+            let bound_any = py_any.bind(py);
+
+            // Try extract a Py<T>.
+            let obj: Py<T> = bound_any
+                .extract::<Py<T>>()
+                .expect("spawn() did not return expected Python type");
+
+            Ok(obj)
+        })
+    }
+}