gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-21 03:46:37 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'main' into dada/fix-entry

Browse Source
This commit is contained in:
weisd
2025-05-09 14:33:30 +08:00
committed by GitHub
parent 79d58a98f4 e90b531567
commit d7cac5c727
76 changed files with 1384 additions and 550 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ecstore/src/bitrot.rs
+1
View File
@@ -534,6 +534,7 @@ impl Writer for BitrotFileWriter {
self
}
#[tracing::instrument(level = "info", skip_all)]
async fn write(&mut self, buf: Bytes) -> Result<()> {
if buf.is_empty() {
return Ok(());
ecstore/src/disk/local.rs
+13 -10
View File
@@ -38,7 +38,9 @@ use crate::set_disk::{
CHECK_PART_VOLUME_NOT_FOUND,
};
use crate::store_api::{BitrotAlgorithm, StorageAPI};
use crate::utils::fs::{access, lstat, remove, remove_all, rename, O_APPEND, O_CREATE, O_RDONLY, O_WRONLY};
use crate::utils::fs::{
access, lstat, remove, remove_all, remove_all_std, remove_std, rename, O_APPEND, O_CREATE, O_RDONLY, O_WRONLY,
};
use crate::utils::os::get_info;
use crate::utils::path::{
self, clean, decode_dir_object, encode_dir_object, has_suffix, path_join, path_join_buf, GLOBAL_DIR_SUFFIX,
@@ -259,7 +261,7 @@ impl LocalDisk {
#[tracing::instrument(level = "debug", skip(self))]
async fn check_format_json(&self) -> Result<Metadata> {
let md = fs::metadata(&self.format_path).await.map_err(|e| match e.kind() {
let md = std::fs::metadata(&self.format_path).map_err(|e| match e.kind() {
ErrorKind::NotFound => DiskError::DiskNotFound,
ErrorKind::PermissionDenied => DiskError::FileAccessDenied,
_ => {
@@ -315,9 +317,9 @@ impl LocalDisk {
#[allow(unused_variables)]
pub async fn move_to_trash(&self, delete_path: &PathBuf, recursive: bool, immediate_purge: bool) -> Result<()> {
if recursive {
remove_all(delete_path).await?;
remove_all_std(delete_path)?;
} else {
remove(delete_path).await?;
remove_std(delete_path)?;
}
return Ok(());
@@ -365,7 +367,7 @@ impl LocalDisk {
Ok(())
}
// #[tracing::instrument(skip(self))]
#[tracing::instrument(level = "debug", skip(self))]
pub async fn delete_file(
&self,
base_path: &PathBuf,
@@ -688,6 +690,7 @@ impl LocalDisk {
}
// write_all_private with check_path_length
#[tracing::instrument(level = "debug", skip_all)]
pub async fn write_all_private(
&self,
volume: &str,
@@ -1213,7 +1216,7 @@ impl DiskAPI for LocalDisk {
Ok(data)
}
#[tracing::instrument(skip(self))]
#[tracing::instrument(level = "debug", skip_all)]
async fn write_all(&self, volume: &str, path: &str, data: Vec<u8>) -> Result<()> {
self.write_all_public(volume, path, data).await
}
@@ -1721,7 +1724,7 @@ impl DiskAPI for LocalDisk {
Ok(())
}
#[tracing::instrument(skip(self))]
#[tracing::instrument(level = "debug", skip(self))]
async fn rename_data(
&self,
src_volume: &str,
@@ -1732,7 +1735,7 @@ impl DiskAPI for LocalDisk {
) -> Result<RenameDataResp> {
let src_volume_dir = self.get_bucket_path(src_volume)?;
if !skip_access_checks(src_volume) {
if let Err(e) = utils::fs::access(&src_volume_dir).await {
if let Err(e) = utils::fs::access_std(&src_volume_dir) {
info!("access checks failed, src_volume_dir: {:?}, err: {}", src_volume_dir, e.to_string());
return Err(convert_access_error(e, DiskError::VolumeAccessDenied));
}
@@ -1740,7 +1743,7 @@ impl DiskAPI for LocalDisk {
let dst_volume_dir = self.get_bucket_path(dst_volume)?;
if !skip_access_checks(dst_volume) {
if let Err(e) = utils::fs::access(&dst_volume_dir).await {
if let Err(e) = utils::fs::access_std(&dst_volume_dir) {
info!("access checks failed, dst_volume_dir: {:?}, err: {}", dst_volume_dir, e.to_string());
return Err(convert_access_error(e, DiskError::VolumeAccessDenied));
}
@@ -1913,7 +1916,7 @@ impl DiskAPI for LocalDisk {
if let Some(src_file_path_parent) = src_file_path.parent() {
if src_volume != super::RUSTFS_META_MULTIPART_BUCKET {
let _ = utils::fs::remove(src_file_path_parent).await;
let _ = utils::fs::remove_std(src_file_path_parent);
} else {
let _ = self
.delete_file(&dst_volume_dir, &src_file_path_parent.to_path_buf(), true, false)
ecstore/src/disk/os.rs
+5 -4
View File
@@ -108,6 +108,7 @@ pub async fn read_dir(path: impl AsRef<Path>, count: i32) -> Result<Vec<String>>
Ok(volumes)
}
#[tracing::instrument(level = "debug", skip_all)]
pub async fn rename_all(
src_file_path: impl AsRef<Path>,
dst_file_path: impl AsRef<Path>,
@@ -136,7 +137,7 @@ pub async fn reliable_rename(
base_dir: impl AsRef<Path>,
) -> io::Result<()> {
if let Some(parent) = dst_file_path.as_ref().parent() {
if !file_exists(parent).await {
if !file_exists(parent) {
info!("reliable_rename reliable_mkdir_all parent: {:?}", parent);
reliable_mkdir_all(parent, base_dir.as_ref()).await?;
}
@@ -144,7 +145,7 @@ pub async fn reliable_rename(
let mut i = 0;
loop {
if let Err(e) = utils::fs::rename(src_file_path.as_ref(), dst_file_path.as_ref()).await {
if let Err(e) = utils::fs::rename_std(src_file_path.as_ref(), dst_file_path.as_ref()) {
if os_is_not_exist(&e) && i == 0 {
i += 1;
continue;
@@ -221,6 +222,6 @@ pub async fn os_mkdir_all(dir_path: impl AsRef<Path>, base_dir: impl AsRef<Path>
Ok(())
}
pub async fn file_exists(path: impl AsRef<Path>) -> bool {
fs::metadata(path.as_ref()).await.map(|_| true).unwrap_or(false)
pub fn file_exists(path: impl AsRef<Path>) -> bool {
std::fs::metadata(path.as_ref()).map(|_| true).unwrap_or(false)
}
ecstore/src/erasure.rs
+53 -124
View File
@@ -1,5 +1,6 @@
use crate::bitrot::{BitrotReader, BitrotWriter};
use crate::error::clone_err;
use crate::io::Etag;
use crate::quorum::{object_op_ignored_errs, reduce_write_quorum_errs};
use bytes::{Bytes, BytesMut};
use common::error::{Error, Result};
@@ -8,8 +9,10 @@ use reed_solomon_erasure::galois_8::ReedSolomon;
use smallvec::SmallVec;
use std::any::Any;
use std::io::ErrorKind;
use std::sync::Arc;
use tokio::io::{AsyncRead, AsyncWrite};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::sync::mpsc;
use tracing::warn;
use tracing::{error, info};
// use tracing::debug;
@@ -26,7 +29,7 @@ pub struct Erasure {
encoder: Option<ReedSolomon>,
pub block_size: usize,
_id: Uuid,
buf: Vec<u8>,
_buf: Vec<u8>,
}
impl Erasure {
@@ -46,61 +49,65 @@ impl Erasure {
block_size,
encoder,
_id: Uuid::new_v4(),
buf: vec![0u8; block_size],
_buf: vec![0u8; block_size],
}
}
#[tracing::instrument(level = "debug", skip(self, reader, writers))]
#[tracing::instrument(level = "info", skip(self, reader, writers))]
pub async fn encode<S>(
&mut self,
reader: &mut S,
self: Arc<Self>,
mut reader: S,
writers: &mut [Option<BitrotWriter>],
// block_size: usize,
total_size: usize,
write_quorum: usize,
) -> Result<usize>
) -> Result<(usize, String)>
where
S: AsyncRead + Unpin + Send + 'static,
S: AsyncRead + Etag + Unpin + Send + 'static,
{
// pin_mut!(body);
// let mut reader = tokio_util::io::StreamReader::new(
// body.map(|f| f.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))),
// );
let (tx, mut rx) = mpsc::channel(5);
let task = tokio::spawn(async move {
let mut buf = vec![0u8; self.block_size];
let mut total: usize = 0;
loop {
if total_size > 0 {
let new_len = {
let remain = total_size - total;
if remain > self.block_size {
self.block_size
} else {
remain
}
};
let mut total: usize = 0;
let mut blocks = <SmallVec<[Bytes; 16]>>::new();
loop {
if total_size > 0 {
let new_len = {
let remain = total_size - total;
if remain > self.block_size {
self.block_size
} else {
remain
if new_len == 0 && total > 0 {
break;
}
};
if new_len == 0 && total > 0 {
buf.resize(new_len, 0u8);
match reader.read_exact(&mut buf).await {
Ok(res) => res,
Err(e) => {
if let ErrorKind::UnexpectedEof = e.kind() {
break;
} else {
return Err(Error::new(e));
}
}
};
total += buf.len();
}
let blocks = Arc::new(Box::pin(self.clone().encode_data(&buf)?));
let _ = tx.send(blocks).await;
if total_size == 0 {
break;
}
self.buf.resize(new_len, 0u8);
match reader.read_exact(&mut self.buf).await {
Ok(res) => res,
Err(e) => {
if let ErrorKind::UnexpectedEof = e.kind() {
break;
} else {
return Err(Error::new(e));
}
}
};
total += self.buf.len();
}
let etag = reader.etag().await;
Ok((total, etag))
});
self.encode_data(&self.buf, &mut blocks)?;
while let Some(blocks) = rx.recv().await {
let write_futures = writers.iter_mut().enumerate().map(|(i, w_op)| {
let i_inner = i;
let blocks_inner = blocks.clone();
@@ -125,84 +132,8 @@ impl Erasure {
warn!("Erasure encode errs {:?}", &errs);
return Err(err);
}
if total_size == 0 {
break;
}
}
Ok(total)
// // let stream = ChunkedStream::new(body, self.block_size);
// let stream = ChunkedStream::new(body, total_size, self.block_size, false);
// let mut total: usize = 0;
// // let mut idx = 0;
// pin_mut!(stream);
// // warn!("encode start...");
// loop {
// match stream.next().await {
// Some(result) => match result {
// Ok(data) => {
// total += data.len();
// // EOF
// if data.is_empty() {
// break;
// }
// // idx += 1;
// // warn!("encode {} get data {:?}", data.len(), data.to_vec());
// let blocks = self.encode_data(data.as_ref())?;
// // warn!(
// // "encode shard size: {}/{} from block_size {}, total_size {} ",
// // blocks[0].len(),
// // blocks.len(),
// // data.len(),
// // total_size
// // );
// let mut errs = Vec::new();
// for (i, w_op) in writers.iter_mut().enumerate() {
// if let Some(w) = w_op {
// match w.write(blocks[i].as_ref()).await {
// Ok(_) => errs.push(None),
// Err(e) => errs.push(Some(e)),
// }
// } else {
// errs.push(Some(Error::new(DiskError::DiskNotFound)));
// }
// }
// let none_count = errs.iter().filter(|&x| x.is_none()).count();
// if none_count >= write_quorum {
// continue;
// }
// if let Some(err) = reduce_write_quorum_errs(&errs, object_op_ignored_errs().as_ref(), write_quorum) {
// warn!("Erasure encode errs {:?}", &errs);
// return Err(err);
// }
// }
// Err(e) => {
// warn!("poll result err {:?}", &e);
// return Err(Error::msg(e.to_string()));
// }
// },
// None => {
// // warn!("poll empty result");
// break;
// }
// }
// }
// let _ = close_bitrot_writers(writers).await?;
// Ok(total)
task.await?
}
pub async fn decode<W>(
@@ -356,8 +287,8 @@ impl Erasure {
self.data_shards + self.parity_shards
}
#[tracing::instrument(level = "debug", skip_all, fields(data_len=data.len()))]
pub fn encode_data(&self, data: &[u8], shards: &mut SmallVec<[Bytes; 16]>) -> Result<()> {
#[tracing::instrument(level = "info", skip_all, fields(data_len=data.len()))]
pub fn encode_data(self: Arc<Self>, data: &[u8]) -> Result<Vec<Bytes>> {
let (shard_size, total_size) = self.need_size(data.len());
// 生成一个新的 所需的所有分片数据长度
@@ -379,14 +310,13 @@ impl Erasure {
// 零拷贝分片，所有 shard 引用 data_buffer
let mut data_buffer = data_buffer.freeze();
shards.clear();
shards.reserve(self.total_shard_count());
let mut shards = Vec::with_capacity(self.total_shard_count());
for _ in 0..self.total_shard_count() {
let shard = data_buffer.split_to(shard_size);
shards.push(shard);
}
Ok(())
Ok(shards)
}
pub fn decode_data(&self, shards: &mut [Option<Vec<u8>>]) -> Result<()> {
@@ -617,7 +547,6 @@ impl ShardReader {
#[cfg(test)]
mod test {
use super::*;
#[test]
@@ -626,8 +555,7 @@ mod test {
let parity_shards = 2;
let data: &[u8] = &[0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11];
let ec = Erasure::new(data_shards, parity_shards, 1);
let mut shards = SmallVec::new();
ec.encode_data(data, &mut shards).unwrap();
let shards = Arc::new(ec).encode_data(data).unwrap();
println!("shards:{:?}", shards);
let mut s: Vec<_> = shards
@@ -643,6 +571,7 @@ mod test {
println!("sss:{:?}", &s);
let ec = Erasure::new(data_shards, parity_shards, 1);
ec.decode_data(&mut s).unwrap();
// ec.encoder.reconstruct(&mut s).unwrap();
ecstore/src/file_meta.rs
+5 -1
View File
@@ -58,10 +58,12 @@ impl FileMeta {
}
// isXL2V1Format
#[tracing::instrument(level = "debug", skip_all)]
pub fn is_xl2_v1_format(buf: &[u8]) -> bool {
!matches!(Self::check_xl2_v1(buf), Err(_e))
}
#[tracing::instrument(level = "debug", skip_all)]
pub fn load(buf: &[u8]) -> Result<FileMeta> {
let mut xl = FileMeta::default();
xl.unmarshal_msg(buf)?;
@@ -245,7 +247,7 @@ impl FileMeta {
}
}
#[tracing::instrument]
#[tracing::instrument(level = "debug", skip_all)]
pub fn marshal_msg(&self) -> Result<Vec<u8>> {
let mut wr = Vec::new();
@@ -363,6 +365,7 @@ impl FileMeta {
}
// shard_data_dir_count 查询 vid下data_dir的数量
#[tracing::instrument(level = "debug", skip_all)]
pub fn shard_data_dir_count(&self, vid: &Option<Uuid>, data_dir: &Option<Uuid>) -> usize {
self.versions
.iter()
@@ -434,6 +437,7 @@ impl FileMeta {
}
// 添加版本
#[tracing::instrument(level = "debug", skip_all)]
pub fn add_version(&mut self, fi: FileInfo) -> Result<()> {
let vid = fi.version_id;
ecstore/src/global.rs
+8 -4
View File
@@ -20,8 +20,6 @@ pub const DISK_MIN_INODES: u64 = 1000;
pub const DISK_FILL_FRACTION: f64 = 0.99;
pub const DISK_RESERVE_FRACTION: f64 = 0.15;
pub const DEFAULT_PORT: u16 = 9000;
lazy_static! {
static ref GLOBAL_RUSTFS_PORT: OnceLock<u16> = OnceLock::new();
pub static ref GLOBAL_OBJECT_API: OnceLock<Arc<ECStore>> = OnceLock::new();
@@ -41,31 +39,37 @@ lazy_static! {
pub static ref GLOBAL_BOOT_TIME: OnceCell<SystemTime> = OnceCell::new();
}
/// Get the global rustfs port
pub fn global_rustfs_port() -> u16 {
if let Some(p) = GLOBAL_RUSTFS_PORT.get() {
*p
} else {
DEFAULT_PORT
rustfs_config::DEFAULT_PORT
}
}
/// Set the global rustfs port
pub fn set_global_rustfs_port(value: u16) {
GLOBAL_RUSTFS_PORT.set(value).expect("set_global_rustfs_port fail");
}
/// Get the global rustfs port
pub fn set_global_deployment_id(id: Uuid) {
globalDeploymentIDPtr.set(id).unwrap();
}
/// Get the global deployment id
pub fn get_global_deployment_id() -> Option<String> {
globalDeploymentIDPtr.get().map(|v| v.to_string())
}
/// Get the global deployment id
pub fn set_global_endpoints(eps: Vec<PoolEndpoints>) {
GLOBAL_Endpoints
.set(EndpointServerPools::from(eps))
.expect("GLOBAL_Endpoints set failed")
}
/// Get the global endpoints
pub fn get_global_endpoints() -> EndpointServerPools {
if let Some(eps) = GLOBAL_Endpoints.get() {
eps.clone()
ecstore/src/heal/heal_ops.rs
+2 -2
View File
@@ -17,7 +17,7 @@ use crate::{
global::GLOBAL_IsDistErasure,
heal::heal_commands::{HealStartSuccess, HEAL_UNKNOWN_SCAN},
new_object_layer_fn,
utils::path::has_profix,
utils::path::has_prefix,
};
use crate::{
heal::heal_commands::{HEAL_ITEM_BUCKET, HEAL_ITEM_OBJECT},
@@ -786,7 +786,7 @@ impl AllHealState {
let _ = self.mu.write().await;
for (k, v) in self.heal_seq_map.read().await.iter() {
if (has_profix(k, path_s) || has_profix(path_s, k)) && !v.has_ended().await {
if (has_prefix(k, path_s) || has_prefix(path_s, k)) && !v.has_ended().await {
return Err(Error::from_string(format!(
"The provided heal sequence path overlaps with an existing heal path: {}",
k
ecstore/src/io.rs
+33 -12
View File
@@ -1,8 +1,12 @@
use async_trait::async_trait;
use bytes::Bytes;
use futures::TryStreamExt;
use md5::Digest;
use md5::Md5;
use pin_project_lite::pin_project;
use std::io;
use std::pin::Pin;
use std::task::ready;
use std::task::Context;
use std::task::Poll;
use tokio::io::AsyncRead;
@@ -125,10 +129,17 @@ impl AsyncRead for HttpFileReader {
}
}
pub struct EtagReader<R> {
inner: R,
bytes_tx: mpsc::Sender<Bytes>,
md5_rx: oneshot::Receiver<String>,
#[async_trait]
pub trait Etag {
async fn etag(self) -> String;
}
pin_project! {
pub struct EtagReader<R> {
inner: R,
bytes_tx: mpsc::Sender<Bytes>,
md5_rx: oneshot::Receiver<String>,
}
}
impl<R> EtagReader<R> {
@@ -148,8 +159,11 @@ impl<R> EtagReader<R> {
EtagReader { inner, bytes_tx, md5_rx }
}
}
pub async fn etag(self) -> String {
#[async_trait]
impl<R: Send> Etag for EtagReader<R> {
async fn etag(self) -> String {
drop(self.inner);
drop(self.bytes_tx);
self.md5_rx.await.unwrap()
@@ -157,21 +171,28 @@ impl<R> EtagReader<R> {
}
impl<R: AsyncRead + Unpin> AsyncRead for EtagReader<R> {
#[tracing::instrument(level = "debug", skip_all)]
fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<tokio::io::Result<()>> {
let poll = Pin::new(&mut self.inner).poll_read(cx, buf);
if let Poll::Ready(Ok(())) = &poll {
if buf.remaining() == 0 {
#[tracing::instrument(level = "info", skip_all)]
fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<tokio::io::Result<()>> {
let me = self.project();
loop {
let rem = buf.remaining();
if rem != 0 {
ready!(Pin::new(&mut *me.inner).poll_read(cx, buf))?;
if buf.remaining() == rem {
return Err(io::Error::new(io::ErrorKind::UnexpectedEof, "early eof")).into();
}
} else {
let bytes = buf.filled();
let bytes = Bytes::copy_from_slice(bytes);
let tx = self.bytes_tx.clone();
let tx = me.bytes_tx.clone();
tokio::spawn(async move {
if let Err(e) = tx.send(bytes).await {
warn!("EtagReader send error: {:?}", e);
}
});
return Poll::Ready(Ok(()));
}
}
poll
}
}
ecstore/src/quorum.rs
+1
View File
@@ -147,6 +147,7 @@ pub fn reduce_read_quorum_errs(
// 根据写quorum验证错误数量
// 返回最大错误数量的下标，或QuorumError
#[tracing::instrument(level = "info", skip_all)]
pub fn reduce_write_quorum_errs(
errs: &[Option<Error>],
ignored_errs: &[Box<dyn CheckErrorFn>],
ecstore/src/set_disk.rs
+62 -54
View File
@@ -284,10 +284,20 @@ impl SetDisks {
// let mut ress = Vec::with_capacity(disks.len());
let mut errs = Vec::with_capacity(disks.len());
for (i, disk) in disks.iter().enumerate() {
let mut file_info = file_infos[i].clone();
let src_bucket = Arc::new(src_bucket.to_string());
let src_object = Arc::new(src_object.to_string());
let dst_bucket = Arc::new(dst_bucket.to_string());
let dst_object = Arc::new(dst_object.to_string());
futures.push(async move {
for (i, (disk, file_info)) in disks.iter().zip(file_infos.iter()).enumerate() {
let mut file_info = file_info.clone();
let disk = disk.clone();
let src_bucket = src_bucket.clone();
let src_object = src_object.clone();
let dst_object = dst_object.clone();
let dst_bucket = dst_bucket.clone();
futures.push(tokio::spawn(async move {
if file_info.erasure.index == 0 {
file_info.erasure.index = i + 1;
}
@@ -297,12 +307,12 @@ impl SetDisks {
}
if let Some(disk) = disk {
disk.rename_data(src_bucket, src_object, file_info, dst_bucket, dst_object)
disk.rename_data(&src_bucket, &src_object, file_info, &dst_bucket, &dst_object)
.await
} else {
Err(Error::new(DiskError::DiskNotFound))
}
})
}));
}
let mut disk_versions = vec![None; disks.len()];
@@ -311,15 +321,13 @@ impl SetDisks {
let results = join_all(futures).await;
for (idx, result) in results.iter().enumerate() {
match result {
match result.as_ref().map_err(|_| Error::new(DiskError::Unexpected))? {
Ok(res) => {
data_dirs[idx] = res.old_data_dir;
disk_versions[idx].clone_from(&res.sign);
// ress.push(Some(res));
errs.push(None);
}
Err(e) => {
// ress.push(None);
errs.push(Some(clone_err(e)));
}
}
@@ -336,11 +344,14 @@ impl SetDisks {
if let Some(disk) = disks[i].as_ref() {
let fi = file_infos[i].clone();
let old_data_dir = data_dirs[i];
futures.push(async move {
let disk = disk.clone();
let src_bucket = src_bucket.clone();
let src_object = src_object.clone();
futures.push(tokio::spawn(async move {
let _ = disk
.delete_version(
src_bucket,
src_object,
&src_bucket,
&src_object,
fi,
false,
DeleteOptions {
@@ -354,7 +365,7 @@ impl SetDisks {
debug!("rename_data delete_version err {:?}", e);
e
});
});
}));
}
}
@@ -416,41 +427,41 @@ impl SetDisks {
data_dir: &str,
write_quorum: usize,
) -> Result<()> {
let file_path = format!("{}/{}", object, data_dir);
let mut futures = Vec::with_capacity(disks.len());
let mut errs = Vec::with_capacity(disks.len());
for disk in disks.iter() {
let file_path = Arc::new(format!("{}/{}", object, data_dir));
let bucket = Arc::new(bucket.to_string());
let futures = disks.iter().map(|disk| {
let file_path = file_path.clone();
futures.push(async move {
let bucket = bucket.clone();
let disk = disk.clone();
tokio::spawn(async move {
if let Some(disk) = disk {
disk.delete(
bucket,
&file_path,
DeleteOptions {
recursive: true,
..Default::default()
},
)
.await
match disk
.delete(
&bucket,
&file_path,
DeleteOptions {
recursive: true,
..Default::default()
},
)
.await
{
Ok(_) => None,
Err(e) => Some(e),
}
} else {
Err(Error::new(DiskError::DiskNotFound))
Some(Error::new(DiskError::DiskNotFound))
}
});
}
let results = join_all(futures).await;
for result in results {
match result {
Ok(_) => {
errs.push(None);
}
Err(e) => {
errs.push(Some(e));
}
}
}
})
});
let errs: Vec<Option<Error>> = join_all(futures)
.await
.into_iter()
.map(|e| match e {
Ok(e) => e,
Err(_) => Some(Error::new(DiskError::Unexpected)),
})
.collect();
if let Some(err) = reduce_write_quorum_errs(&errs, object_op_ignored_errs().as_ref(), write_quorum) {
return Err(err);
@@ -3759,7 +3770,7 @@ impl ObjectIO for SetDisks {
let tmp_object = format!("{}/{}/part.1", tmp_dir, fi.data_dir.unwrap());
let mut erasure = Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
let erasure = Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
let is_inline_buffer = {
if let Some(sc) = GLOBAL_StorageClass.get() {
@@ -3798,19 +3809,18 @@ impl ObjectIO for SetDisks {
}
let stream = replace(&mut data.stream, Box::new(empty()));
let mut etag_stream = EtagReader::new(stream);
let etag_stream = EtagReader::new(stream);
// TODO: etag from header
let w_size = erasure
.encode(&mut etag_stream, &mut writers, data.content_length, write_quorum)
let (w_size, etag) = Arc::new(erasure)
.encode(etag_stream, &mut writers, data.content_length, write_quorum)
.await?; // TODO: 出错，删除临时目录
if let Err(err) = close_bitrot_writers(&mut writers).await {
error!("close_bitrot_writers err {:?}", err);
}
let etag = etag_stream.etag().await;
//TODO: userDefined
user_defined.insert("etag".to_owned(), etag.clone());
@@ -4408,21 +4418,19 @@ impl StorageAPI for SetDisks {
}
}
let mut erasure = Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
let erasure = Erasure::new(fi.erasure.data_blocks, fi.erasure.parity_blocks, fi.erasure.block_size);
let stream = replace(&mut data.stream, Box::new(empty()));
let mut etag_stream = EtagReader::new(stream);
let etag_stream = EtagReader::new(stream);
let w_size = erasure
.encode(&mut etag_stream, &mut writers, data.content_length, write_quorum)
let (w_size, mut etag) = Arc::new(erasure)
.encode(etag_stream, &mut writers, data.content_length, write_quorum)
.await?;
if let Err(err) = close_bitrot_writers(&mut writers).await {
error!("close_bitrot_writers err {:?}", err);
}
let mut etag = etag_stream.etag().await;
if let Some(ref tag) = opts.preserve_etag {
etag = tag.clone();
}
ecstore/src/store.rs
+1
View File
@@ -2560,6 +2560,7 @@ fn check_abort_multipart_args(bucket: &str, object: &str, upload_id: &str) -> Re
check_multipart_object_args(bucket, object, upload_id)
}
#[tracing::instrument(level = "debug")]
fn check_put_object_args(bucket: &str, object: &str) -> Result<()> {
if !is_meta_bucketname(bucket) && check_valid_bucket_name_strict(bucket).is_err() {
return Err(Error::new(StorageError::BucketNameInvalid(bucket.to_string())));
ecstore/src/utils/fs.rs
+30
View File
@@ -106,6 +106,11 @@ pub async fn access(path: impl AsRef<Path>) -> io::Result<()> {
Ok(())
}
pub fn access_std(path: impl AsRef<Path>) -> io::Result<()> {
std::fs::metadata(path)?;
Ok(())
}
pub async fn lstat(path: impl AsRef<Path>) -> io::Result<Metadata> {
fs::metadata(path).await
}
@@ -114,6 +119,7 @@ pub async fn make_dir_all(path: impl AsRef<Path>) -> io::Result<()> {
fs::create_dir_all(path.as_ref()).await
}
#[tracing::instrument(level = "debug", skip_all)]
pub async fn remove(path: impl AsRef<Path>) -> io::Result<()> {
let meta = fs::metadata(path.as_ref()).await?;
if meta.is_dir() {
@@ -132,6 +138,25 @@ pub async fn remove_all(path: impl AsRef<Path>) -> io::Result<()> {
}
}
#[tracing::instrument(level = "debug", skip_all)]
pub fn remove_std(path: impl AsRef<Path>) -> io::Result<()> {
let meta = std::fs::metadata(path.as_ref())?;
if meta.is_dir() {
std::fs::remove_dir(path.as_ref())
} else {
std::fs::remove_file(path.as_ref())
}
}
pub fn remove_all_std(path: impl AsRef<Path>) -> io::Result<()> {
let meta = std::fs::metadata(path.as_ref())?;
if meta.is_dir() {
std::fs::remove_dir_all(path.as_ref())
} else {
std::fs::remove_file(path.as_ref())
}
}
pub async fn mkdir(path: impl AsRef<Path>) -> io::Result<()> {
fs::create_dir(path.as_ref()).await
}
@@ -140,6 +165,11 @@ pub async fn rename(from: impl AsRef<Path>, to: impl AsRef<Path>) -> io::Result<
fs::rename(from, to).await
}
pub fn rename_std(from: impl AsRef<Path>, to: impl AsRef<Path>) -> io::Result<()> {
std::fs::rename(from, to)
}
#[tracing::instrument(level = "debug", skip_all)]
pub async fn read_file(path: impl AsRef<Path>) -> io::Result<Vec<u8>> {
fs::read(path.as_ref()).await
}
ecstore/src/utils/path.rs
+1 -1
View File
@@ -52,7 +52,7 @@ pub fn strings_has_prefix_fold(s: &str, prefix: &str) -> bool {
s.len() >= prefix.len() && (s[..prefix.len()] == *prefix || s[..prefix.len()].eq_ignore_ascii_case(prefix))
}
pub fn has_profix(s: &str, prefix: &str) -> bool {
pub fn has_prefix(s: &str, prefix: &str) -> bool {
if cfg!(target_os = "windows") {
return strings_has_prefix_fold(s, prefix);
}