br-stream: don't clone the key & value in encode_event()

components/br-stream/src/endpoint.rs

@@ -1,5 +1,6 @@
 // Copyright 2021 TiKV Project Authors. Licensed under Apache-2.0.
 
+use std::convert::AsRef;
 use std::fmt;
 use std::path::PathBuf;
 
@@ -18,7 +19,7 @@ use raftstore::coprocessor::CmdBatch;
 use tikv::config::BackupStreamConfig;
 
 use tikv_util::worker::{Runnable, Scheduler};
-use tikv_util::{debug, error, info};
+use tikv_util::{debug, error, info, Either};
 
 use super::metrics::{HANDLE_EVENT_DURATION_HISTOGRAM, HANDLE_KV_HISTOGRAM};
 
@@ -137,17 +138,16 @@ where
             }
         }
     }
-    // TODO use a more efficent encode kv event format
     // TODO move this function to a indepentent module.
-    pub fn encode_event(key: &[u8], value: &[u8]) -> Vec<u8> {
-        let mut buf = vec![];
-        let key_len = (key.len() as u32).to_ne_bytes();
-        let val_len = value.len().to_ne_bytes();
-        buf.extend_from_slice(&key_len);
-        buf.extend_from_slice(key);
-        buf.extend_from_slice(&val_len);
-        buf.extend_from_slice(value);
-        buf
+    pub fn encode_event<'e>(key: &'e [u8], value: &'e [u8]) -> [impl AsRef<[u8]> + 'e; 4] {
+        let key_len = (key.len() as u32).to_le_bytes();
+        let val_len = (value.len() as u32).to_le_bytes();
+        [
+            Either::Left(key_len),
+            Either::Right(key),
+            Either::Left(val_len),
+            Either::Right(value),
+        ]
     }
 
     fn backup_batch(&self, batch: CmdBatch) {

components/br-stream/src/router.rs

@@ -539,16 +539,20 @@ impl DataFile {
     /// Add a new KV pair to the file, returning its size.
     async fn on_event(&mut self, mut kv: ApplyEvent) -> Result<usize> {
         let encoded = super::endpoint::Endpoint::<EtcdStore>::encode_event(&kv.key, &kv.value);
-        let size = encoded.len();
+        let mut size = 0;
+        for slice in encoded {
+            let slice = slice.as_ref();
+            self.inner.write_all(slice).await?;
+            self.sha256.update(slice).map_err(|err| {
+                Error::Other(box_err!("openssl hasher failed to update: {}", err))
+            })?;
+            size += slice.len();
+        }
         let key = Key::from_encoded(std::mem::take(&mut kv.key));
         let ts = key.decode_ts().expect("key without ts");
         self.min_ts = self.min_ts.min(ts);
         self.max_ts = self.max_ts.max(ts);
         self.resolved_ts = self.resolved_ts.max(kv.region_resolved_ts.into());
-        self.inner.write_all(encoded.as_slice()).await?;
-        self.sha256
-            .update(encoded.as_slice())
-            .map_err(|err| Error::Other(box_err!("openssl hasher failed to update: {}", err)))?;
         self.update_key_bound(key.into_encoded());
         Ok(size)
     }

components/tikv_util/src/lib.rs

@@ -9,6 +9,7 @@ extern crate test;
 
 use std::collections::hash_map::Entry;
 use std::collections::vec_deque::{Iter, VecDeque};
+use std::convert::AsRef;
 use std::fs::File;
 use std::ops::{Deref, DerefMut};
 use std::path::{Path, PathBuf};
@@ -329,6 +330,20 @@ impl<L, R> Either<L, R> {
     }
 }
 
+impl<L, R, T> AsRef<T> for Either<L, R>
+where
+    T: ?Sized,
+    L: AsRef<T>,
+    R: AsRef<T>,
+{
+    fn as_ref(&self) -> &T {
+        match self {
+            Self::Left(l) => l.as_ref(),
+            Self::Right(r) => r.as_ref(),
+        }
+    }
+}
+
 /// A simple ring queue with fixed capacity.
 pub struct RingQueue<T> {
     buf: VecDeque<T>,