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

fix: 优化unused

Browse Source
This commit is contained in:
weisd
2024-07-09 11:31:20 +08:00
parent e3fe26ae72
commit c4775920f4
10 changed files with 247 additions and 258 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ecstore/src/disk.rs
+68 -62
View File
@@ -1,17 +1,15 @@
use std::{ use std::{
fs::Metadata, fs::Metadata,
path::{Path, PathBuf}, path::{Path, PathBuf},
sync::{Arc, RwLock}, sync::Arc,
}; };
use anyhow::{Error, Result}; use anyhow::{Error, Result};
use bytes::Bytes; use bytes::Bytes;
use futures::{future::join_all, Stream}; use futures::future::join_all;
use path_absolutize::Absolutize; use path_absolutize::Absolutize;
use s3s::StdError;
use time::OffsetDateTime; use time::OffsetDateTime;
use tokio::io::{self, AsyncRead, AsyncReadExt, AsyncSeekExt}; use tokio::io::{self, BufWriter, ErrorKind};
use tokio::io::{AsyncWrite, BufWriter, ErrorKind};
use tokio::{ use tokio::{
fs::{self, File}, fs::{self, File},
io::DuplexStream, io::DuplexStream,
@@ -23,7 +21,7 @@ use crate::{
disk_api::{DiskAPI, DiskError, ReadOptions, VolumeInfo}, disk_api::{DiskAPI, DiskError, ReadOptions, VolumeInfo},
endpoint::{Endpoint, Endpoints}, endpoint::{Endpoint, Endpoints},
file_meta::FileMeta, file_meta::FileMeta,
format::{DistributionAlgoVersion, FormatV3}, format::FormatV3,
store_api::{FileInfo, RawFileInfo}, store_api::{FileInfo, RawFileInfo},
utils, utils,
}; };
@@ -48,6 +46,7 @@ pub async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> Result<DiskStore> {
let s = LocalDisk::new(ep, opt.cleanup).await?; let s = LocalDisk::new(ep, opt.cleanup).await?;
Ok(Arc::new(Box::new(s))) Ok(Arc::new(Box::new(s)))
} else { } else {
let _ = opt.health_check;
unimplemented!() unimplemented!()
// Ok(Disk::Remote(RemoteDisk::new(ep, opt.health_check)?)) // Ok(Disk::Remote(RemoteDisk::new(ep, opt.health_check)?))
} }
@@ -168,22 +167,22 @@ impl LocalDisk {
self.resolve_abs_path(dir) self.resolve_abs_path(dir)
} }
/// Write to the filesystem atomically. // /// Write to the filesystem atomically.
/// This is done by first writing to a temporary location and then moving the file. // /// This is done by first writing to a temporary location and then moving the file.
pub(crate) async fn prepare_file_write<'a>(&self, path: &'a PathBuf) -> Result<FileWriter<'a>> { // pub(crate) async fn prepare_file_write<'a>(&self, path: &'a PathBuf) -> Result<FileWriter<'a>> {
let tmp_path = self.get_object_path(RUSTFS_META_TMP_BUCKET, Uuid::new_v4().to_string().as_str())?; // let tmp_path = self.get_object_path(RUSTFS_META_TMP_BUCKET, Uuid::new_v4().to_string().as_str())?;
debug!("prepare_file_write tmp_path:{:?}, path:{:?}", &tmp_path, &path); // debug!("prepare_file_write tmp_path:{:?}, path:{:?}", &tmp_path, &path);
let file = File::create(&tmp_path).await?; // let file = File::create(&tmp_path).await?;
let writer = BufWriter::new(file); // let writer = BufWriter::new(file);
Ok(FileWriter { // Ok(FileWriter {
tmp_path, // tmp_path,
dest_path: path, // dest_path: path,
writer, // writer,
clean_tmp: true, // clean_tmp: true,
}) // })
} // }
pub async fn rename_all(&self, src_data_path: &PathBuf, dst_data_path: &PathBuf, skip: &PathBuf) -> Result<()> { pub async fn rename_all(&self, src_data_path: &PathBuf, dst_data_path: &PathBuf, skip: &PathBuf) -> Result<()> {
if !skip.starts_with(&src_data_path) { if !skip.starts_with(&src_data_path) {
@@ -247,8 +246,8 @@ impl LocalDisk {
async fn read_all_data( async fn read_all_data(
&self, &self,
bucket: &str, _bucket: &str,
volume_dir: impl AsRef<Path>, _volume_dir: impl AsRef<Path>,
path: impl AsRef<Path>, path: impl AsRef<Path>,
) -> Result<(Vec<u8>, OffsetDateTime)> { ) -> Result<(Vec<u8>, OffsetDateTime)> {
let (data, meta) = read_file_all(path).await?; let (data, meta) = read_file_all(path).await?;
@@ -404,7 +403,14 @@ impl DiskAPI for LocalDisk {
Ok(()) Ok(())
} }
async fn create_file(&self, origvolume: &str, volume: &str, path: &str, fileSize: usize, mut r: DuplexStream) -> Result<()> { async fn create_file(
&self,
_origvolume: &str,
volume: &str,
path: &str,
_file_size: usize,
mut r: DuplexStream,
) -> Result<()> {
let fpath = self.get_object_path(volume, path)?; let fpath = self.get_object_path(volume, path)?;
debug!("CreateFile fpath: {:?}", fpath); debug!("CreateFile fpath: {:?}", fpath);
@@ -461,15 +467,15 @@ impl DiskAPI for LocalDisk {
let (dst_buf, _) = read_file_exists(&dst_file_path).await?; let (dst_buf, _) = read_file_exists(&dst_file_path).await?;
let mut skipParent = dst_volume_path; let mut skip_parent = dst_volume_path;
if !&dst_buf.is_empty() { if !&dst_buf.is_empty() {
skipParent = PathBuf::from(&dst_file_path.parent().unwrap_or(Path::new("/"))); skip_parent = PathBuf::from(&dst_file_path.parent().unwrap_or(Path::new("/")));
} }
if !dst_buf.is_empty() { if !dst_buf.is_empty() {
meta = match FileMeta::unmarshal(&dst_buf) { meta = match FileMeta::unmarshal(&dst_buf) {
Ok(m) => m, Ok(m) => m,
Err(e) => FileMeta::new(), Err(_) => FileMeta::new(),
} }
// xl.load // xl.load
// meta.from(dst_buf); // meta.from(dst_buf);
@@ -483,10 +489,10 @@ impl DiskAPI for LocalDisk {
let no_inline = src_data_path.has_root() && fi.data.is_none() && fi.size > 0; let no_inline = src_data_path.has_root() && fi.data.is_none() && fi.size > 0;
if no_inline { if no_inline {
self.rename_all(&src_data_path, &dst_data_path, &skipParent).await?; self.rename_all(&src_data_path, &dst_data_path, &skip_parent).await?;
} }
self.rename_all(&src_file_path, &dst_file_path, &skipParent).await?; self.rename_all(&src_file_path, &dst_file_path, &skip_parent).await?;
if src_volume != RUSTFS_META_MULTIPART_BUCKET { if src_volume != RUSTFS_META_MULTIPART_BUCKET {
fs::remove_dir(&src_file_path.parent().unwrap()).await?; fs::remove_dir(&src_file_path.parent().unwrap()).await?;
@@ -542,7 +548,7 @@ impl DiskAPI for LocalDisk {
}) })
} }
async fn write_metadata(&self, org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> { async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> {
let p = self.get_object_path(&volume, format!("{}/{}", path, STORAGE_FORMAT_FILE).as_str())?; let p = self.get_object_path(&volume, format!("{}/{}", path, STORAGE_FORMAT_FILE).as_str())?;
let mut meta = FileMeta::new(); let mut meta = FileMeta::new();
@@ -564,7 +570,7 @@ impl DiskAPI for LocalDisk {
async fn read_version( async fn read_version(
&self, &self,
org_volume: &str, _org_volume: &str,
volume: &str, volume: &str,
path: &str, path: &str,
version_id: Uuid, version_id: Uuid,
@@ -618,44 +624,44 @@ impl DiskAPI for LocalDisk {
// } // }
// } // }
pub(crate) struct FileWriter<'a> { // pub(crate) struct FileWriter<'a> {
tmp_path: PathBuf, // tmp_path: PathBuf,
dest_path: &'a Path, // dest_path: &'a Path,
writer: BufWriter<File>, // writer: BufWriter<File>,
clean_tmp: bool, // clean_tmp: bool,
} // }
impl<'a> FileWriter<'a> { // impl<'a> FileWriter<'a> {
pub(crate) fn tmp_path(&self) -> &Path { // pub(crate) fn tmp_path(&self) -> &Path {
&self.tmp_path // &self.tmp_path
} // }
pub(crate) fn dest_path(&self) -> &'a Path { // pub(crate) fn dest_path(&self) -> &'a Path {
self.dest_path // self.dest_path
} // }
pub(crate) fn writer(&mut self) -> &mut BufWriter<File> { // pub(crate) fn writer(&mut self) -> &mut BufWriter<File> {
&mut self.writer // &mut self.writer
} // }
pub(crate) async fn done(mut self) -> Result<()> { // pub(crate) async fn done(mut self) -> Result<()> {
if let Some(final_dir_path) = self.dest_path().parent() { // if let Some(final_dir_path) = self.dest_path().parent() {
fs::create_dir_all(&final_dir_path).await?; // fs::create_dir_all(&final_dir_path).await?;
} // }
fs::rename(&self.tmp_path, self.dest_path()).await?; // fs::rename(&self.tmp_path, self.dest_path()).await?;
self.clean_tmp = false; // self.clean_tmp = false;
Ok(()) // Ok(())
} // }
} // }
impl<'a> Drop for FileWriter<'a> { // impl<'a> Drop for FileWriter<'a> {
fn drop(&mut self) { // fn drop(&mut self) {
if self.clean_tmp { // if self.clean_tmp {
let _ = std::fs::remove_file(&self.tmp_path); // let _ = std::fs::remove_file(&self.tmp_path);
} // }
} // }
} // }
#[cfg(test)] #[cfg(test)]
mod test { mod test {
ecstore/src/disk_api.rs
+1 -1
View File
@@ -15,7 +15,7 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes>; async fn read_all(&self, volume: &str, path: &str) -> Result<Bytes>;
async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()>; async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()>;
async fn rename_file(&self, src_volume: &str, src_path: &str, dst_volume: &str, dst_path: &str) -> 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, fileSize: usize, r: DuplexStream) -> Result<()>; async fn create_file(&self, origvolume: &str, volume: &str, path: &str, file_size: usize, r: DuplexStream) -> Result<()>;
async fn rename_data( async fn rename_data(
&self, &self,
src_volume: &str, src_volume: &str,
ecstore/src/erasure.rs
+14 -14
View File
@@ -1,5 +1,5 @@
use anyhow::anyhow; use anyhow::anyhow;
use anyhow::{Error, Result}; use anyhow::Result;
use bytes::Bytes; use bytes::Bytes;
use futures::{Stream, StreamExt}; use futures::{Stream, StreamExt};
use reed_solomon_erasure::galois_8::ReedSolomon; use reed_solomon_erasure::galois_8::ReedSolomon;
@@ -11,16 +11,16 @@ use tracing::debug;
use crate::chunk_stream::ChunkedStream; use crate::chunk_stream::ChunkedStream;
pub struct Erasure { pub struct Erasure {
data_shards: usize, // data_shards: usize,
parity_shards: usize, // parity_shards: usize,
encoder: ReedSolomon, encoder: ReedSolomon,
} }
impl Erasure { impl Erasure {
pub fn new(data_shards: usize, parity_shards: usize) -> Self { pub fn new(data_shards: usize, parity_shards: usize) -> Self {
Erasure { Erasure {
data_shards, // data_shards,
parity_shards, // parity_shards,
encoder: ReedSolomon::new(data_shards, parity_shards).unwrap(), encoder: ReedSolomon::new(data_shards, parity_shards).unwrap(),
} }
} }
@@ -31,7 +31,7 @@ impl Erasure {
writers: &mut Vec<W>, writers: &mut Vec<W>,
block_size: usize, block_size: usize,
data_size: usize, data_size: usize,
write_quorum: usize, _write_quorum: usize,
) -> Result<usize> ) -> Result<usize>
where where
S: Stream<Item = Result<Bytes, StdError>> + Send + Sync + 'static, S: Stream<Item = Result<Bytes, StdError>> + Send + Sync + 'static,
@@ -116,15 +116,15 @@ impl Erasure {
} }
} }
fn shards_to_option_shards<T: Clone>(shards: &[Vec<T>]) -> Vec<Option<Vec<T>>> { // fn shards_to_option_shards<T: Clone>(shards: &[Vec<T>]) -> Vec<Option<Vec<T>>> {
let mut result = Vec::with_capacity(shards.len()); // let mut result = Vec::with_capacity(shards.len());
for v in shards.iter() { // for v in shards.iter() {
let inner: Vec<T> = v.clone(); // let inner: Vec<T> = v.clone();
result.push(Some(inner)); // result.push(Some(inner));
} // }
result // result
} // }
#[cfg(test)] #[cfg(test)]
mod test { mod test {
ecstore/src/format.rs
+1 -1
View File
@@ -3,7 +3,7 @@ use serde::{Deserialize, Serialize};
use serde_json::Error as JsonError; use serde_json::Error as JsonError;
use uuid::Uuid; use uuid::Uuid;
use crate::{disk, disk_api::DiskError}; use crate::disk_api::DiskError;
#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)] #[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)]
pub enum FormatMetaVersion { pub enum FormatMetaVersion {
ecstore/src/peer.rs
+10 -10
View File
@@ -2,7 +2,6 @@ use anyhow::{Error, Result};
use async_trait::async_trait; use async_trait::async_trait;
use futures::future::join_all; use futures::future::join_all;
use std::{fmt::Debug, sync::Arc}; use std::{fmt::Debug, sync::Arc};
use tracing::debug;
use crate::{ use crate::{
disk::DiskStore, disk::DiskStore,
@@ -141,15 +140,15 @@ impl PeerS3Client for S3PeerSys {
#[derive(Debug)] #[derive(Debug)]
pub struct LocalPeerS3Client { pub struct LocalPeerS3Client {
pub local_disks: Vec<DiskStore>, pub local_disks: Vec<DiskStore>,
pub node: Node, // pub node: Node,
pub pools: Vec<usize>, pub pools: Vec<usize>,
} }
impl LocalPeerS3Client { impl LocalPeerS3Client {
fn new(local_disks: Vec<DiskStore>, node: Node, pools: Vec<usize>) -> Self { fn new(local_disks: Vec<DiskStore>, _node: Node, pools: Vec<usize>) -> Self {
Self { Self {
local_disks, local_disks,
node, // node,
pools, pools,
} }
} }
@@ -192,7 +191,7 @@ impl PeerS3Client for LocalPeerS3Client {
Ok(()) Ok(())
} }
async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> { async fn get_bucket_info(&self, bucket: &str, _opts: &BucketOptions) -> Result<BucketInfo> {
let mut futures = Vec::with_capacity(self.local_disks.len()); let mut futures = Vec::with_capacity(self.local_disks.len());
for disk in self.local_disks.iter() { for disk in self.local_disks.iter() {
futures.push(disk.stat_volume(bucket)); futures.push(disk.stat_volume(bucket));
@@ -233,13 +232,14 @@ impl PeerS3Client for LocalPeerS3Client {
#[derive(Debug)] #[derive(Debug)]
pub struct RemotePeerS3Client { pub struct RemotePeerS3Client {
pub node: Node, // pub node: Node,
pub pools: Vec<usize>, // pub pools: Vec<usize>,
} }
impl RemotePeerS3Client { impl RemotePeerS3Client {
fn new(node: Node, pools: Vec<usize>) -> Self { fn new(_node: Node, _pools: Vec<usize>) -> Self {
Self { node, pools } // Self { node, pools }
Self {}
} }
} }
@@ -251,7 +251,7 @@ impl PeerS3Client for RemotePeerS3Client {
async fn make_bucket(&self, _bucket: &str, _opts: &MakeBucketOptions) -> Result<()> { async fn make_bucket(&self, _bucket: &str, _opts: &MakeBucketOptions) -> Result<()> {
unimplemented!() unimplemented!()
} }
async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> { async fn get_bucket_info(&self, _bucket: &str, _opts: &BucketOptions) -> Result<BucketInfo> {
unimplemented!() unimplemented!()
} }
} }
ecstore/src/sets.rs
+20 -30
View File
@@ -1,13 +1,11 @@
use std::sync::Arc;
use anyhow::{Error, Result}; use anyhow::{Error, Result};
use futures::{future::join_all, AsyncWrite, StreamExt}; use futures::future::join_all;
use time::OffsetDateTime; use time::OffsetDateTime;
use tracing::debug; use tracing::debug;
use uuid::Uuid; use uuid::Uuid;
use crate::{ use crate::{
disk::{self, DiskStore, RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_TMP_BUCKET}, disk::{DiskStore, RUSTFS_META_MULTIPART_BUCKET, RUSTFS_META_TMP_BUCKET},
endpoint::PoolEndpoints, endpoint::PoolEndpoints,
erasure::Erasure, erasure::Erasure,
format::{DistributionAlgoVersion, FormatV3}, format::{DistributionAlgoVersion, FormatV3},
@@ -21,8 +19,6 @@ use crate::{
}, },
}; };
const DEFAULT_INLINE_BLOCKS: usize = 128 * 1024;
#[derive(Debug)] #[derive(Debug)]
pub struct Sets { pub struct Sets {
pub id: Uuid, pub id: Uuid,
@@ -135,16 +131,16 @@ impl Sets {
errors errors
} }
async fn commit_rename_data_dir( // async fn commit_rename_data_dir(
&self, // &self,
disks: &Vec<Option<DiskStore>>, // disks: &Vec<Option<DiskStore>>,
bucket: &str, // bucket: &str,
object: &str, // object: &str,
data_dir: &str, // data_dir: &str,
// write_quorum: usize, // // write_quorum: usize,
) -> Vec<Option<Error>> { // ) -> Vec<Option<Error>> {
unimplemented!() // unimplemented!()
} // }
} }
async fn write_unique_file_info( async fn write_unique_file_info(
@@ -202,14 +198,6 @@ fn get_multipart_sha_dir(bucket: &str, object: &str) -> String {
hex(sha256(path.as_bytes()).as_ref()) hex(sha256(path.as_bytes()).as_ref())
} }
async fn check_upload_idexists() -> Result<()> {
unimplemented!()
}
async fn read_all_file_info() -> Result<()> {
unimplemented!()
}
// #[derive(Debug)] // #[derive(Debug)]
// pub struct Objects { // pub struct Objects {
// pub endpoints: Vec<Endpoint>, // pub endpoints: Vec<Endpoint>,
@@ -222,11 +210,11 @@ async fn read_all_file_info() -> Result<()> {
#[async_trait::async_trait] #[async_trait::async_trait]
impl StorageAPI for Sets { impl StorageAPI for Sets {
async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> { async fn make_bucket(&self, _bucket: &str, _opts: &MakeBucketOptions) -> Result<()> {
unimplemented!() unimplemented!()
} }
async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result<BucketInfo> { async fn get_bucket_info(&self, _bucket: &str, _opts: &BucketOptions) -> Result<BucketInfo> {
unimplemented!() unimplemented!()
} }
@@ -339,11 +327,11 @@ impl StorageAPI for Sets {
bucket: &str, bucket: &str,
object: &str, object: &str,
upload_id: &str, upload_id: &str,
part_id: usize, _part_id: usize,
data: PutObjReader, _data: PutObjReader,
opts: &ObjectOptions, _opts: &ObjectOptions,
) -> Result<PartInfo> { ) -> Result<PartInfo> {
let upload_path = get_upload_id_dir(bucket, object, upload_id); let _upload_path = get_upload_id_dir(bucket, object, upload_id);
// TODO: checkUploadIDExists // TODO: checkUploadIDExists
@@ -364,6 +352,8 @@ impl StorageAPI for Sets {
write_quorum += 1 write_quorum += 1
} }
let _ = write_quorum;
let mut fi = FileInfo::new([bucket, object].join("/").as_str(), data_drives, parity_drives); let mut fi = FileInfo::new([bucket, object].join("/").as_str(), data_drives, parity_drives);
fi.data_dir = Uuid::new_v4(); fi.data_dir = Uuid::new_v4();
ecstore/src/store.rs
-1
View File
@@ -3,7 +3,6 @@ use std::collections::HashMap;
use anyhow::{Error, Result}; use anyhow::{Error, Result};
use s3s::{dto::StreamingBlob, Body}; use s3s::{dto::StreamingBlob, Body};
use tracing::debug;
use uuid::Uuid; use uuid::Uuid;
use crate::{ use crate::{
ecstore/src/store_api.rs
+1 -6
View File
@@ -1,11 +1,6 @@
use std::{default, sync::Arc};
use anyhow::Result; use anyhow::Result;
use bytes::Bytes; use s3s::dto::StreamingBlob;
use futures::Stream;
use s3s::{dto::StreamingBlob, Body};
use time::OffsetDateTime; use time::OffsetDateTime;
use tracing::debug;
use uuid::Uuid; use uuid::Uuid;
pub const ERASURE_ALGORITHM: &str = "rs-vandermonde"; pub const ERASURE_ALGORITHM: &str = "rs-vandermonde";
ecstore/src/utils/string.rs
+115 -115
View File
@@ -1,133 +1,133 @@
use std::collections::HashMap; // use std::collections::HashMap;
use std::fmt; // use std::fmt;
#[derive(Debug, Clone)] // #[derive(Debug, Clone)]
pub struct StringSet(HashMap<String, ()>); // pub struct StringSet(HashMap<String, ()>);
impl StringSet { // impl StringSet {
// ToSlice - returns StringSet as a vector of strings. // // ToSlice - returns StringSet as a vector of strings.
pub fn to_slice(&self) -> Vec<String> { // pub fn to_slice(&self) -> Vec<String> {
let mut keys = self.0.keys().cloned().collect::<Vec<String>>(); // let mut keys = self.0.keys().cloned().collect::<Vec<String>>();
keys.sort(); // keys.sort();
keys // keys
} // }
// IsEmpty - returns whether the set is empty or not. // // IsEmpty - returns whether the set is empty or not.
pub fn is_empty(&self) -> bool { // pub fn is_empty(&self) -> bool {
self.0.len() == 0 // self.0.len() == 0
} // }
// Add - adds a string to the set. // // Add - adds a string to the set.
pub fn add(&mut self, s: String) { // pub fn add(&mut self, s: String) {
self.0.insert(s, ()); // self.0.insert(s, ());
} // }
// Remove - removes a string from the set. It does nothing if the string does not exist in the set. // // Remove - removes a string from the set. It does nothing if the string does not exist in the set.
pub fn remove(&mut self, s: &str) { // pub fn remove(&mut self, s: &str) {
self.0.remove(s); // self.0.remove(s);
} // }
// Contains - checks if a string is in the set. // // Contains - checks if a string is in the set.
pub fn contains(&self, s: &str) -> bool { // pub fn contains(&self, s: &str) -> bool {
self.0.contains_key(s) // self.0.contains_key(s)
} // }
// FuncMatch - returns a new set containing each value that passes the match function. // // FuncMatch - returns a new set containing each value that passes the match function.
pub fn func_match<F>(&self, match_fn: F, match_string: &str) -> StringSet // pub fn func_match<F>(&self, match_fn: F, match_string: &str) -> StringSet
where // where
F: Fn(&str, &str) -> bool, // F: Fn(&str, &str) -> bool,
{ // {
StringSet( // StringSet(
self.0 // self.0
.iter() // .iter()
.filter(|(k, _)| match_fn(k, match_string)) // .filter(|(k, _)| match_fn(k, match_string))
.map(|(k, _)| (k.clone(), ())) // .map(|(k, _)| (k.clone(), ()))
.collect::<HashMap<String, ()>>(), // .collect::<HashMap<String, ()>>(),
) // )
} // }
// ApplyFunc - returns a new set containing each value processed by 'apply_fn'. // // ApplyFunc - returns a new set containing each value processed by 'apply_fn'.
pub fn apply_func<F>(&self, apply_fn: F) -> StringSet // pub fn apply_func<F>(&self, apply_fn: F) -> StringSet
where // where
F: Fn(&str) -> String, // F: Fn(&str) -> String,
{ // {
StringSet( // StringSet(
self.0 // self.0
.iter() // .iter()
.map(|(k, _)| (apply_fn(k), ())) // .map(|(k, _)| (apply_fn(k), ()))
.collect::<HashMap<String, ()>>(), // .collect::<HashMap<String, ()>>(),
) // )
} // }
// Equals - checks whether the given set is equal to the current set or not. // // Equals - checks whether the given set is equal to the current set or not.
pub fn equals(&self, other: &StringSet) -> bool { // pub fn equals(&self, other: &StringSet) -> bool {
if self.0.len() != other.0.len() { // if self.0.len() != other.0.len() {
return false; // return false;
} // }
self.0.iter().all(|(k, _)| other.0.contains_key(k)) // self.0.iter().all(|(k, _)| other.0.contains_key(k))
} // }
// Intersection - returns the intersection with the given set as a new set. // // Intersection - returns the intersection with the given set as a new set.
pub fn intersection(&self, other: &StringSet) -> StringSet { // pub fn intersection(&self, other: &StringSet) -> StringSet {
StringSet( // StringSet(
self.0 // self.0
.iter() // .iter()
.filter(|(k, _)| other.0.contains_key::<String>(k)) // .filter(|(k, _)| other.0.contains_key::<String>(k))
.map(|(k, _)| (k.clone(), ())) // .map(|(k, _)| (k.clone(), ()))
.collect::<HashMap<String, ()>>(), // .collect::<HashMap<String, ()>>(),
) // )
} // }
// Difference - returns the difference with the given set as a new set. // // Difference - returns the difference with the given set as a new set.
pub fn difference(&self, other: &StringSet) -> StringSet { // pub fn difference(&self, other: &StringSet) -> StringSet {
StringSet( // StringSet(
self.0 // self.0
.iter() // .iter()
.filter(|(k, _)| !other.0.contains_key::<String>(k)) // .filter(|(k, _)| !other.0.contains_key::<String>(k))
.map(|(k, _)| (k.clone(), ())) // .map(|(k, _)| (k.clone(), ()))
.collect::<HashMap<String, ()>>(), // .collect::<HashMap<String, ()>>(),
) // )
} // }
// Union - returns the union with the given set as a new set. // // Union - returns the union with the given set as a new set.
pub fn union(&self, other: &StringSet) -> StringSet { // pub fn union(&self, other: &StringSet) -> StringSet {
let mut new_set = self.clone(); // let mut new_set = self.clone();
for (k, _) in other.0.iter() { // for (k, _) in other.0.iter() {
new_set.0.insert(k.clone(), ()); // new_set.0.insert(k.clone(), ());
} // }
new_set // new_set
} // }
} // }
// Implementing JSON serialization and deserialization would require the serde crate. // // Implementing JSON serialization and deserialization would require the serde crate.
// You would also need to implement Display and PartialEq traits for more idiomatic Rust. // // You would also need to implement Display and PartialEq traits for more idiomatic Rust.
// Implementing Display trait to provide a string representation of the set. // // Implementing Display trait to provide a string representation of the set.
impl fmt::Display for StringSet { // impl fmt::Display for StringSet {
fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result { // fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
write!(f, "{}", self.to_slice().join(", ")) // write!(f, "{}", self.to_slice().join(", "))
} // }
} // }
// Implementing PartialEq and Eq traits to allow comparison of StringSet instances. // // Implementing PartialEq and Eq traits to allow comparison of StringSet instances.
impl PartialEq for StringSet { // impl PartialEq for StringSet {
fn eq(&self, other: &StringSet) -> bool { // fn eq(&self, other: &StringSet) -> bool {
self.equals(other) // self.equals(other)
} // }
} // }
impl Eq for StringSet {} // impl Eq for StringSet {}
// NewStringSet - creates a new string set. // // NewStringSet - creates a new string set.
pub fn new_string_set() -> StringSet { // pub fn new_string_set() -> StringSet {
StringSet(HashMap::new()) // StringSet(HashMap::new())
} // }
// CreateStringSet - creates a new string set with given string values. // // CreateStringSet - creates a new string set with given string values.
pub fn create_string_set(sl: Vec<String>) -> StringSet { // pub fn create_string_set(sl: Vec<String>) -> StringSet {
let mut set = new_string_set(); // let mut set = new_string_set();
for k in sl { // for k in sl {
set.add(k); // set.add(k);
} // }
set // set
} // }
rustfs/src/storage/ecfs.rs
+17 -18
View File
@@ -17,7 +17,6 @@ use s3s::{S3Request, S3Response};
use anyhow::Result; use anyhow::Result;
use ecstore::store::ECStore; use ecstore::store::ECStore;
use tracing::error;
macro_rules! try_ { macro_rules! try_ {
($result:expr) => { ($result:expr) => {
@@ -54,14 +53,14 @@ impl S3 for FS {
.await .await
); );
let output = CreateBucketOutput::default(); // TODO: handle other fields let output = CreateBucketOutput::default();
Ok(S3Response::new(output)) Ok(S3Response::new(output))
} }
#[tracing::instrument] #[tracing::instrument]
async fn copy_object(&self, req: S3Request<CopyObjectInput>) -> S3Result<S3Response<CopyObjectOutput>> { async fn copy_object(&self, req: S3Request<CopyObjectInput>) -> S3Result<S3Response<CopyObjectOutput>> {
let input = req.input; let input = req.input;
let (bucket, key) = match input.copy_source { let (_bucket, _key) = match input.copy_source {
CopySource::AccessPoint { .. } => return Err(s3_error!(NotImplemented)), CopySource::AccessPoint { .. } => return Err(s3_error!(NotImplemented)),
CopySource::Bucket { ref bucket, ref key, .. } => (bucket, key), CopySource::Bucket { ref bucket, ref key, .. } => (bucket, key),
}; };
@@ -72,22 +71,22 @@ impl S3 for FS {
#[tracing::instrument] #[tracing::instrument]
async fn delete_bucket(&self, req: S3Request<DeleteBucketInput>) -> S3Result<S3Response<DeleteBucketOutput>> { async fn delete_bucket(&self, req: S3Request<DeleteBucketInput>) -> S3Result<S3Response<DeleteBucketOutput>> {
let input = req.input; let _input = req.input;
Ok(S3Response::new(DeleteBucketOutput {})) Ok(S3Response::new(DeleteBucketOutput {}))
} }
#[tracing::instrument] #[tracing::instrument]
async fn delete_object(&self, req: S3Request<DeleteObjectInput>) -> S3Result<S3Response<DeleteObjectOutput>> { async fn delete_object(&self, req: S3Request<DeleteObjectInput>) -> S3Result<S3Response<DeleteObjectOutput>> {
let input = req.input; let _input = req.input;
let output = DeleteObjectOutput::default(); // TODO: handle other fields let output = DeleteObjectOutput::default();
Ok(S3Response::new(output)) Ok(S3Response::new(output))
} }
#[tracing::instrument] #[tracing::instrument]
async fn delete_objects(&self, req: S3Request<DeleteObjectsInput>) -> S3Result<S3Response<DeleteObjectsOutput>> { async fn delete_objects(&self, req: S3Request<DeleteObjectsInput>) -> S3Result<S3Response<DeleteObjectsOutput>> {
let input = req.input; let _input = req.input;
let output = DeleteObjectsOutput { ..Default::default() }; let output = DeleteObjectsOutput { ..Default::default() };
Ok(S3Response::new(output)) Ok(S3Response::new(output))
@@ -146,7 +145,7 @@ impl S3 for FS {
#[tracing::instrument] #[tracing::instrument]
async fn head_object(&self, req: S3Request<HeadObjectInput>) -> S3Result<S3Response<HeadObjectOutput>> { async fn head_object(&self, req: S3Request<HeadObjectInput>) -> S3Result<S3Response<HeadObjectOutput>> {
let input = req.input; let _input = req.input;
let output = HeadObjectOutput { ..Default::default() }; let output = HeadObjectOutput { ..Default::default() };
Ok(S3Response::new(output)) Ok(S3Response::new(output))
@@ -175,7 +174,7 @@ impl S3 for FS {
#[tracing::instrument] #[tracing::instrument]
async fn list_objects_v2(&self, req: S3Request<ListObjectsV2Input>) -> S3Result<S3Response<ListObjectsV2Output>> { async fn list_objects_v2(&self, req: S3Request<ListObjectsV2Input>) -> S3Result<S3Response<ListObjectsV2Output>> {
let input = req.input; let _input = req.input;
let output = ListObjectsV2Output { ..Default::default() }; let output = ListObjectsV2Output { ..Default::default() };
Ok(S3Response::new(output)) Ok(S3Response::new(output))
@@ -196,7 +195,7 @@ impl S3 for FS {
body, body,
bucket, bucket,
key, key,
metadata, // metadata,
content_length, content_length,
.. ..
} = input; } = input;
@@ -244,14 +243,14 @@ impl S3 for FS {
async fn upload_part(&self, req: S3Request<UploadPartInput>) -> S3Result<S3Response<UploadPartOutput>> { async fn upload_part(&self, req: S3Request<UploadPartInput>) -> S3Result<S3Response<UploadPartOutput>> {
let UploadPartInput { let UploadPartInput {
body, body,
upload_id, // upload_id,
part_number, // part_number,
content_length, content_length,
.. ..
} = req.input; } = req.input;
let body = body.ok_or_else(|| s3_error!(IncompleteBody))?; let _body = body.ok_or_else(|| s3_error!(IncompleteBody))?;
let content_length = content_length.ok_or_else(|| s3_error!(IncompleteBody))?; let _content_length = content_length.ok_or_else(|| s3_error!(IncompleteBody))?;
// mc cp step 4 // mc cp step 4
@@ -261,7 +260,7 @@ impl S3 for FS {
#[tracing::instrument] #[tracing::instrument]
async fn upload_part_copy(&self, req: S3Request<UploadPartCopyInput>) -> S3Result<S3Response<UploadPartCopyOutput>> { async fn upload_part_copy(&self, req: S3Request<UploadPartCopyInput>) -> S3Result<S3Response<UploadPartCopyOutput>> {
let input = req.input; let _input = req.input;
let output = UploadPartCopyOutput { ..Default::default() }; let output = UploadPartCopyOutput { ..Default::default() };
@@ -289,10 +288,10 @@ impl S3 for FS {
req: S3Request<CompleteMultipartUploadInput>, req: S3Request<CompleteMultipartUploadInput>,
) -> S3Result<S3Response<CompleteMultipartUploadOutput>> { ) -> S3Result<S3Response<CompleteMultipartUploadOutput>> {
let CompleteMultipartUploadInput { let CompleteMultipartUploadInput {
multipart_upload, // multipart_upload,
bucket, bucket,
key, key,
upload_id, // upload_id,
.. ..
} = req.input; } = req.input;
@@ -307,7 +306,7 @@ impl S3 for FS {
#[tracing::instrument] #[tracing::instrument]
async fn abort_multipart_upload( async fn abort_multipart_upload(
&self, &self,
req: S3Request<AbortMultipartUploadInput>, _req: S3Request<AbortMultipartUploadInput>,
) -> S3Result<S3Response<AbortMultipartUploadOutput>> { ) -> S3Result<S3Response<AbortMultipartUploadOutput>> {
Ok(S3Response::new(AbortMultipartUploadOutput { ..Default::default() })) Ok(S3Response::new(AbortMultipartUploadOutput { ..Default::default() }))
} }