fix pid in log_writer (#500)

Summary:
Pull Request resolved: #500

Previously process ID was derived wrong by using the process ID of the current process. That is wrong because the log_writer is spawned on the parent process.
This diff corrects it by attaching the Child process's pid to the log writer so this pid can be used directly in generating the Log Object Name.

Reviewed By: highker

Differential Revision: D78105348

fbshipit-source-id: 42fbbbaee04e03e7c9690d351dbe6450971bef59

Author: kaiyuan-li

hyperactor_mesh/src/alloc/process.rs

@@ -174,6 +174,7 @@ impl Child {
             match hyperactor_state::log_writer::create_log_writers(
                 state_actor_addr,
                 state_actor_ref,
+                process.id().unwrap_or(0),
             ) {
                 Ok((stdout_writer, stderr_writer)) => {
                     stdout_tee = stdout_writer;

hyperactor_state/src/log_writer.rs

@@ -83,20 +83,25 @@ pub struct LogWriterActor {
     // Sequence counters for stdout and stderr
     stdout_seq: AtomicU64,
     stderr_seq: AtomicU64,
+    // Process ID of the process that generates the logs.
+    pid: u32,
 }
 
 #[derive(Debug, Clone)]
 pub struct LogWriterActorParams {
     mailbox_router: DialMailboxRouter,
     client: Mailbox,
+    // Process ID of the process that generates the logs.
+    pid: u32,
 }
 
 impl LogWriterActorParams {
     // Constructor for LogWriterActorParams
-    pub fn new(mailbox_router: DialMailboxRouter, client: Mailbox) -> Self {
+    pub fn new(mailbox_router: DialMailboxRouter, client: Mailbox, pid: u32) -> Self {
         Self {
             mailbox_router,
             client,
+            pid,
         }
     }
 }
@@ -112,6 +117,7 @@ impl Actor for LogWriterActor {
             mailbox_router: params.mailbox_router,
             stdout_seq: AtomicU64::new(0),
             stderr_seq: AtomicU64::new(0),
+            pid: params.pid,
         })
     }
 }
@@ -156,7 +162,7 @@ impl LogWriterMessageHandler for LogWriterActor {
 
             // Create the log state object
             let metadata = StateMetadata {
-                name: Name::from(output_target),
+                name: Name::from_target_with_pid(output_target, self.pid),
                 kind: Kind::Log,
             };
             let spec = LogSpec {};
@@ -226,14 +232,15 @@ impl StateActorLogSender {
     pub fn new(
         state_actor_addr: ChannelAddr,
         state_actor_ref: ActorRef<StateActor>,
+        pid: u32,
     ) -> Result<Self, anyhow::Error> {
         // Create a LogWriterActor
         let proc_id = id!(log_writer_proc).random_user_proc();
         let router = DialMailboxRouter::new();
         let proc = Proc::new(proc_id, BoxedMailboxSender::new(router.clone()));
         // Create client mailbox
         let client = proc.attach("client")?;
-        let log_writer_params = LogWriterActorParams::new(router, client.clone());
+        let log_writer_params = LogWriterActorParams::new(router, client.clone(), pid);
 
         // Spawn the LogWriterActor using our helper function
         let log_writer_handle = futures::executor::block_on(Self::spawn_log_writer_actor(
@@ -289,13 +296,16 @@ pub struct LogWriter<T: LogSender + Unpin + 'static, S: io::AsyncWrite + Send +
 /// # Arguments
 ///
 /// * `state_actor_addr` - The address of the state actor to send logs to
+/// * `state_actor_ref` - A reference to the state actor
+/// * `pid` - The process ID of the process that generates the logs
 ///
 /// # Returns
 ///
 /// A tuple of boxed writers for stdout and stderr
 pub fn create_log_writers(
     state_actor_addr: ChannelAddr,
     state_actor_ref: ActorRef<StateActor>,
+    pid: u32,
 ) -> Result<
     (
         Box<dyn io::AsyncWrite + Send + Unpin + 'static>,
@@ -304,7 +314,7 @@ pub fn create_log_writers(
     anyhow::Error,
 > {
     // Create a single StateActorLogSender to be shared between stdout and stderr
-    let log_sender = StateActorLogSender::new(state_actor_addr, state_actor_ref)?;
+    let log_sender = StateActorLogSender::new(state_actor_addr, state_actor_ref, pid)?;
 
     // Create LogWriter instances for stdout and stderr using the shared log sender
     let stdout_writer = LogWriter::with_default_writer(OutputTarget::Stdout, log_sender.clone())?;

hyperactor_state/src/object.rs

@@ -21,13 +21,14 @@ pub enum Name {
     StderrLog((String, u32)),
 }
 
-impl From<OutputTarget> for Name {
-    fn from(target: OutputTarget) -> Self {
+impl Name {
+    /// Creates a new Name from an OutputTarget and a specific process ID.
+    /// This allows passing in a PID from outside instead of using the current process ID.
+    pub fn from_target_with_pid(target: OutputTarget, pid: u32) -> Self {
         let hostname = hostname::get()
             .unwrap_or_else(|_| "unknown_host".into())
             .into_string()
             .unwrap_or("unknown_host".to_string());
-        let pid = std::process::id();
 
         match target {
             OutputTarget::Stdout => Name::StdoutLog((hostname, pid)),