(6/n) delete usage of state actor (#521)

Summary:
Pull Request resolved: #521

we are streaming log directly back now

Reviewed By: shayne-fletcher

Differential Revision: D78199598

fbshipit-source-id: 6a993b7a5c2317e3ffd6c2d3a014a80ab6cdbe48

Author: James Sun

hyperactor_mesh/src/alloc.rs

@@ -36,7 +36,6 @@ use serde::Deserialize;
 use serde::Serialize;
 
 use crate::alloc::test_utils::MockAllocWrapper;
-use crate::log_source::LogSource;
 use crate::proc_mesh::mesh_agent::MeshAgent;
 
 /// Errors that occur during allocation operations.
@@ -220,12 +219,6 @@ pub trait Alloc {
     /// The channel transport used the procs in this alloc.
     fn transport(&self) -> ChannelTransport;
 
-    /// The log source for this alloc.
-    /// It allows remote processes to stream stdout and stderr back to the client.
-    /// A client can connect to the log source to obtain the streamed logs.
-    /// A log source is allocation specific. Each allocator can decide how to stream the logs back.
-    async fn log_source(&self) -> Result<LogSource, AllocatorError>;
-
     /// Stop this alloc, shutting down all of its procs. A clean
     /// shutdown should result in Stop events from all allocs,
     /// followed by the end of the event stream.
@@ -360,10 +353,6 @@ pub mod test_utils {
             self.alloc.transport()
         }
 
-        async fn log_source(&self) -> Result<LogSource, AllocatorError> {
-            self.alloc.log_source().await
-        }
-
         async fn stop(&mut self) -> Result<(), AllocatorError> {
             self.alloc.stop().await
         }

hyperactor_mesh/src/alloc/local.rs

@@ -33,7 +33,6 @@ use crate::alloc::AllocSpec;
 use crate::alloc::Allocator;
 use crate::alloc::AllocatorError;
 use crate::alloc::ProcState;
-use crate::log_source::LogSource;
 use crate::proc_mesh::mesh_agent::MeshAgent;
 use crate::shortuuid::ShortUuid;
 
@@ -259,14 +258,6 @@ impl Alloc for LocalAlloc {
         self.transport.clone()
     }
 
-    async fn log_source(&self) -> Result<LogSource, AllocatorError> {
-        // Local alloc does not need to stream logs back.
-        // The client can subscribe to it but local actors will not stream logs into it.
-        LogSource::new_with_local_actor()
-            .await
-            .map_err(AllocatorError::from)
-    }
-
     async fn stop(&mut self) -> Result<(), AllocatorError> {
         for rank in 0..self.size() {
             self.todo_tx

hyperactor_mesh/src/alloc/process.rs

@@ -50,7 +50,6 @@ use crate::bootstrap;
 use crate::bootstrap::Allocator2Process;
 use crate::bootstrap::Process2Allocator;
 use crate::bootstrap::Process2AllocatorMessage;
-use crate::log_source::LogSource;
 use crate::shortuuid::ShortUuid;
 
 /// The maximum number of log lines to tail keep for managed processes.
@@ -87,9 +86,6 @@ impl Allocator for ProcessAllocator {
         let (bootstrap_addr, rx) = channel::serve(ChannelAddr::any(ChannelTransport::Unix))
             .await
             .map_err(anyhow::Error::from)?;
-        let log_source = LogSource::new_with_local_actor()
-            .await
-            .map_err(AllocatorError::from)?;
 
         let name = ShortUuid::generate();
         let n = spec.shape.slice().len();
@@ -98,7 +94,6 @@ impl Allocator for ProcessAllocator {
             world_id: WorldId(name.to_string()),
             spec: spec.clone(),
             bootstrap_addr,
-            log_source,
             rx,
             index: 0,
             active: HashMap::new(),
@@ -117,7 +112,6 @@ pub struct ProcessAlloc {
     world_id: WorldId, // to provide storage
     spec: AllocSpec,
     bootstrap_addr: ChannelAddr,
-    log_source: LogSource,
     rx: channel::ChannelRx<Process2Allocator>,
     index: usize,
     active: HashMap<usize, Child>,
@@ -492,10 +486,6 @@ impl Alloc for ProcessAlloc {
         ChannelTransport::Unix
     }
 
-    async fn log_source(&self) -> Result<LogSource, AllocatorError> {
-        Ok(self.log_source.clone())
-    }
-
     async fn stop(&mut self) -> Result<(), AllocatorError> {
         // We rely on the teardown here, and that the process should
         // exit on its own. We should have a hard timeout here as well,