[monarch] Revert D83360166 and D83489725

Pull Request resolved: #1408

Disable v1 log forwarding with LoggingManager and reenable @ShayneFletcher's auto-spawn mechanism in an attempt to get github CI stable again.
ghstack-source-id: 313909163
@exported-using-ghexport

Differential Revision: [D83777450](https://our.internmc.facebook.com/intern/diff/D83777450/)

**NOTE FOR REVIEWERS**: This PR has internal Meta-specific changes or comments, please review them on [Phabricator](https://our.internmc.facebook.com/intern/diff/D83777450/)!

Author: samlurye

hyperactor_mesh/src/bootstrap.rs

@@ -2226,20 +2226,16 @@ mod tests {
             proc.spawn("log_client", ()).await.unwrap().bind();
         log_client.set_aggregate(&client, None).await.unwrap();
 
-        // Spawn the forwarder in this proc (it will serve
-        // BOOTSTRAP_LOG_CHANNEL).
+        // Spawn the forwarder in this proc (it will serve BOOTSTRAP_LOG_CHANNEL).
         let _log_forwarder: ActorRef<LogForwardActor> = proc
             .spawn("log_forwarder", log_client.clone())
             .await
             .unwrap()
             .bind();
 
-        // Dial the channel but don't post until we know the forwarder
-        // is receiving.
-        let tx = channel::dial::<LogMessage>(log_channel.clone()).unwrap();
-
         // Send a fake log message as if it came from the proc
         // manager's writer.
+        let tx = channel::dial::<LogMessage>(log_channel.clone()).unwrap();
         tx.post(LogMessage::Log {
             hostname: "testhost".into(),
             pid: 12345,
@@ -2248,6 +2244,8 @@ mod tests {
         });
 
         // Assert we see it via the tap.
+        // Give it up to 2 seconds to travel through forwarder ->
+        // client -> print_log_line -> tap.
         let line = RealClock
             .timeout(Duration::from_secs(2), tap_rx.recv())
             .await

hyperactor_mesh/src/proc_mesh/mesh_agent.rs

@@ -212,6 +212,25 @@ impl ProcMeshAgent {
     }
 
     pub(crate) async fn boot_v1(proc: Proc) -> Result<ActorHandle<Self>, anyhow::Error> {
+        // Spawn a LogClientActor in this proc. It aggregates and
+        // prints logs coming from our stdout/stderr (via the parent's
+        // log writers). This is the sink for all forwarded LogMessage
+        // traffic.
+        let log_client_ref = proc
+            .spawn("log_client", ())
+            .await?
+            .bind::<crate::logging::LogClientActor>();
+
+        // Spawn a LogForwardActor. It serves BOOTSTRAP_LOG_CHANNEL
+        // (set by the parent ProcManager) and forwards any LogMessage
+        // it receives there into the above LogClientActor. Together
+        // these give us structured log forwarding without blocking
+        // pipes.
+        let _log_fwd_ref = proc
+            .spawn("log_forwarder", log_client_ref.clone())
+            .await?
+            .bind::<crate::logging::LogForwardActor>();
+
         let agent = ProcMeshAgent {
             proc: proc.clone(),
             remote: Remote::collect(),

monarch_hyperactor/src/v1/logging.rs

@@ -17,7 +17,6 @@ use hyperactor_mesh::logging::LogClientMessage;
 use hyperactor_mesh::logging::LogForwardActor;
 use hyperactor_mesh::logging::LogForwardMessage;
 use hyperactor_mesh::v1::ActorMesh;
-use hyperactor_mesh::v1::Name;
 use hyperactor_mesh::v1::actor_mesh::ActorMeshRef;
 use ndslice::View;
 use pyo3::Bound;
@@ -81,10 +80,7 @@ impl LoggingMeshClient {
         PyPythonTask::new(async move {
             let client_actor: ActorHandle<LogClientActor> =
                 instance_dispatch!(instance, async move |cx_instance| {
-                    cx_instance
-                        .proc()
-                        .spawn(&Name::new("log_client").to_string(), ())
-                        .await
+                    cx_instance.proc().spawn("log_client", ()).await
                 })?;
             let client_actor_ref = client_actor.bind();
             let forwarder_mesh = instance_dispatch!(instance, async |cx_instance| {

python/monarch/_src/actor/logging.py

@@ -50,8 +50,7 @@ def flush_all_proc_mesh_logs(v1: bool = False) -> None:
         from monarch._src.actor.v1.proc_mesh import get_active_proc_meshes
 
     for pm in get_active_proc_meshes():
-        if pm._logging_manager._logging_mesh_client is not None:
-            pm._logging_manager.flush()
+        pm._logging_manager.flush()
 
 
 class LoggingManager:

python/monarch/_src/actor/v1/proc_mesh.py

@@ -200,7 +200,8 @@ async def task(
         ) -> HyProcMesh:
             hy_proc_mesh = await hy_proc_mesh_task
 
-            await pm._logging_manager.init(hy_proc_mesh, stream_log_to_client)
+            # FIXME: Fix log forwarding.
+            # await pm._logging_manager.init(hy_proc_mesh, stream_log_to_client)
 
             if setup_actor is not None:
                 await setup_actor.setup.call()