use CheapCloneForRead trait to prevent efficiency bugs

Signed-off-by: Yuchen Liang <yuchen@neon.tech>

pageserver/src/virtual_file/owned_buffers_io/io_buf_ext.rs

@@ -5,6 +5,8 @@ use bytes::{Bytes, BytesMut};
 use std::ops::{Deref, Range};
 use tokio_epoll_uring::{BoundedBuf, IoBuf, Slice};
 
+use super::write::CheapCloneForRead;
+
 /// The true owned equivalent for Rust [`slice`]. Use this for the write path.
 ///
 /// Unlike [`tokio_epoll_uring::Slice`], which we unfortunately inherited from `tokio-uring`,
@@ -29,17 +31,6 @@ where
     }
 }
 
-impl<B> Clone for FullSlice<B>
-where
-    B: IoBuf + Clone,
-{
-    fn clone(&self) -> Self {
-        Self {
-            slice: self.slice.get_ref().clone().slice_full(),
-        }
-    }
-}
-
 impl<B> Deref for FullSlice<B>
 where
     B: IoBuf,
@@ -54,6 +45,17 @@ where
     }
 }
 
+impl<B> CheapCloneForRead for FullSlice<B>
+where
+    B: IoBuf + CheapCloneForRead,
+{
+    fn cheap_clone(&self) -> Self {
+        Self {
+            slice: self.slice.get_ref().cheap_clone().slice_full(),
+        }
+    }
+}
+
 pub(crate) trait IoBufExt {
     /// Get a [`FullSlice`] for the entire buffer, i.e., `self[..]` or `self[0..self.len()]`.
     fn slice_len(self) -> FullSlice<Self>

pageserver/src/virtual_file/owned_buffers_io/write.rs

@@ -17,6 +17,18 @@ use super::{
 
 pub(crate) use flush::FlushControl;
 
+pub(crate) trait CheapCloneForRead {
+    /// Returns a cheap clone of the buffer.
+    fn cheap_clone(&self) -> Self;
+}
+
+impl CheapCloneForRead for IoBuffer {
+    fn cheap_clone(&self) -> Self {
+        // Cheap clone over an `Arc`.
+        self.clone()
+    }
+}
+
 /// A trait for doing owned-buffer write IO.
 /// Think [`tokio::io::AsyncWrite`] but with owned buffers.
 /// The owned buffers need to be aligned due to Direct IO requirements.
@@ -61,7 +73,7 @@ pub struct BufferedWriter<B: Buffer, W> {
 impl<B, Buf, W> BufferedWriter<B, W>
 where
     B: Buffer<IoBuf = Buf> + Send + 'static,
-    Buf: IoBufAligned + Send + Sync + Clone,
+    Buf: IoBufAligned + Send + Sync + CheapCloneForRead,
     W: OwnedAsyncWriter + Send + Sync + 'static + std::fmt::Debug,
 {
     /// Creates a new buffered writer.

pageserver/src/virtual_file/owned_buffers_io/write/flush.rs

@@ -7,7 +7,7 @@ use crate::{
     virtual_file::owned_buffers_io::{io_buf_aligned::IoBufAligned, io_buf_ext::FullSlice},
 };
 
-use super::{Buffer, OwnedAsyncWriter};
+use super::{Buffer, CheapCloneForRead, OwnedAsyncWriter};
 
 /// A handle to the flush task.
 pub struct FlushHandle<Buf, W> {
@@ -110,7 +110,7 @@ impl FlushControl {
 
 impl<Buf, W> FlushHandle<Buf, W>
 where
-    Buf: IoBufAligned + Send + Sync + Clone,
+    Buf: IoBufAligned + Send + Sync + CheapCloneForRead,
     W: OwnedAsyncWriter + Send + Sync + 'static + std::fmt::Debug,
 {
     /// Spawns a new background flush task and obtains a handle.
@@ -148,7 +148,7 @@ where
         let slice = buf.flush();
 
         // Saves a buffer for read while flushing. This also removes reference to the old buffer.
-        self.maybe_flushed = Some(slice.clone());
+        self.maybe_flushed = Some(slice.cheap_clone());
 
         let (request, flush_control) = new_flush_op(slice, offset);