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

use FileWriter

Browse Source
This commit is contained in:
weisd
2024-07-10 17:16:44 +08:00
parent f991e7b5f3
commit d84f85e636
7 changed files with 119 additions and 187 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ecstore/src/chunk_stream.rs
+34 -35
View File
@@ -5,7 +5,6 @@ use s3s::StdError;
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use tracing::debug;
use transform_stream::AsyncTryStream;
pub type SyncBoxFuture<'a, T> = Pin<Box<dyn Future<Output = T> + Send + Sync + 'a>>;
@@ -37,13 +36,13 @@ impl ChunkedStream {
None => break,
Some(Err(e)) => return Err(e),
Some(Ok((data, remaining_bytes))) => {
debug!(
"content_length:{},readed_size:{}, read_data data:{}, remaining_bytes: {} ",
content_length,
readed_size,
data.len(),
remaining_bytes.len()
);
// debug!(
// "content_length:{},readed_size:{}, read_data data:{}, remaining_bytes: {} ",
// content_length,
// readed_size,
// data.len(),
// remaining_bytes.len()
// );
prev_bytes = remaining_bytes;
data
@@ -53,17 +52,17 @@ impl ChunkedStream {
for bytes in data {
readed_size += bytes.len();
println!("readed_size {}, content_length {}", readed_size, content_length,);
// debug!("readed_size {}, content_length {}", readed_size, content_length,);
y.yield_ok(bytes).await;
}
if readed_size + prev_bytes.len() >= content_length {
println!(
"读完了 readed_size:{} + prev_bytes.len({}) == content_length {}",
readed_size,
prev_bytes.len(),
content_length,
);
// debug!(
// "读完了 readed_size:{} + prev_bytes.len({}) == content_length {}",
// readed_size,
// prev_bytes.len(),
// content_length,
// );
// 填充0？
if !need_padding {
@@ -81,7 +80,7 @@ impl ChunkedStream {
}
}
debug!("chunked stream exit");
// debug!("chunked stream exit");
Ok(())
})
@@ -104,7 +103,7 @@ impl ChunkedStream {
// 只执行一次
let mut push_data_bytes = |mut bytes: Bytes| {
debug!("read from body {} split per {}, prev_bytes: {}", bytes.len(), data_size, prev_bytes.len());
// debug!("read from body {} split per {}, prev_bytes: {}", bytes.len(), data_size, prev_bytes.len());
if bytes.is_empty() {
return None;
@@ -117,24 +116,24 @@ impl ChunkedStream {
// 合并上一次数据
if !prev_bytes.is_empty() {
let need_size = data_size.wrapping_sub(prev_bytes.len());
println!(
" 上一次有剩余{},从这一次中取{},共：{}",
prev_bytes.len(),
need_size,
prev_bytes.len() + need_size
);
// debug!(
// " 上一次有剩余{},从这一次中取{},共：{}",
// prev_bytes.len(),
// need_size,
// prev_bytes.len() + need_size
// );
if bytes.len() >= need_size {
let data = bytes.split_to(need_size);
let mut combined = Vec::new();
combined.extend_from_slice(&prev_bytes);
combined.extend_from_slice(&data);
debug!(
"取到的长度大于所需，取出需要的长度：{},与上一次合并得到：{}，bytes剩余：{}",
need_size,
combined.len(),
bytes.len(),
);
// debug!(
// "取到的长度大于所需，取出需要的长度：{},与上一次合并得到：{}，bytes剩余：{}",
// need_size,
// combined.len(),
// bytes.len(),
// );
bytes_buffer.push(Bytes::from(combined));
} else {
@@ -142,12 +141,12 @@ impl ChunkedStream {
combined.extend_from_slice(&prev_bytes);
combined.extend_from_slice(&bytes);
debug!(
"取到的长度小于所需，取出需要的长度：{},与上一次合并得到：{}，bytes剩余：{}，直接返回",
need_size,
combined.len(),
bytes.len(),
);
// debug!(
// "取到的长度小于所需，取出需要的长度：{},与上一次合并得到：{}，bytes剩余：{}，直接返回",
// need_size,
// combined.len(),
// bytes.len(),
// );
return Some(Bytes::from(combined));
}
ecstore/src/disk.rs
+22 -27
View File
@@ -9,16 +9,13 @@ use bytes::Bytes;
use futures::future::join_all;
use path_absolutize::Absolutize;
use time::OffsetDateTime;
use tokio::io::{self, AsyncWriteExt, BufWriter, ErrorKind};
use tokio::{
fs::{self, File},
io::DuplexStream,
};
use tokio::fs::{self, File};
use tokio::io::ErrorKind;
use tracing::debug;
use uuid::Uuid;
use crate::{
disk_api::{DiskAPI, DiskError, ReadOptions, VolumeInfo},
disk_api::{DiskAPI, DiskError, FileWriter, ReadOptions, VolumeInfo},
endpoint::{Endpoint, Endpoints},
file_meta::FileMeta,
format::FormatV3,
@@ -407,14 +404,7 @@ impl DiskAPI for LocalDisk {
Ok(())
}
async fn create_file(
&self,
_origvolume: &str,
volume: &str,
path: &str,
_file_size: usize,
mut r: DuplexStream,
) -> Result<()> {
async fn create_file(&self, _origvolume: &str, volume: &str, path: &str, _file_size: usize) -> Result<FileWriter> {
let fpath = self.get_object_path(volume, path)?;
debug!("CreateFile fpath: {:?}", fpath);
@@ -425,39 +415,44 @@ impl DiskAPI for LocalDisk {
let file = File::create(&fpath).await?;
let mut writer = BufWriter::new(file);
Ok(FileWriter::new(file))
io::copy(&mut r, &mut writer).await?;
// let mut writer = BufWriter::new(file);
Ok(())
// io::copy(&mut r, &mut writer).await?;
// Ok(())
}
async fn append_file(&self, volume: &str, path: &str, mut r: DuplexStream) -> Result<()> {
// async fn append_file(&self, volume: &str, path: &str, mut r: DuplexStream) -> Result<File> {
async fn append_file(&self, volume: &str, path: &str) -> Result<FileWriter> {
let p = self.get_object_path(&volume, &path)?;
debug!("append_file start {} {:?}", self.id(), &p);
// if let Some(dir_path) = p.parent() {
// fs::create_dir_all(&dir_path).await?;
// }
if let Some(dir_path) = p.parent() {
fs::create_dir_all(&dir_path).await?;
}
// debug!("append_file open {} {:?}", self.id(), &p);
let mut file = File::options()
let file = File::options()
.read(true)
.create(true)
.write(true)
// .append(true)
.append(true)
.open(&p)
.await?;
let mut writer = BufWriter::new(file);
Ok(FileWriter::new(file))
io::copy(&mut r, &mut writer).await?;
// let mut writer = BufWriter::new(file);
debug!("append_file end {} {}", self.id(), path);
// io::copy(&mut r, &mut writer).await?;
// debug!("append_file end {} {}", self.id(), path);
// io::copy(&mut r, &mut file).await?;
Ok(())
// Ok(())
}
async fn rename_data(&self, src_volume: &str, src_path: &str, fi: &FileInfo, dst_volume: &str, dst_path: &str) -> Result<()> {
ecstore/src/disk_api.rs
+41 -4
View File
@@ -1,9 +1,9 @@
use std::fmt::Debug;
use std::{fmt::Debug, pin::Pin};
use anyhow::{Error, Result};
use bytes::Bytes;
use time::OffsetDateTime;
use tokio::io::DuplexStream;
use tokio::io::AsyncWrite;
use uuid::Uuid;
use crate::store_api::{FileInfo, RawFileInfo};
@@ -16,8 +16,8 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes>;
async fn write_all(&self, volume: &str, path: &str, data: Vec<u8>) -> Result<()>;
async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> Result<()>;
async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: usize, r: DuplexStream) -> Result<()>;
async fn append_file(&self, volume: &str, path: &str, r: DuplexStream) -> Result<()>;
async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: usize) -> Result<FileWriter>;
async fn append_file(&self, volume: &str, path: &str) -> Result<FileWriter>;
async fn rename_data(
&self,
src_volume: &str,
@@ -53,6 +53,43 @@ pub struct ReadOptions {
pub healing: bool,
}
pub struct FileWriter {
pub inner: Pin<Box<dyn AsyncWrite + Send + Sync + 'static>>,
}
impl AsyncWrite for FileWriter {
fn poll_write(
mut self: Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &[u8],
) -> std::task::Poll<std::result::Result<usize, std::io::Error>> {
Pin::new(&mut self.inner).poll_write(cx, buf)
}
fn poll_flush(
mut self: Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<std::result::Result<(), std::io::Error>> {
Pin::new(&mut self.inner).poll_flush(cx)
}
fn poll_shutdown(
mut self: Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
) -> std::task::Poll<std::result::Result<(), std::io::Error>> {
Pin::new(&mut self.inner).poll_shutdown(cx)
}
}
impl FileWriter {
pub fn new<W>(inner: W) -> Self
where
W: AsyncWrite + Send + Sync + 'static,
{
Self { inner: Box::pin(inner) }
}
}
#[derive(Debug, thiserror::Error)]
pub enum DiskError {
#[error("file not found")]
ecstore/src/erasure.rs
+18 -18
View File
@@ -54,35 +54,35 @@ impl Erasure {
for (i, w) in writers.iter_mut().enumerate() {
total += blocks[i].len();
debug!(
"{} {}-{} encode write {} , total:{}, readed:{}",
self.id,
idx,
i,
blocks[i].len(),
data_size,
total
);
// debug!(
// "{} {}-{} encode write {} , total:{}, readed:{}",
// self.id,
// idx,
// i,
// blocks[i].len(),
// data_size,
// total
// );
match w.write(blocks[i].as_ref()).await {
match w.write_all(blocks[i].as_ref()).await {
Ok(_) => errs.push(None),
Err(e) => errs.push(Some(e)),
}
}
debug!("{} encode_data write errs:{:?}", self.id, errs);
// TODO: reduceWriteQuorumErrs
for err in errs.iter() {
if err.is_some() {
return Err(Error::msg("message"));
}
}
// debug!("{} encode_data write errs:{:?}", self.id, errs);
// // TODO: reduceWriteQuorumErrs
// for err in errs.iter() {
// if err.is_some() {
// return Err(Error::msg("message"));
// }
// }
}
Err(e) => return Err(anyhow!(e)),
}
}
debug!("{} encode_data done {}", self.id, total);
// debug!("{} encode_data done {}", self.id, total);
Ok(total)
ecstore/src/lib.rs
-1
View File
@@ -16,4 +16,3 @@ pub mod store;
pub mod store_api;
mod store_init;
mod utils;
mod writer;
ecstore/src/sets.rs
+4 -12
View File
@@ -1,23 +1,15 @@
use anyhow::{Error, Result};
use futures::future::join_all;
use time::OffsetDateTime;
use tracing::debug;
use anyhow::Result;
use uuid::Uuid;
use crate::{
disk::{DiskStore, RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_TMP_BUCKET},
disk::DiskStore,
endpoint::PoolEndpoints,
erasure::Erasure,
format::{DistributionAlgoVersion, FormatV3},
set_disk::SetDisks,
store_api::{
BucketInfo, BucketOptions, FileInfo, MakeBucketOptions, MultipartUploadResult, ObjectOptions, PartInfo, PutObjReader,
StorageAPI,
},
utils::{
crypto::{base64_decode, base64_encode, hex, sha256},
hash,
BucketInfo, BucketOptions, MakeBucketOptions, MultipartUploadResult, ObjectOptions, PartInfo, PutObjReader, StorageAPI,
},
utils::hash,
};
#[derive(Debug)]
ecstore/src/writer.rs
-90
View File
@@ -1,90 +0,0 @@
use std::{io, task::Poll};
use futures::{ready, Future};
use tokio::io::{AsyncWrite, BufWriter};
use tracing::debug;
use uuid::Uuid;
use crate::disk::DiskStore;
pub struct AppendWriter<'a> {
disk: DiskStore,
volume: &'a str,
path: &'a str,
}
impl<'a> AppendWriter<'a> {
pub fn new(disk: DiskStore, volume: &'a str, path: &'a str) -> Self {
debug!("AppendWriter new {}: {}/{}", disk.id(), volume, path);
Self { disk, volume, path }
}
async fn async_write(&self, buf: &[u8]) -> Result<(), std::io::Error> {
debug!("async_write {}: {}: {}", self.disk.id(), &self.path, buf.len());
// self.disk
// .append_file(&self.volume, &self.path, buf)
// .await
// .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))?;
Ok(())
}
}
impl<'a> AsyncWrite for AppendWriter<'a> {
fn poll_write(
self: std::pin::Pin<&mut Self>,
cx: &mut std::task::Context<'_>,
buf: &[u8],
) -> std::task::Poll<Result<usize, std::io::Error>> {
let mut fut = Box::pin(self.async_write(buf));
debug!("AsyncWrite poll_write {}, buf:{}", self.disk.id(), buf.len());
// while let Poll::Ready(e) = fut.as_mut().poll(cx) {
// let a = match e {
// Ok(_) => {
// debug!("Ready ok {}", self.disk.id());
// Poll::Ready(Ok(buf.len()))
// }
// Err(e) => {
// debug!("Ready err {}", self.disk.id());
// Poll::Ready(Err(e))
// }
// };
// return a;
// }
// Poll::Pending
match fut.as_mut().poll(cx) {
Poll::Pending => {
debug!("Pending {}", self.disk.id());
Poll::Pending
}
Poll::Ready(e) => match e {
Ok(_) => {
debug!("Ready ok {}", self.disk.id());
Poll::Ready(Ok(buf.len()))
}
Err(e) => {
debug!("Ready err {}", self.disk.id());
Poll::Ready(Err(e))
}
},
}
}
fn poll_flush(
self: std::pin::Pin<&mut Self>,
_cx: &mut std::task::Context<'_>,
) -> std::task::Poll<Result<(), std::io::Error>> {
Poll::Ready(Ok(()))
}
fn poll_shutdown(
self: std::pin::Pin<&mut Self>,
_cx: &mut std::task::Context<'_>,
) -> std::task::Poll<Result<(), std::io::Error>> {
Poll::Ready(Ok(()))
}
}