ignore sigpipe (#835)

Summary:
Pull Request resolved: #835

SAFETY: ignore SIGPIPE as hyperactor tx/rx channels have the following issue:
When tx tries to send a message, it can either do try_post/post or send.
try_post/post is async and send is sync. However, both methods do not have a
way to tell the receiver's state.
For try_post/post, it will send messages in the background. However, if the
receiver process gets killed,
try_post/post will continue sending messages leading to SIGPIPE and crash the
parent process.
For send, it is a sync call. It waits on the receiver to ack. However, if the
receiver is dead or not started yet,
the send will be blocked.

We need to fix the above issues but before that, ignore SIGPIPE as a
mitigation.

As a specific victim of this, the log sender streams the log to the log
forwarder. However, it doesn't know the state of the log forwarder. It could
have not started yet or being killed. If we use tx.send, it will be blocked
leading to stuck allocator. If we use tx.post, it will get back sigpipe on
child process shutdown.

Reviewed By: LucasLLC, vidhyav

Differential Revision: D80131609

fbshipit-source-id: 374165d1ceb835a3ea618ba0931b7464304022f2

Author: James Sun

hyperactor_mesh/src/logging.rs

@@ -289,7 +289,7 @@ pub trait LogSender: Send + Sync {
 
     /// Flush the log channel, ensuring all messages are delivered
     /// Returns when the flush message has been acknowledged
-    async fn flush(&mut self) -> anyhow::Result<()>;
+    fn flush(&mut self) -> anyhow::Result<()>;
 }
 
 /// Represents the target output stream (stdout or stderr)
@@ -331,7 +331,7 @@ impl LocalLogSender {
 impl LogSender for LocalLogSender {
     fn send(&mut self, target: OutputTarget, payload: Vec<u8>) -> anyhow::Result<()> {
         if TxStatus::Active == *self.status.borrow() {
-            // post does not guarantee the message to be delivered
+            // Do not use tx.send, it will block the allocator as the child process state is unknown.
             self.tx.post(LogMessage::Log {
                 hostname: self.hostname.clone(),
                 pid: self.pid,
@@ -348,23 +348,18 @@ impl LogSender for LocalLogSender {
         Ok(())
     }
 
-    async fn flush(&mut self) -> anyhow::Result<()> {
+    fn flush(&mut self) -> anyhow::Result<()> {
         // send will make sure message is delivered
         if TxStatus::Active == *self.status.borrow() {
-            match self.tx.send(LogMessage::Flush {}).await {
-                Ok(()) => Ok(()),
-                Err(e) => {
-                    tracing::error!("log sender {} error sending flush message: {}", self.pid, e);
-                    Err(anyhow::anyhow!("error sending flush message: {}", e))
-                }
-            }
+            // Do not use tx.send, it will block the allocator as the child process state is unknown.
+            self.tx.post(LogMessage::Flush {});
         } else {
             tracing::debug!(
                 "log sender {} is not active, skip sending flush message",
                 self.tx.addr()
             );
-            Ok(())
         }
+        Ok(())
     }
 }
 
@@ -523,30 +518,12 @@ impl<T: LogSender + Unpin + 'static, S: io::AsyncWrite + Send + Unpin + 'static>
     fn poll_flush(self: Pin<&mut Self>, cx: &mut TaskContext<'_>) -> Poll<Result<(), io::Error>> {
         let this = self.get_mut();
 
-        // First, flush the standard writer
         match Pin::new(&mut this.std_writer).poll_flush(cx) {
             Poll::Ready(Ok(())) => {
-                // Now send a Flush message to the other side of the channel.
-                let mut flush_future = this.log_sender.flush();
-                match flush_future.as_mut().poll(cx) {
-                    Poll::Ready(Ok(())) => {
-                        // Successfully sent the flush message
-                        Poll::Ready(Ok(()))
-                    }
-                    Poll::Ready(Err(e)) => {
-                        // Error sending the flush message
-                        tracing::error!("error sending flush message: {}", e);
-                        Poll::Ready(Err(io::Error::other(format!(
-                            "error sending flush message: {}",
-                            e
-                        ))))
-                    }
-                    Poll::Pending => {
-                        // The future is not ready yet, so we return Pending
-                        // The waker is already registered by polling the future
-                        Poll::Pending
-                    }
+                if let Err(e) = this.log_sender.flush() {
+                    tracing::error!("error sending flush: {}", e);
                 }
+                Poll::Ready(Ok(()))
             }
             other => other, // Propagate any errors or Pending state from the std_writer flush
         }
@@ -1044,7 +1021,7 @@ mod tests {
                 .map_err(|e| anyhow::anyhow!("Failed to send log in test: {}", e))
         }
 
-        async fn flush(&mut self) -> anyhow::Result<()> {
+        fn flush(&mut self) -> anyhow::Result<()> {
             // Mark that flush was called
             let mut flush_called = self.flush_called.lock().unwrap();
             *flush_called = true;

monarch_hyperactor/Cargo.toml

@@ -26,6 +26,7 @@ hyperactor_multiprocess = { version = "0.0.0", path = "../hyperactor_multiproces
 hyperactor_telemetry = { version = "0.0.0", path = "../hyperactor_telemetry" }
 inventory = "0.3.8"
 lazy_static = "1.5"
+libc = "0.2.139"
 monarch_types = { version = "0.0.0", path = "../monarch_types" }
 ndslice = { version = "0.0.0", path = "../ndslice" }
 nix = { version = "0.30.1", features = ["dir", "event", "hostname", "inotify", "ioctl", "mman", "mount", "net", "poll", "ptrace", "reboot", "resource", "sched", "signal", "term", "time", "user", "zerocopy"] }

monarch_hyperactor/src/bin/process_allocator/common.rs

@@ -52,6 +52,23 @@ pub fn main_impl(
     program: Command,
     timeout: Option<Duration>,
 ) -> tokio::task::JoinHandle<Result<(), anyhow::Error>> {
+    #[cfg(unix)]
+    fn ignore_sigpipe() {
+        // SAFETY: ignore SIGPIPE as hyperactor tx/rx channels have the following issue:
+        // When tx tries to send a message, it can either do try_post/post or send.
+        // try_post/post is async and send is sync. However, both methods do not have a way to tell the receiver's state.
+        // For try_post/post, it will send messages in the background. However, if the receiver process gets killed,
+        // try_post/post will continue sending messages leading to SIGPIPE and crash the parent process.
+        // For send, it is a sync call. It waits on the receiver to ack. However, if the receiver is dead or not started yet,
+        // the send will be blocked.
+        //
+        // TODO: We need to fix the above issues but before that, ignore SIGPIPE as a mitigation.
+        unsafe {
+            assert!(libc::signal(libc::SIGPIPE, libc::SIG_IGN) != libc::SIG_ERR);
+        }
+    }
+    ignore_sigpipe();
+
     tracing::info!("bind address is: {}", serve_address);
     tracing::info!("program to spawn on allocation request: [{:?}]", &program);