gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-27 00:38:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #482 from rustfs/nugine/refactor/bytes-io

Browse Source 
refactor: Bytes IO
This commit is contained in:
loverustfs
2025-06-17 20:20:31 +08:00
committed by GitHub
parent e254ddc947 4a786618d4
commit 04cfe837a3
9 changed files with 101 additions and 127 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/rio/src/http_reader.rs
+28 -64
View File
@@ -1,15 +1,17 @@
use bytes::Bytes;
use futures::{Stream, StreamExt};
use futures::{Stream, TryStreamExt as _};
use http::HeaderMap;
use pin_project_lite::pin_project;
use reqwest::{Client, Method, RequestBuilder};
use std::error::Error as _;
use std::io::{self, Error};
use std::ops::Not as _;
use std::pin::Pin;
use std::sync::LazyLock;
use std::task::{Context, Poll};
use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt, DuplexStream, ReadBuf};
use tokio::sync::{mpsc, oneshot};
use tokio::io::{AsyncRead, AsyncWrite, ReadBuf};
use tokio::sync::mpsc;
use tokio_util::io::StreamReader;
use crate::{EtagResolvable, HashReaderDetector, HashReaderMut};
@@ -38,8 +40,7 @@ pin_project! {
url:String,
method: Method,
headers: HeaderMap,
inner: DuplexStream,
err_rx: oneshot::Receiver<std::io::Error>,
inner: StreamReader<Pin<Box<dyn Stream<Item=std::io::Result<Bytes>>+Send+Sync>>, Bytes>,
}
}
@@ -54,11 +55,11 @@ impl HttpReader {
method: Method,
headers: HeaderMap,
body: Option<Vec<u8>>,
mut read_buf_size: usize,
_read_buf_size: usize,
) -> io::Result<Self> {
http_log!(
"[HttpReader::with_capacity] url: {url}, method: {method:?}, headers: {headers:?}, buf_size: {}",
read_buf_size
_read_buf_size
);
// First, check if the connection is available (HEAD)
let client = get_http_client();
@@ -76,59 +77,30 @@ impl HttpReader {
}
}
let url_clone = url.clone();
let method_clone = method.clone();
let headers_clone = headers.clone();
if read_buf_size == 0 {
read_buf_size = 8192; // Default buffer size
let client = get_http_client();
let mut request: RequestBuilder = client.request(method.clone(), url.clone()).headers(headers.clone());
if let Some(body) = body {
request = request.body(body);
}
let (rd, mut wd) = tokio::io::duplex(read_buf_size);
let (err_tx, err_rx) = oneshot::channel::<io::Error>();
tokio::spawn(async move {
let client = get_http_client();
let mut request: RequestBuilder = client.request(method_clone, url_clone).headers(headers_clone);
if let Some(body) = body {
request = request.body(body);
}
let response = request.send().await;
match response {
Ok(resp) => {
if resp.status().is_success() {
let mut stream = resp.bytes_stream();
while let Some(chunk) = stream.next().await {
match chunk {
Ok(data) => {
if let Err(e) = wd.write_all(&data).await {
let _ = err_tx.send(Error::other(format!("HttpReader write error: {}", e)));
break;
}
}
Err(e) => {
let _ = err_tx.send(Error::other(format!("HttpReader stream error: {}", e)));
break;
}
}
}
} else {
http_log!("[HttpReader::spawn] HTTP request failed with status: {}", resp.status());
let _ = err_tx.send(Error::other(format!(
"HttpReader HTTP request failed with non-200 status {}",
resp.status()
)));
}
}
Err(e) => {
let _ = err_tx.send(Error::other(format!("HttpReader HTTP request error: {}", e)));
}
}
let resp = request
.send()
.await
.map_err(|e| Error::other(format!("HttpReader HTTP request error: {}", e)))?;
if resp.status().is_success().not() {
return Err(Error::other(format!(
"HttpReader HTTP request failed with non-200 status {}",
resp.status()
)));
}
let stream = resp
.bytes_stream()
.map_err(|e| Error::other(format!("HttpReader stream error: {}", e)));
http_log!("[HttpReader::spawn] HTTP request completed, exiting");
});
Ok(Self {
inner: rd,
err_rx,
inner: StreamReader::new(Box::pin(stream)),
url,
method,
headers,
@@ -153,14 +125,6 @@ impl AsyncRead for HttpReader {
self.method,
buf.remaining()
);
// Check for errors from the request
match Pin::new(&mut self.err_rx).try_recv() {
Ok(e) => return Poll::Ready(Err(e)),
Err(oneshot::error::TryRecvError::Empty) => {}
Err(oneshot::error::TryRecvError::Closed) => {
// return Poll::Ready(Err(Error::new(ErrorKind::Other, "HTTP request closed")));
}
}
// Read from the inner stream
Pin::new(&mut self.inner).poll_read(cx, buf)
}
ecstore/src/disk/local.rs
+24 -24
View File
@@ -68,7 +68,7 @@ use uuid::Uuid;
#[derive(Debug)]
pub struct FormatInfo {
pub id: Option<Uuid>,
pub data: Vec<u8>,
pub data: Bytes,
pub file_info: Option<Metadata>,
pub last_check: Option<OffsetDateTime>,
}
@@ -138,7 +138,7 @@ impl LocalDisk {
let mut format_last_check = None;
if !format_data.is_empty() {
let s = format_data.as_slice();
let s = format_data.as_ref();
let fm = FormatV3::try_from(s).map_err(Error::other)?;
let (set_idx, disk_idx) = fm.find_disk_index_by_disk_id(fm.erasure.this)?;
@@ -629,7 +629,7 @@ impl LocalDisk {
}
// write_all_public for trail
async fn write_all_public(&self, volume: &str, path: &str, data: Vec<u8>) -> Result<()> {
async fn write_all_public(&self, volume: &str, path: &str, data: Bytes) -> Result<()> {
if volume == RUSTFS_META_BUCKET && path == super::FORMAT_CONFIG_FILE {
let mut format_info = self.format_info.write().await;
format_info.data.clone_from(&data);
@@ -637,7 +637,7 @@ impl LocalDisk {
let volume_dir = self.get_bucket_path(volume)?;
self.write_all_private(volume, path, data.into(), true, &volume_dir).await?;
self.write_all_private(volume, path, data, true, &volume_dir).await?;
Ok(())
}
@@ -980,13 +980,13 @@ fn is_root_path(path: impl AsRef<Path>) -> bool {
}
// 过滤 std::io::ErrorKind::NotFound
pub async fn read_file_exists(path: impl AsRef<Path>) -> Result<(Vec<u8>, Option<Metadata>)> {
pub async fn read_file_exists(path: impl AsRef<Path>) -> Result<(Bytes, Option<Metadata>)> {
let p = path.as_ref();
let (data, meta) = match read_file_all(&p).await {
Ok((data, meta)) => (data, Some(meta)),
Err(e) => {
if e == Error::FileNotFound {
(Vec::new(), None)
(Bytes::new(), None)
} else {
return Err(e);
}
@@ -1001,13 +1001,13 @@ pub async fn read_file_exists(path: impl AsRef<Path>) -> Result<(Vec<u8>, Option
Ok((data, meta))
}
pub async fn read_file_all(path: impl AsRef<Path>) -> Result<(Vec<u8>, Metadata)> {
pub async fn read_file_all(path: impl AsRef<Path>) -> Result<(Bytes, Metadata)> {
let p = path.as_ref();
let meta = read_file_metadata(&path).await?;
let data = fs::read(&p).await.map_err(to_file_error)?;
Ok((data, meta))
Ok((data.into(), meta))
}
pub async fn read_file_metadata(p: impl AsRef<Path>) -> Result<Metadata> {
@@ -1131,7 +1131,7 @@ impl DiskAPI for LocalDisk {
format_info.id = Some(disk_id);
format_info.file_info = Some(file_meta);
format_info.data = b;
format_info.data = b.into();
format_info.last_check = Some(OffsetDateTime::now_utc());
Ok(Some(disk_id))
@@ -1147,7 +1147,7 @@ impl DiskAPI for LocalDisk {
}
#[tracing::instrument(skip(self))]
async fn read_all(&self, volume: &str, path: &str) -> Result<Vec<u8>> {
async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes> {
if volume == RUSTFS_META_BUCKET && path == super::FORMAT_CONFIG_FILE {
let format_info = self.format_info.read().await;
if !format_info.data.is_empty() {
@@ -1162,7 +1162,7 @@ impl DiskAPI for LocalDisk {
}
#[tracing::instrument(level = "debug", skip_all)]
async fn write_all(&self, volume: &str, path: &str, data: Vec<u8>) -> Result<()> {
async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()> {
self.write_all_public(volume, path, data).await
}
@@ -1331,7 +1331,7 @@ impl DiskAPI for LocalDisk {
rename_all(&src_file_path, &dst_file_path, &dst_volume_dir).await?;
self.write_all(dst_volume, format!("{}.meta", dst_path).as_str(), meta.to_vec())
self.write_all(dst_volume, format!("{}.meta", dst_path).as_str(), meta)
.await?;
if let Some(parent) = src_file_path.parent() {
@@ -1700,7 +1700,7 @@ impl DiskAPI for LocalDisk {
let new_dst_buf = xlmeta.marshal_msg()?;
self.write_all(src_volume, format!("{}/{}", &src_path, STORAGE_FORMAT_FILE).as_str(), new_dst_buf)
self.write_all(src_volume, format!("{}/{}", &src_path, STORAGE_FORMAT_FILE).as_str(), new_dst_buf.into())
.await?;
if let Some((src_data_path, dst_data_path)) = has_data_dir_path.as_ref() {
let no_inline = fi.data.is_none() && fi.size > 0;
@@ -1866,11 +1866,11 @@ impl DiskAPI for LocalDisk {
}
})?;
if !FileMeta::is_xl2_v1_format(buf.as_slice()) {
if !FileMeta::is_xl2_v1_format(buf.as_ref()) {
return Err(DiskError::FileVersionNotFound);
}
let mut xl_meta = FileMeta::load(buf.as_slice())?;
let mut xl_meta = FileMeta::load(buf.as_ref())?;
xl_meta.update_object_version(fi)?;
@@ -1902,7 +1902,7 @@ impl DiskAPI for LocalDisk {
let fm_data = meta.marshal_msg()?;
self.write_all(volume, format!("{}/{}", path, STORAGE_FORMAT_FILE).as_str(), fm_data)
self.write_all(volume, format!("{}/{}", path, STORAGE_FORMAT_FILE).as_str(), fm_data.into())
.await?;
Ok(())
@@ -2076,7 +2076,7 @@ impl DiskAPI for LocalDisk {
}
res.exists = true;
res.data = data;
res.data = data.into();
res.mod_time = match meta.modified() {
Ok(md) => Some(OffsetDateTime::from(md)),
Err(_) => {
@@ -2461,8 +2461,8 @@ mod test {
disk.make_volume("test-volume").await.unwrap();
// Test write and read operations
let test_data = vec![1, 2, 3, 4, 5];
disk.write_all("test-volume", "test-file.txt", test_data.clone())
let test_data: Vec<u8> = vec![1, 2, 3, 4, 5];
disk.write_all("test-volume", "test-file.txt", test_data.clone().into())
.await
.unwrap();
@@ -2587,7 +2587,7 @@ mod test {
// Valid format info
let valid_format_info = FormatInfo {
id: Some(Uuid::new_v4()),
data: vec![1, 2, 3],
data: vec![1, 2, 3].into(),
file_info: Some(fs::metadata(".").await.unwrap()),
last_check: Some(now),
};
@@ -2596,7 +2596,7 @@ mod test {
// Invalid format info (missing id)
let invalid_format_info = FormatInfo {
id: None,
data: vec![1, 2, 3],
data: vec![1, 2, 3].into(),
file_info: Some(fs::metadata(".").await.unwrap()),
last_check: Some(now),
};
@@ -2606,7 +2606,7 @@ mod test {
let old_time = OffsetDateTime::now_utc() - time::Duration::seconds(10);
let old_format_info = FormatInfo {
id: Some(Uuid::new_v4()),
data: vec![1, 2, 3],
data: vec![1, 2, 3].into(),
file_info: Some(fs::metadata(".").await.unwrap()),
last_check: Some(old_time),
};
@@ -2627,7 +2627,7 @@ mod test {
// Test existing file
let (data, metadata) = read_file_exists(test_file).await.unwrap();
assert_eq!(data, b"test content");
assert_eq!(data.as_ref(), b"test content");
assert!(metadata.is_some());
// Clean up
@@ -2644,7 +2644,7 @@ mod test {
// Test reading file
let (data, metadata) = read_file_all(test_file).await.unwrap();
assert_eq!(data, test_content);
assert_eq!(data.as_ref(), test_content);
assert!(metadata.is_file());
assert_eq!(metadata.len(), test_content.len() as u64);
ecstore/src/disk/mod.rs
+4 -4
View File
@@ -364,7 +364,7 @@ impl DiskAPI for Disk {
}
#[tracing::instrument(skip(self))]
async fn write_all(&self, volume: &str, path: &str, data: Vec<u8>) -> Result<()> {
async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()> {
match self {
Disk::Local(local_disk) => local_disk.write_all(volume, path, data).await,
Disk::Remote(remote_disk) => remote_disk.write_all(volume, path, data).await,
@@ -372,7 +372,7 @@ impl DiskAPI for Disk {
}
#[tracing::instrument(skip(self))]
async fn read_all(&self, volume: &str, path: &str) -> Result<Vec<u8>> {
async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes> {
match self {
Disk::Local(local_disk) => local_disk.read_all(volume, path).await,
Disk::Remote(remote_disk) => remote_disk.read_all(volume, path).await,
@@ -504,8 +504,8 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
// ReadParts
async fn read_multiple(&self, req: ReadMultipleReq) -> Result<Vec<ReadMultipleResp>>;
// CleanAbandonedData
async fn write_all(&self, volume: &str, path: &str, data: Vec<u8>) -> Result<()>;
async fn read_all(&self, volume: &str, path: &str) -> Result<Vec<u8>>;
async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()>;
async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes>;
async fn disk_info(&self, opts: &DiskInfoOptions) -> Result<DiskInfo>;
async fn ns_scanner(
&self,
ecstore/src/disk/remote.rs
+4 -4
View File
@@ -804,7 +804,7 @@ impl DiskAPI for RemoteDisk {
}
#[tracing::instrument(skip(self))]
async fn write_all(&self, volume: &str, path: &str, data: Vec<u8>) -> Result<()> {
async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()> {
info!("write_all");
let mut client = node_service_time_out_client(&self.addr)
.await
@@ -813,7 +813,7 @@ impl DiskAPI for RemoteDisk {
disk: self.endpoint.to_string(),
volume: volume.to_string(),
path: path.to_string(),
data: data.into(),
data,
});
let response = client.write_all(request).await?.into_inner();
@@ -826,7 +826,7 @@ impl DiskAPI for RemoteDisk {
}
#[tracing::instrument(skip(self))]
async fn read_all(&self, volume: &str, path: &str) -> Result<Vec<u8>> {
async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes> {
info!("read_all {}/{}", volume, path);
let mut client = node_service_time_out_client(&self.addr)
.await
@@ -843,7 +843,7 @@ impl DiskAPI for RemoteDisk {
return Err(response.error.unwrap_or_default().into());
}
Ok(response.data.into())
Ok(response.data)
}
#[tracing::instrument(skip(self))]
ecstore/src/erasure_coding/encode.rs
+28 -19
View File
@@ -4,6 +4,8 @@ use crate::disk::error::Error;
use crate::disk::error_reduce::count_errs;
use crate::disk::error_reduce::{OBJECT_OP_IGNORED_ERRS, reduce_write_quorum_errs};
use bytes::Bytes;
use futures::StreamExt;
use futures::stream::FuturesUnordered;
use std::sync::Arc;
use std::vec;
use tokio::io::AsyncRead;
@@ -26,33 +28,41 @@ impl<'a> MultiWriter<'a> {
}
}
#[allow(clippy::needless_range_loop)]
pub async fn write(&mut self, data: Vec<Bytes>) -> std::io::Result<()> {
for i in 0..self.writers.len() {
if self.errs[i].is_some() {
continue; // Skip if we already have an error for this writer
}
let writer_opt = &mut self.writers[i];
let shard = &data[i];
if let Some(writer) = writer_opt {
async fn write_shard(writer_opt: &mut Option<BitrotWriterWrapper>, err: &mut Option<Error>, shard: &Bytes) {
match writer_opt {
Some(writer) => {
match writer.write(shard).await {
Ok(n) => {
if n < shard.len() {
self.errs[i] = Some(Error::ShortWrite);
self.writers[i] = None; // Mark as failed
*err = Some(Error::ShortWrite);
*writer_opt = None; // Mark as failed
} else {
self.errs[i] = None;
*err = None;
}
}
Err(e) => {
self.errs[i] = Some(Error::from(e));
*err = Some(Error::from(e));
}
}
} else {
self.errs[i] = Some(Error::DiskNotFound);
}
None => {
*err = Some(Error::DiskNotFound);
}
}
}
pub async fn write(&mut self, data: Vec<Bytes>) -> std::io::Result<()> {
assert_eq!(data.len(), self.writers.len());
{
let mut futures = FuturesUnordered::new();
for ((writer_opt, err), shard) in self.writers.iter_mut().zip(self.errs.iter_mut()).zip(data.iter()) {
if err.is_some() {
continue; // Skip if we already have an error for this writer
}
futures.push(Self::write_shard(writer_opt, err, shard));
}
while let Some(()) = futures.next().await {}
}
let nil_count = self.errs.iter().filter(|&e| e.is_none()).count();
@@ -104,8 +114,8 @@ impl Erasure {
let task = tokio::spawn(async move {
let block_size = self.block_size;
let mut total = 0;
let mut buf = vec![0u8; block_size];
loop {
let mut buf = vec![0u8; block_size];
match rustfs_utils::read_full(&mut reader, &mut buf).await {
Ok(n) if n > 0 => {
total += n;
@@ -122,7 +132,6 @@ impl Erasure {
return Err(e);
}
}
buf.clear();
}
Ok((reader, total))
ecstore/src/heal/heal_commands.rs
+1 -1
View File
@@ -232,7 +232,7 @@ impl HealingTracker {
if let Some(disk) = &self.disk {
let file_path = Path::new(BUCKET_META_PREFIX).join(HEALING_TRACKER_FILENAME);
disk.write_all(RUSTFS_META_BUCKET, file_path.to_str().unwrap(), htracker_bytes)
disk.write_all(RUSTFS_META_BUCKET, file_path.to_str().unwrap(), htracker_bytes.into())
.await?;
}
Ok(())
ecstore/src/store_init.rs
+2 -2
View File
@@ -256,7 +256,7 @@ pub async fn load_format_erasure(disk: &DiskStore, heal: bool) -> disk::error::R
_ => e,
})?;
let mut fm = FormatV3::try_from(data.as_slice())?;
let mut fm = FormatV3::try_from(data.as_ref())?;
if heal {
let info = disk
@@ -311,7 +311,7 @@ pub async fn save_format_file(disk: &Option<DiskStore>, format: &Option<FormatV3
let tmpfile = Uuid::new_v4().to_string();
let disk = disk.as_ref().unwrap();
disk.write_all(RUSTFS_META_BUCKET, tmpfile.as_str(), json_data.into_bytes())
disk.write_all(RUSTFS_META_BUCKET, tmpfile.as_str(), json_data.into_bytes().into())
.await?;
disk.rename_file(RUSTFS_META_BUCKET, tmpfile.as_str(), RUSTFS_META_BUCKET, FORMAT_CONFIG_FILE)
rustfs/src/admin/rpc.rs
+8 -7
View File
@@ -6,7 +6,7 @@ use ecstore::disk::DiskAPI;
use ecstore::disk::WalkDirOptions;
use ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE;
use ecstore::store::find_local_disk;
use futures::TryStreamExt;
use futures::StreamExt;
use http::StatusCode;
use hyper::Method;
use matchit::Params;
@@ -17,8 +17,8 @@ use s3s::S3Result;
use s3s::dto::StreamingBlob;
use s3s::s3_error;
use serde_urlencoded::from_bytes;
use tokio::io::AsyncWriteExt;
use tokio_util::io::ReaderStream;
use tokio_util::io::StreamReader;
use tracing::warn;
pub const RPC_PREFIX: &str = "/rustfs/rpc";
@@ -194,11 +194,12 @@ impl Operation for PutFile {
.map_err(|e| s3_error!(InternalError, "read file err {}", e))?
};
let mut body = StreamReader::new(req.input.into_stream().map_err(std::io::Error::other));
tokio::io::copy(&mut body, &mut file)
.await
.map_err(|e| s3_error!(InternalError, "copy err {}", e))?;
let mut body = req.input;
while let Some(item) = body.next().await {
let bytes = item.map_err(|e| s3_error!(InternalError, "body stream err {}", e))?;
let result = file.write_all(&bytes).await;
result.map_err(|e| s3_error!(InternalError, "write file err {}", e))?;
}
Ok(S3Response::new((StatusCode::OK, Body::empty())))
}
rustfs/src/grpc.rs
+2 -2
View File
@@ -277,7 +277,7 @@ impl Node for NodeService {
match disk.read_all(&request.volume, &request.path).await {
Ok(data) => Ok(tonic::Response::new(ReadAllResponse {
success: true,
data: data.into(),
data,
error: None,
})),
Err(err) => Ok(tonic::Response::new(ReadAllResponse {
@@ -298,7 +298,7 @@ impl Node for NodeService {
async fn write_all(&self, request: Request<WriteAllRequest>) -> Result<Response<WriteAllResponse>, Status> {
let request = request.into_inner();
if let Some(disk) = self.find_disk(&request.disk).await {
match disk.write_all(&request.volume, &request.path, request.data.into()).await {
match disk.write_all(&request.volume, &request.path, request.data).await {
Ok(_) => Ok(tonic::Response::new(WriteAllResponse {
success: true,
error: None,