From 907054775dc71a10a92ab96112889db9113130ab Mon Sep 17 00:00:00 2001
From: Alex Auvolat <alex@adnab.me>
Date: Tue, 6 Sep 2022 22:25:23 +0200
Subject: Faster copy, better get error message

---
 src/api/s3/copy.rs | 12 +++++-------
 src/api/s3/get.rs  |  4 ++--
 2 files changed, 7 insertions(+), 9 deletions(-)

(limited to 'src/api/s3')

diff --git a/src/api/s3/copy.rs b/src/api/s3/copy.rs
index 10cf5935..a1a8c9a4 100644
--- a/src/api/s3/copy.rs
+++ b/src/api/s3/copy.rs
@@ -9,6 +9,7 @@ use bytes::Bytes;
 use hyper::{Body, Request, Response};
 use serde::Serialize;
 
+use garage_rpc::netapp::bytes_buf::BytesBuf;
 use garage_rpc::rpc_helper::OrderTag;
 use garage_table::*;
 use garage_util::data::*;
@@ -566,7 +567,7 @@ type BlockStreamItem = Result<BlockStreamItemOk, garage_util::error::Error>;
 struct Defragmenter<S: Stream<Item = BlockStreamItem>> {
 	block_size: usize,
 	block_stream: Pin<Box<stream::Peekable<S>>>,
-	buffer: Vec<u8>,
+	buffer: BytesBuf,
 	hash: Option<Hash>,
 }
 
@@ -575,7 +576,7 @@ impl<S: Stream<Item = BlockStreamItem>> Defragmenter<S> {
 		Self {
 			block_size,
 			block_stream,
-			buffer: vec![],
+			buffer: BytesBuf::new(),
 			hash: None,
 		}
 	}
@@ -593,7 +594,7 @@ impl<S: Stream<Item = BlockStreamItem>> Defragmenter<S> {
 
 			if self.buffer.is_empty() {
 				let (next_block, next_block_hash) = self.block_stream.next().await.unwrap()?;
-				self.buffer = next_block.to_vec(); // TODO TOO MUCH COPY
+				self.buffer.extend(next_block);
 				self.hash = next_block_hash;
 			} else if self.buffer.len() + peeked_next_block.len() > self.block_size {
 				break;
@@ -604,10 +605,7 @@ impl<S: Stream<Item = BlockStreamItem>> Defragmenter<S> {
 			}
 		}
 
-		Ok((
-			Bytes::from(std::mem::take(&mut self.buffer)),
-			self.hash.take(),
-		))
+		Ok((self.buffer.take_all(), self.hash.take()))
 	}
 }
 
diff --git a/src/api/s3/get.rs b/src/api/s3/get.rs
index dfc284fe..dd95f6e7 100644
--- a/src/api/s3/get.rs
+++ b/src/api/s3/get.rs
@@ -274,11 +274,11 @@ pub async fn handle_get(
 								.block_manager
 								.rpc_get_block_streaming(&hash, Some(order_stream.order(i as u64)))
 								.await
-								.unwrap_or_else(|_| {
+								.unwrap_or_else(|e| {
 									Box::pin(futures::stream::once(async move {
 										Err(std::io::Error::new(
 											std::io::ErrorKind::Other,
-											"Could not get next block",
+											format!("Could not get block {}: {}", i, e),
 										))
 									}))
 								})
-- 
cgit v1.2.3