Remove a superfluous copy during write serialization (#1087)

Right now, we copy data from a `BlockOp::Write/WriteUnwritten` into a `BytesMut`
buffer, encrypt it in-place, then serialize it using `CrucibleEncoder`.
Serialization requires a *second* full copy of the data.  This is noticeable in
our flamegraphs for large writes; see the PR for images.

This change removes that `memcpy`, speeding up large writes by a noticeable
amount; when the PR was first opened, I saw a ~30% speedup for 1M and 4M random
writes.

Author: mkeeter

Cargo.lock

@@ -90,6 +90,7 @@ slog-term = { version = "2.9" }
 static_assertions = "1.1.0"
 statistical = "1.0.0"
 subprocess = "0.2.9"
+strum_macros = "0.25.3"
 tempfile = "3"
 test-strategy = "0.3.1"
 thiserror = "1"

Cargo.toml

@@ -53,6 +53,75 @@ fn deadline_secs(secs: u64) -> Instant {
         .unwrap()
 }
 
+#[allow(clippy::derive_partial_eq_without_eq)]
+#[derive(Debug, Clone, PartialEq)]
+enum IOop {
+    Write {
+        dependencies: Vec<JobId>, // Jobs that must finish before this
+        writes: Vec<crucible_protocol::Write>,
+    },
+    WriteUnwritten {
+        dependencies: Vec<JobId>, // Jobs that must finish before this
+        writes: Vec<crucible_protocol::Write>,
+    },
+    Read {
+        dependencies: Vec<JobId>, // Jobs that must finish before this
+        requests: Vec<ReadRequest>,
+    },
+    Flush {
+        dependencies: Vec<JobId>, // Jobs that must finish before this
+        flush_number: u64,
+        gen_number: u64,
+        snapshot_details: Option<SnapshotDetails>,
+        extent_limit: Option<usize>,
+    },
+    /*
+     * These operations are for repairing a bad downstairs
+     */
+    ExtentClose {
+        dependencies: Vec<JobId>, // Jobs that must finish before this
+        extent: usize,
+    },
+    ExtentFlushClose {
+        dependencies: Vec<JobId>, // Jobs that must finish before this
+        extent: usize,
+        flush_number: u64,
+        gen_number: u64,
+        source_downstairs: ClientId,
+        repair_downstairs: Vec<ClientId>,
+    },
+    ExtentLiveRepair {
+        dependencies: Vec<JobId>, // Jobs that must finish before this
+        extent: usize,
+        source_downstairs: ClientId,
+        source_repair_address: SocketAddr,
+        repair_downstairs: Vec<ClientId>,
+    },
+    ExtentLiveReopen {
+        dependencies: Vec<JobId>, // Jobs that must finish before this
+        extent: usize,
+    },
+    ExtentLiveNoOp {
+        dependencies: Vec<JobId>, // Jobs that must finish before this
+    },
+}
+
+impl IOop {
+    fn deps(&self) -> &[JobId] {
+        match &self {
+            IOop::Write { dependencies, .. }
+            | IOop::Flush { dependencies, .. }
+            | IOop::Read { dependencies, .. }
+            | IOop::WriteUnwritten { dependencies, .. }
+            | IOop::ExtentClose { dependencies, .. }
+            | IOop::ExtentFlushClose { dependencies, .. }
+            | IOop::ExtentLiveRepair { dependencies, .. }
+            | IOop::ExtentLiveReopen { dependencies, .. }
+            | IOop::ExtentLiveNoOp { dependencies } => dependencies,
+        }
+    }
+}
+
 /*
  * Export the contents or partial contents of a Downstairs Region to
  * the file indicated.

downstairs/src/lib.rs

@@ -13,6 +13,7 @@ crucible-common.workspace = true
 num_enum.workspace = true
 schemars.workspace = true
 serde.workspace = true
+strum_macros.workspace = true
 tokio-util.workspace = true
 uuid.workspace = true
 crucible-workspace-hack.workspace = true

protocol/Cargo.toml

@@ -6,6 +6,7 @@ use anyhow::bail;
 use bytes::{Buf, BufMut, BytesMut};
 use num_enum::IntoPrimitive;
 use serde::{Deserialize, Serialize};
+use strum_macros::EnumDiscriminants;
 use tokio_util::codec::{Decoder, Encoder};
 use uuid::Uuid;
 
@@ -282,7 +283,10 @@ pub const CRUCIBLE_MESSAGE_VERSION: u32 = 5;
  * go do that right now before you forget.
  */
 #[allow(clippy::derive_partial_eq_without_eq)]
-#[derive(Debug, PartialEq, Clone, Serialize, Deserialize)]
+#[derive(
+    Debug, PartialEq, Clone, Serialize, Deserialize, EnumDiscriminants,
+)]
+#[strum_discriminants(derive(Serialize))]
 #[repr(u16)]
 pub enum Message {
     /**
@@ -531,6 +535,8 @@ pub enum Message {
     /*
      * IO related
      */
+    // Message::Write must contain the same fields in the same order as
+    // RawMessage::Write which is used for zero-copy serialization.
     Write {
         upstairs_id: Uuid,
         session_id: Uuid,
@@ -580,6 +586,8 @@ pub enum Message {
         responses: Result<Vec<ReadResponse>, CrucibleError>,
     },
 
+    // Message::WriteUnwritten must contain the same fields in the same order as
+    // RawMessage::WriteUnwritten, which is used for zero-copy serialization.
     WriteUnwritten {
         upstairs_id: Uuid,
         session_id: Uuid,

protocol/src/lib.rs

@@ -17,6 +17,7 @@ anyhow.workspace = true
 async-trait.workspace = true
 async-recursion.workspace = true
 base64.workspace = true
+bincode.workspace = true
 bytes.workspace = true
 chrono.workspace = true
 crucible-common.workspace = true