multipart channels (#948)

Summary:
Pull Request resolved: #948

This change introduces multipart support to net channels in Hyperactor. It is currently configurable (disabled by default) until we fully roll it out.

In addition to adding configurability, through a new key, CHANNEL_MULTIPART, this change makes two changes:

First, the framer now write Buf-typed frames (requiring the framer to become generic in the type of buffer). By abstracting over the buffer type, the underlying buffer is allowed to be noncontiguous; the framer performs vectored writes. Accordingly, the multi-part message frame implements Buf, presenting each part as a chunk, which will be included in the vectored write from the net framer.

Second, in order to reduce the complexity/divergence of this frame, we implement the standard bincode encoding by adding an "illegal" message. These are not valid multipart messages, but rather are a way for us to smuggle a (single) encoded blob into
serde_multipart::Message (and serde_multipart::Frame), which allows us to support both encodings using the same data structures, simplifying the net channel implementation.

Once we are confident in the multipart implementation, we will remove this illegal encoding, and also the configuration option to enable multipart encoding.

This change will have follow-ups to:

1) Clean up naming: we now have several things called "frames" and "messages", confusingly overlapping concepts; and
2) Homogenize client-to-server and server-to-client messages: currently, client-to-server messaging uses a framed multipart encoding, while server-to-client messages (today these are used only for acks) have a different encoding. We should keep everything consistent except for the message types.

## Benchmarking

Multipart encoding substantially improves the performance of the ping-pong benchmark, improving throughput by about **2.7x**.

```
# Baseline:
$ buck run mode/opt :hyperactor-example-channel
Average latency: 2.38006 ms
Min latency: 1.854 ms
Max latency: 2.789 ms
Total iterations: 100
Total time: 0 seconds
Bytes sent: 100000000 bytes
Bytes received: 100000000 bytes
Total bytes transferred: 200000000 bytes
Bandwidth: 840336134.4537816 bytes/sec (6411.255908613446 Mbps)

# Multipart:
$ HYPERACTOR_DEFAULT_ENCODING=multipart HYPERACTOR_CHANNEL_MULTIPART=true buck run mode/opt :hyperactor-example-channel
Average latency: 0.86634 ms
Min latency: 0.826 ms
Max latency: 0.935 ms
Total iterations: 100
Total time: 0 seconds
Bytes sent: 100000000 bytes
Bytes received: 100000000 bytes
Total bytes transferred: 200000000 bytes
Bandwidth: 2325581395.3488374 bytes/sec (17742.777979651164 Mbps)
```
ghstack-source-id: 305028344

Reviewed By: shayne-fletcher, vidhyav

Differential Revision: D80704703

fbshipit-source-id: 0722d5836ef9cbc08802bd1f3da5d2f8b4e14915

Author: mariusae

hyperactor/benches/main.rs

@@ -34,6 +34,7 @@ use hyperactor::reference::ProcId;
 use hyperactor::reference::WorldId;
 use serde::Deserialize;
 use serde::Serialize;
+use serde_multipart::Part;
 use tokio::runtime;
 use tokio::runtime::Runtime;
 use tokio::select;
@@ -208,12 +209,17 @@ async fn channel_ping_pong(
     message_size: usize,
     num_iter: usize,
 ) -> Duration {
-    let (client_addr, mut client_rx) = channel::serve::<Bytes>(ChannelAddr::any(transport.clone()))
-        .await
-        .unwrap();
-    let (server_addr, mut server_rx) = channel::serve::<Bytes>(ChannelAddr::any(transport.clone()))
-        .await
-        .unwrap();
+    #[derive(Clone, Debug, Named, Serialize, Deserialize)]
+    struct Message(Part);
+
+    let (client_addr, mut client_rx) =
+        channel::serve::<Message>(ChannelAddr::any(transport.clone()))
+            .await
+            .unwrap();
+    let (server_addr, mut server_rx) =
+        channel::serve::<Message>(ChannelAddr::any(transport.clone()))
+            .await
+            .unwrap();
 
     let _server_handle: tokio::task::JoinHandle<Result<(), anyhow::Error>> =
         tokio::spawn(async move {
@@ -227,7 +233,7 @@ async fn channel_ping_pong(
     let client_handle: tokio::task::JoinHandle<Result<(), anyhow::Error>> =
         tokio::spawn(async move {
             let server_tx = channel::dial(server_addr)?;
-            let message = Bytes::from(vec![0u8; message_size]);
+            let message = Message(Part::from(vec![0u8; message_size]));
             for _ in 0..num_iter {
                 server_tx.post(message.clone() /*cheap */);
                 client_rx.recv().await?;

hyperactor/example/channel.rs

@@ -7,17 +7,23 @@
  */
 
 use bytes::Bytes;
+use hyperactor::Named;
 use hyperactor::channel;
 use hyperactor::channel::ChannelAddr;
 use hyperactor::channel::ChannelRx;
 use hyperactor::channel::ChannelTransport;
 use hyperactor::channel::Rx;
 use hyperactor::channel::Tx;
+use serde::Deserialize;
+use serde::Serialize;
 use tokio::time::Duration;
 use tokio::time::Instant;
 
+#[derive(Clone, Debug, Named, Serialize, Deserialize)]
+struct Message(serde_multipart::Part);
+
 async fn server(
-    mut server_rx: ChannelRx<Bytes>,
+    mut server_rx: ChannelRx<Message>,
     client_addr: ChannelAddr,
 ) -> Result<(), anyhow::Error> {
     let client_tx = channel::dial(client_addr)?;
@@ -35,19 +41,21 @@ async fn main() -> Result<(), anyhow::Error> {
     let transport = ChannelTransport::Tcp;
     // let transport = ChannelTransport::Local;
     let message_size = 1_000_000;
-    let num_iter = 100;
+    let num_iter = 1000;
 
-    let (client_addr, mut client_rx) = channel::serve::<Bytes>(ChannelAddr::any(transport.clone()))
-        .await
-        .unwrap();
-    let (server_addr, server_rx) = channel::serve::<Bytes>(ChannelAddr::any(transport.clone()))
+    let (client_addr, mut client_rx) =
+        channel::serve::<Message>(ChannelAddr::any(transport.clone()))
+            .await
+            .unwrap();
+    let (server_addr, server_rx) = channel::serve::<Message>(ChannelAddr::any(transport.clone()))
         .await
         .unwrap();
 
     let _server_handle = tokio::spawn(server(server_rx, client_addr));
 
     let server_tx = channel::dial(server_addr)?;
-    let message = Bytes::from(vec![0u8; message_size]);
+
+    let message = Message(serde_multipart::Part::from(vec![0u8; message_size]));
 
     for _ in 0..10 {
         // Warmup
@@ -62,10 +70,10 @@ async fn main() -> Result<(), anyhow::Error> {
 
     let start = Instant::now();
     for _ in 0..num_iter {
-        total_bytes_sent += message.len();
+        total_bytes_sent += message.0.len();
         let start = Instant::now();
         server_tx.post(message.clone() /*cheap */);
-        total_bytes_received += client_rx.recv().await?.len();
+        total_bytes_received += client_rx.recv().await?.0.len();
         latencies.push(start.elapsed());
     }
     let elapsed = start.elapsed();