Use MessageWriter downstairs (#1286)

While staring at flamegraphs, I noticed a `memcpy` in the flamegraph.

The whole point of #1206 was to reorganize messages so that bulk data
could be sent down the wire in-place, without an intermediate copy.
However, it turns out that I forgot to actually use the new
`MessageWriter` in the Crucible Downstairs!

These changes switch to using the `MessageWriter` downstairs,
removing the suspicious `memcpy`.

It's hard to tell if there's a performance impact, but this still seems
worthwhile – if only to remove another big chunk of the flamegraph.

Author: mkeeter

downstairs/src/lib.rs

@@ -19,20 +19,20 @@ use crucible_common::{
     verbose_timeout, Block, CrucibleError, RegionDefinition, MAX_BLOCK_SIZE,
 };
 use crucible_protocol::{
-    BlockContext, CrucibleDecoder, CrucibleEncoder, JobId, Message,
-    ReadRequest, ReconciliationId, SnapshotDetails, CRUCIBLE_MESSAGE_VERSION,
+    BlockContext, CrucibleDecoder, JobId, Message, MessageWriter, ReadRequest,
+    ReconciliationId, SnapshotDetails, CRUCIBLE_MESSAGE_VERSION,
 };
 use repair_client::Client;
 
 use anyhow::{bail, Result};
 use bytes::BytesMut;
-use futures::{SinkExt, StreamExt};
+use futures::StreamExt;
 use rand::prelude::*;
 use slog::{debug, error, info, o, warn, Logger};
 use tokio::net::TcpListener;
 use tokio::sync::{mpsc, oneshot};
 use tokio::time::{sleep_until, Instant};
-use tokio_util::codec::{FramedRead, FramedWrite};
+use tokio_util::codec::FramedRead;
 use uuid::Uuid;
 
 pub mod admin;
@@ -594,15 +594,15 @@ async fn proc_stream(
             let (read, write) = sock.into_split();
 
             let fr = FramedRead::new(read, CrucibleDecoder::new());
-            let fw = FramedWrite::new(write, CrucibleEncoder::new());
+            let fw = MessageWriter::new(write);
 
             proc(ads, fr, fw).await
         }
         WrappedStream::Https(stream) => {
             let (read, write) = tokio::io::split(stream);
 
             let fr = FramedRead::new(read, CrucibleDecoder::new());
-            let fw = FramedWrite::new(write, CrucibleEncoder::new());
+            let fw = MessageWriter::new(write);
 
             proc(ads, fr, fw).await
         }
@@ -626,7 +626,7 @@ pub struct UpstairsConnection {
 async fn proc<RT, WT>(
     ads: &Arc<Mutex<Downstairs>>,
     mut fr: FramedRead<RT, CrucibleDecoder>,
-    mut fw: FramedWrite<WT, CrucibleEncoder>,
+    mut fw: MessageWriter<WT>,
 ) -> Result<()>
 where
     RT: tokio::io::AsyncRead + std::marker::Unpin + std::marker::Send,
@@ -1074,7 +1074,7 @@ where
 
 async fn reply_task<WT>(
     mut resp_channel_rx: mpsc::UnboundedReceiver<Message>,
-    mut fw: FramedWrite<WT, CrucibleEncoder>,
+    mut fw: MessageWriter<WT>,
 ) -> Result<()>
 where
     WT: tokio::io::AsyncWrite
@@ -1096,7 +1096,7 @@ where
 async fn resp_loop<RT, WT>(
     ads: &Arc<Mutex<Downstairs>>,
     mut fr: FramedRead<RT, CrucibleDecoder>,
-    fw: FramedWrite<WT, CrucibleEncoder>,
+    fw: MessageWriter<WT>,
     mut another_upstairs_active_rx: oneshot::Receiver<UpstairsConnection>,
     upstairs_connection: UpstairsConnection,
 ) -> Result<()>
@@ -1120,7 +1120,7 @@ where
      * Create tasks for:
      *  Doing the work then sending the ACK
      *  Pulling work off the socket and putting on the work queue.
-     *  Sending messages back on the FramedWrite
+     *  Sending messages back on the MessageWriter
      *
      * These tasks and this function must be able to handle the
      * Upstairs connection going away at any time, as well as a forced
@@ -6174,7 +6174,7 @@ mod test {
         let tcp = start_ds_and_connect(5555, 5556).await.unwrap();
         let (read, write) = tcp.into_split();
         let mut fr = FramedRead::new(read, CrucibleDecoder::new());
-        let mut fw = FramedWrite::new(write, CrucibleEncoder::new());
+        let mut fw = MessageWriter::new(write);
 
         // Our downstairs version is CRUCIBLE_MESSAGE_VERSION
         let m = Message::HereIAm {
@@ -6212,7 +6212,7 @@ mod test {
         let tcp = start_ds_and_connect(5557, 5558).await.unwrap();
         let (read, write) = tcp.into_split();
         let mut fr = FramedRead::new(read, CrucibleDecoder::new());
-        let mut fw = FramedWrite::new(write, CrucibleEncoder::new());
+        let mut fw = MessageWriter::new(write);
 
         // Our downstairs version is CRUCIBLE_MESSAGE_VERSION
         let m = Message::HereIAm {
@@ -6246,7 +6246,7 @@ mod test {
         let tcp = start_ds_and_connect(5579, 5560).await.unwrap();
         let (read, write) = tcp.into_split();
         let mut fr = FramedRead::new(read, CrucibleDecoder::new());
-        let mut fw = FramedWrite::new(write, CrucibleEncoder::new());
+        let mut fw = MessageWriter::new(write);
 
         // Our downstairs version is CRUCIBLE_MESSAGE_VERSION
         let m = Message::HereIAm {
@@ -6280,7 +6280,7 @@ mod test {
         let tcp = start_ds_and_connect(5561, 5562).await.unwrap();
         let (read, write) = tcp.into_split();
         let mut fr = FramedRead::new(read, CrucibleDecoder::new());
-        let mut fw = FramedWrite::new(write, CrucibleEncoder::new());
+        let mut fw = MessageWriter::new(write);
 
         // Our downstairs version is CRUCIBLE_MESSAGE_VERSION
         let m = Message::HereIAm {
@@ -6321,7 +6321,7 @@ mod test {
         let tcp = start_ds_and_connect(5563, 5564).await.unwrap();
         let (read, write) = tcp.into_split();
         let mut fr = FramedRead::new(read, CrucibleDecoder::new());
-        let mut fw = FramedWrite::new(write, CrucibleEncoder::new());
+        let mut fw = MessageWriter::new(write);
 
         // Our downstairs version is CRUCIBLE_MESSAGE_VERSION
         let m = Message::HereIAm {