Remove unused flag CHANNEL_MULTIPART (#1994)

Summary:

`CHANNEL_MULTIPART` has been set to true for a long time. It is okay to remove it.

Reviewed By: dulinriley, thomasywang

Differential Revision: D87839991

Author: pzhan9

hyperactor/src/channel/net.rs

@@ -64,8 +64,6 @@ use super::*;
 use crate::RemoteMessage;
 use crate::clock::Clock;
 use crate::clock::RealClock;
-use crate::config;
-use crate::config::CHANNEL_MULTIPART;
 
 mod client;
 mod framed;
@@ -120,18 +118,6 @@ fn deserialize_response(data: Bytes) -> Result<NetRxResponse, bincode::Error> {
     bincode::deserialize(&data)
 }
 
-/// Serializes using the "illegal" multipart encoding whenever multipart
-/// is not enabled.
-fn serialize_bincode<S: ?Sized + serde::Serialize>(
-    value: &S,
-) -> Result<serde_multipart::Message, bincode::Error> {
-    if config::global::get(CHANNEL_MULTIPART) {
-        serde_multipart::serialize_bincode(value)
-    } else {
-        serde_multipart::serialize_illegal_bincode(value)
-    }
-}
-
 /// A Tx implemented on top of a Link. The Tx manages the link state,
 /// reconnections, etc.
 #[derive(Debug)]
@@ -914,6 +900,7 @@ mod tests {
     use crate::channel::net::framed::FrameWrite;
     use crate::channel::net::server::ServerConn;
     use crate::channel::net::server::SessionManager;
+    use crate::config;
     use crate::metrics;
     use crate::sync::mvar::MVar;
 

hyperactor/src/channel/net/client.rs

@@ -43,7 +43,6 @@ use crate::channel::net::NetRxResponse;
 use crate::channel::net::NetTx;
 use crate::channel::net::Stream;
 use crate::channel::net::deserialize_response;
-use crate::channel::net::serialize_bincode;
 use crate::clock::Clock;
 use crate::clock::RealClock;
 use crate::config;
@@ -265,7 +264,8 @@ impl<'a, M: RemoteMessage> Outbox<'a, M> {
         );
 
         let frame = Frame::Message(self.next_seq, message);
-        let message = serialize_bincode(&frame).map_err(|e| format!("serialization error: {e}"))?;
+        let message = serde_multipart::serialize_bincode(&frame)
+            .map_err(|e| format!("serialization error: {e}"))?;
         let message_size = message.frame_len();
         metrics::REMOTE_MESSAGE_SEND_SIZE.record(message_size as f64, &[]);
 
@@ -1128,7 +1128,9 @@ where
             } else {
                 match link.connect().await {
                     Ok(stream) => {
-                        let message = serialize_bincode(&Frame::<M>::Init(session_id)).unwrap();
+                        let message =
+                            serde_multipart::serialize_bincode(&Frame::<M>::Init(session_id))
+                                .unwrap();
 
                         let mut write = FrameWrite::new(
                             stream,

hyperactor/src/config.rs

@@ -179,13 +179,6 @@ declare_attrs! {
     })
     pub attr DEFAULT_ENCODING: Encoding = Encoding::Multipart;
 
-    /// Whether to use multipart encoding for network channel communications.
-    @meta(CONFIG = ConfigAttr {
-        env_name: Some("HYPERACTOR_CHANNEL_MULTIPART".to_string()),
-        py_name: None,
-    })
-    pub attr CHANNEL_MULTIPART: bool = true;
-
     /// How often to check for full MPSC channel on NetRx.
     @meta(CONFIG = ConfigAttr {
         env_name: Some("HYPERACTOR_CHANNEL_NET_RX_BUFFER_FULL_CHECK_INTERVAL".to_string()),
@@ -344,7 +337,6 @@ mod tests {
         let expected_lines: HashSet<&str> = indoc! {"
             # export HYPERACTOR_MESSAGE_LATENCY_SAMPLING_RATE=0.01
             # export HYPERACTOR_CHANNEL_NET_RX_BUFFER_FULL_CHECK_INTERVAL=5s
-            # export HYPERACTOR_CHANNEL_MULTIPART=1
             # export HYPERACTOR_DEFAULT_ENCODING=serde_multipart
             # export HYPERACTOR_REMOTE_ALLOCATOR_HEARTBEAT_INTERVAL=5m
             # export HYPERACTOR_STOP_ACTOR_TIMEOUT=10s

hyperactor/src/config/global.rs

@@ -1107,12 +1107,12 @@ mod tests {
         reset_to_defaults();
 
         let mut env = Attrs::new();
-        env[CHANNEL_MULTIPART] = false;
+        env[MESSAGE_DELIVERY_TIMEOUT] = Duration::from_secs(120);
         set(Source::Env, env);
 
-        assert!(!get(CHANNEL_MULTIPART));
-        let v = get_cloned(CHANNEL_MULTIPART);
-        assert!(!v);
+        assert_eq!(get(MESSAGE_DELIVERY_TIMEOUT), Duration::from_secs(120));
+        let v = get_cloned(MESSAGE_DELIVERY_TIMEOUT);
+        assert_eq!(v, Duration::from_secs(120));
     }
 
     #[test]

hyperactor/src/data.rs

@@ -398,12 +398,7 @@ impl std::fmt::Debug for Encoded {
             Encoded::Bincode(data) => write!(f, "Encoded::Bincode({})", HexFmt(data)),
             Encoded::Json(data) => write!(f, "Encoded::Json({})", HexFmt(data)),
             Encoded::Multipart(message) => {
-                write!(
-                    f,
-                    "Encoded::Multipart(illegal?={} body={}",
-                    message.is_illegal(),
-                    HexFmt(message.body())
-                )?;
+                write!(f, "Encoded::Multipart(body={}", HexFmt(message.body()))?;
                 for (index, part) in message.parts().iter().enumerate() {
                     write!(f, ", part[{}]={}", index, HexFmt(part))?;
                 }

hyperactor_mesh/src/actor_mesh.rs

@@ -1526,7 +1526,7 @@ mod tests {
                 hyperactor::mailbox::headers::set_rust_message_type::<Payload>(&mut headers);
                 let envelope = MessageEnvelope::new(src.clone(), dst.clone(), serialized, headers);
                 let frame = Frame::Message(0u64, envelope);
-                let message = serde_multipart::serialize_illegal_bincode(&frame).unwrap();
+                let message = serde_multipart::serialize_bincode(&frame).unwrap();
                 message.frame_len()
             }
 
@@ -1542,7 +1542,6 @@ mod tests {
             };
             let _guard3 =
                 config.override_key(hyperactor::config::DEFAULT_ENCODING, Encoding::Bincode);
-            let _guard4 = config.override_key(hyperactor::config::CHANNEL_MULTIPART, false);
 
             let alloc = process_allocator()
                 .allocate(AllocSpec {

serde_multipart/src/lib.rs

@@ -57,17 +57,12 @@ use serde::Serialize;
 pub struct Message {
     body: Part,
     parts: Vec<Part>,
-    is_illegal: bool,
 }
 
 impl Message {
     /// Returns a new message with the given body and parts.
     pub fn from_body_and_parts(body: Part, parts: Vec<Part>) -> Self {
-        Self {
-            body,
-            parts,
-            is_illegal: false,
-        }
+        Self { body, parts }
     }
 
     /// The body of the message.
@@ -118,17 +113,8 @@ impl Message {
     ///                                                                                      each part
     /// ```
     pub fn framed(self) -> Frame {
-        let is_illegal = self.is_illegal;
         let (body, parts) = self.into_inner();
 
-        if is_illegal {
-            assert!(parts.is_empty(), "illegal illegal message");
-            return Frame::from_buffers(vec![
-                Bytes::from_owner(u64::MAX.to_be_bytes()),
-                body.into_inner(),
-            ]);
-        }
-
         let mut buffers = Vec::with_capacity(2 + 2 * parts.len());
 
         let body = body.into_inner();
@@ -150,14 +136,6 @@ impl Message {
             return Err(std::io::ErrorKind::UnexpectedEof.into());
         }
         let body_len = buf.get_u64();
-        if body_len == u64::MAX {
-            return Ok(Self {
-                body: buf.into(),
-                parts: vec![],
-                is_illegal: true,
-            });
-        }
-
         let body = buf.split_to(body_len as usize);
         let mut parts = Vec::new();
         while !buf.is_empty() {
@@ -166,7 +144,6 @@ impl Message {
         Ok(Self {
             body: body.into(),
             parts,
-            is_illegal: false,
         })
     }
 
@@ -180,10 +157,6 @@ impl Message {
         }
         Ok(buf.split_to(at))
     }
-
-    pub fn is_illegal(&self) -> bool {
-        self.is_illegal
-    }
 }
 
 /// An encoded [`Message`] frame. Implements [`bytes::Buf`],
@@ -325,7 +298,6 @@ pub fn serialize_bincode<S: ?Sized + serde::Serialize>(
     Ok(Message {
         body: Part(buffer.into_inner().freeze()),
         parts: serializer.into_parts(),
-        is_illegal: false,
     })
 }
 
@@ -335,14 +307,6 @@ pub fn deserialize_bincode<T>(message: Message) -> Result<T, bincode::Error>
 where
     T: serde::de::DeserializeOwned,
 {
-    if message.is_illegal {
-        let (body, parts) = message.into_inner();
-        if !parts.is_empty() {
-            return Err(bincode::ErrorKind::Custom("illegal illegal message".to_string()).into());
-        }
-        return bincode::deserialize_from(body.into_inner().reader());
-    }
-
     let (body, parts) = message.into_inner();
     let mut deserializer = part::BincodeDeserializer::new(
         bincode::Deserializer::with_reader(body.into_inner().reader(), options()),
@@ -354,22 +318,6 @@ where
     Ok(value)
 }
 
-/// Serializes the provided value into an "illegal" Message that acts as an ordinary
-/// bincode-encoded buffer (bypassing multipart extraction and encoding). This is
-/// only used to support hyperactor's transition to multipart encoding, and will be
-/// removed after it is complete.
-///
-/// **YOU SHOULD NOT USE THIS**
-pub fn serialize_illegal_bincode<S: ?Sized + serde::Serialize>(
-    value: &S,
-) -> Result<Message, bincode::Error> {
-    Ok(Message {
-        body: Part::from(bincode::serialize(value)?),
-        parts: vec![],
-        is_illegal: true,
-    })
-}
-
 /// Construct the set of options used by the specialized serializer and deserializer.
 fn options() -> part::BincodeOptionsType {
     bincode::DefaultOptions::new()
@@ -412,16 +360,6 @@ mod tests {
         let bincode_serialized = bincode::serialize(&value).unwrap();
         let bincode_deserialized = bincode::deserialize(&bincode_serialized).unwrap();
         assert_eq!(value, bincode_deserialized);
-
-        // Illegal encoding:
-        let bincode_illegal = serialize_illegal_bincode(&value).unwrap();
-        let mut bincode_illegal_framed = bincode_illegal.clone().framed();
-        let bincode_illegal_framed =
-            bincode_illegal_framed.copy_to_bytes(bincode_illegal_framed.remaining());
-        let bincode_illegal_unframed = Message::from_framed(bincode_illegal_framed).unwrap();
-        let bincode_illegal_deserialized_value =
-            deserialize_bincode(bincode_illegal_unframed.clone()).unwrap();
-        assert_eq!(value, bincode_illegal_deserialized_value);
     }
 
     #[test]
@@ -507,7 +445,6 @@ mod tests {
         let message = Message {
             body: Part::from("hello"),
             parts: vec![Part::from("world")],
-            is_illegal: false,
         };
         let err = deserialize_bincode::<String>(message).unwrap_err();
 
@@ -565,7 +502,6 @@ mod tests {
                 Part::from("xyz"),
                 Part::from("xyzd"),
             ],
-            is_illegal: false,
         };
 
         let mut framed = message.clone().framed();